diff options
author | Maxim Devaev <[email protected]> | 2022-09-04 18:08:40 +0300 |
---|---|---|
committer | Maxim Devaev <[email protected]> | 2022-09-04 18:08:40 +0300 |
commit | ee3e224e396494cd0d69bb6167087a071a20349c (patch) | |
tree | 5becd28570e58a03c6e1e231d0db24c264a73f88 /kvmd/apps/ipmi | |
parent | 4b75221e9470b4a009955d7677f16adf8e23e302 (diff) |
new typing style
Diffstat (limited to 'kvmd/apps/ipmi')
-rw-r--r-- | kvmd/apps/ipmi/__init__.py | 5 | ||||
-rw-r--r-- | kvmd/apps/ipmi/auth.py | 7 | ||||
-rw-r--r-- | kvmd/apps/ipmi/server.py | 21 |
3 files changed, 12 insertions, 21 deletions
diff --git a/kvmd/apps/ipmi/__init__.py b/kvmd/apps/ipmi/__init__.py index fc450228..13aa9730 100644 --- a/kvmd/apps/ipmi/__init__.py +++ b/kvmd/apps/ipmi/__init__.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import List -from typing import Optional - from ...clients.kvmd import KvmdClient from ... import htclient @@ -34,7 +31,7 @@ from .server import IpmiServer # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: config = init( prog="kvmd-ipmi", description="IPMI to KVMD proxy", diff --git a/kvmd/apps/ipmi/auth.py b/kvmd/apps/ipmi/auth.py index 0ad3d208..2e495af8 100644 --- a/kvmd/apps/ipmi/auth.py +++ b/kvmd/apps/ipmi/auth.py @@ -22,9 +22,6 @@ import dataclasses -from typing import List -from typing import Dict - # ===== class IpmiPasswdError(Exception): @@ -55,8 +52,8 @@ class IpmiAuthManager: def get_credentials(self, ipmi_user: str) -> IpmiUserCredentials: return self.__credentials[ipmi_user] - def __parse_passwd_file(self, lines: List[str]) -> Dict[str, IpmiUserCredentials]: - credentials: Dict[str, IpmiUserCredentials] = {} + def __parse_passwd_file(self, lines: list[str]) -> dict[str, IpmiUserCredentials]: + credentials: dict[str, IpmiUserCredentials] = {} for (lineno, line) in enumerate(lines): if len(line.strip()) == 0 or line.lstrip().startswith("#"): continue diff --git a/kvmd/apps/ipmi/server.py b/kvmd/apps/ipmi/server.py index 597e6dce..87265607 100644 --- a/kvmd/apps/ipmi/server.py +++ b/kvmd/apps/ipmi/server.py @@ -28,9 +28,6 @@ import multiprocessing import functools import queue -from typing import Dict -from typing import Optional - import aiohttp import serial @@ -83,8 +80,8 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute self.__sol_proxy_port = (sol_proxy_port or port) self.__sol_lock = threading.Lock() - self.__sol_console: Optional[IpmiConsole] = None - self.__sol_thread: Optional[threading.Thread] = None + self.__sol_console: (IpmiConsole | None) = None + self.__sol_thread: (threading.Thread | None) = None self.__sol_stop = False def run(self) -> None: @@ -100,7 +97,7 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute # ===== - def handle_raw_request(self, request: Dict, session: IpmiServerSession) -> None: + def handle_raw_request(self, request: dict, session: IpmiServerSession) -> None: handler = { (6, 1): (lambda _, session: self.send_device_id(session)), # Get device ID (6, 7): self.__get_power_state_handler, # Power state @@ -125,13 +122,13 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute # ===== - def __get_power_state_handler(self, _: Dict, session: IpmiServerSession) -> None: + def __get_power_state_handler(self, _: dict, session: IpmiServerSession) -> None: # https://github.com/arcress0/ipmiutil/blob/e2f6e95127d22e555f959f136d9bb9543c763896/util/ireset.c#L654 result = self.__make_request(session, "atx.get_state() [power]", "atx.get_state") data = [(0 if result["leds"]["power"] else 5)] session.send_ipmi_response(data=data) - def __get_selftest_status_handler(self, _: Dict, session: IpmiServerSession) -> None: + def __get_selftest_status_handler(self, _: dict, session: IpmiServerSession) -> None: # https://github.com/arcress0/ipmiutil/blob/e2f6e95127d22e555f959f136d9bb9543c763896/util/ihealth.c#L858 data = [0x0055] try: @@ -140,12 +137,12 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute data = [0] session.send_ipmi_response(data=data) - def __get_chassis_status_handler(self, _: Dict, session: IpmiServerSession) -> None: + def __get_chassis_status_handler(self, _: dict, session: IpmiServerSession) -> None: result = self.__make_request(session, "atx.get_state() [chassis]", "atx.get_state") data = [int(result["leds"]["power"]), 0, 0] session.send_ipmi_response(data=data) - def __chassis_control_handler(self, request: Dict, session: IpmiServerSession) -> None: + def __chassis_control_handler(self, request: dict, session: IpmiServerSession) -> None: action = { 0: "off_hard", 1: "on", @@ -179,7 +176,7 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute # ===== - def __activate_sol_handler(self, _: Dict, session: IpmiServerSession) -> None: + def __activate_sol_handler(self, _: dict, session: IpmiServerSession) -> None: with self.__sol_lock: if not self.__sol_device_path: session.send_ipmi_response(code=0x81) # SOL disabled @@ -198,7 +195,7 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute ]) self.__start_sol_worker(session) - def __deactivate_sol_handler(self, _: Dict, session: IpmiServerSession) -> None: + def __deactivate_sol_handler(self, _: dict, session: IpmiServerSession) -> None: with self.__sol_lock: if not self.__sol_device_path: session.send_ipmi_response(code=0x81) |