GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2025-12-12 17:20:30 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

lint fixes

Browse Source
This commit is contained in:
Maxim Devaev 2022-07-19 17:02:26 +03:00
parent adf4be9bf7
commit c58cc99582
3 changed files with 20 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
kvmd/apps/vnc/rfb/encodings.py
View File

@ -23,6 +23,7 @@
import dataclasses
from typing import List
from typing import Dict
from typing import FrozenSet
from typing import Union
from typing import Any
@ -44,25 +45,23 @@ class RfbEncodings:
H264 = 50 # Open H.264 Encoding
def _feature(default: Any, variants: Union[int, FrozenSet[int]]) -> dataclasses.Field:
return dataclasses.field(default=default, metadata={
"variants": (frozenset([variants]) if isinstance(variants, int) else variants),
})
def _make_meta(variants: Union[int, FrozenSet[int]]) -> Dict:
return {"variants": (frozenset([variants]) if isinstance(variants, int) else variants)}
@dataclasses.dataclass(frozen=True)
class RfbClientEncodings: # pylint: disable=too-many-instance-attributes
encodings: FrozenSet[int]
has_resize: bool = _feature(False, RfbEncodings.RESIZE)
has_rename: bool = _feature(False, RfbEncodings.RENAME)
has_leds_state: bool = _feature(False, RfbEncodings.LEDS_STATE)
has_ext_keys: bool = _feature(False, RfbEncodings.EXT_KEYS)
has_resize: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.RESIZE)) # noqa: E224
has_rename: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.RENAME)) # noqa: E224
has_leds_state: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.LEDS_STATE)) # noqa: E224
has_ext_keys: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.EXT_KEYS)) # noqa: E224
has_tight: bool = _feature(False, RfbEncodings.TIGHT)
tight_jpeg_quality: int = _feature(0, frozenset(RfbEncodings.TIGHT_JPEG_QUALITIES))
has_tight: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.TIGHT)) # noqa: E224
tight_jpeg_quality: int = dataclasses.field(default=0, metadata=_make_meta(frozenset(RfbEncodings.TIGHT_JPEG_QUALITIES))) # noqa: E224
has_h264: bool = _feature(False, RfbEncodings.H264)
has_h264: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.H264)) # noqa: E224
def get_summary(self) -> List[str]:
summary: List[str] = [f"encodings -- {sorted(self.encodings)}"]
@ -81,7 +80,7 @@ class RfbClientEncodings: # pylint: disable=too-many-instance-attributes
def __set_value(self, key: str, value: Any) -> None:
object.__setattr__(self, key, value)
def __get_found(self, field: dataclasses.Field) -> None:
def __get_found(self, field: dataclasses.Field) -> FrozenSet[int]:
return self.encodings.intersection(field.metadata["variants"])
def __get_tight_jpeg_quality(self) -> int:

2
kvmd/clients/kvmd.py
View File

@ -135,7 +135,7 @@ class KvmdClientWs:
self.__writer_queue: "asyncio.Queue[Tuple[str, Dict]]" = asyncio.Queue()
self.__communicated = False
async def communicate(self) -> AsyncGenerator[Tuple[str, Dict], None]:
async def communicate(self) -> AsyncGenerator[Tuple[str, Dict], None]: # pylint: disable=too-many-branches
assert not self.__communicated
self.__communicated = True
receive_task: Optional[asyncio.Task] = None

9
kvmd/htserver.py
View File

@ -34,8 +34,10 @@ from typing import Dict
from typing import Callable
from typing import AsyncGenerator
from typing import Optional
from typing import Union
from typing import Any
from aiohttp import ClientWebSocketResponse
from aiohttp.web import BaseRequest
from aiohttp.web import Request
from aiohttp.web import Response
@ -208,7 +210,12 @@ async def stream_json_exception(response: StreamResponse, err: Exception) -> Non
}, False)
async def send_ws_event(wsr: WebSocketResponse, event_type: str, event: Optional[Dict]) -> None:
async def send_ws_event(
wsr: Union[ClientWebSocketResponse, WebSocketResponse],
event_type: str,
event: Optional[Dict],
) -> None:
await wsr.send_str(json.dumps({
"event_type": event_type,
"event": event,