diff --git a/kvmd/apps/kvmd/api/switch.py b/kvmd/apps/kvmd/api/switch.py index 105e26f7..f6019611 100644 --- a/kvmd/apps/kvmd/api/switch.py +++ b/kvmd/apps/kvmd/api/switch.py @@ -28,6 +28,7 @@ from ....htserver import make_json_response from ....validators.basic import valid_bool from ....validators.basic import valid_int_f0 +from ....validators.basic import valid_float_f0 from ....validators.basic import valid_stripped_string_not_empty from ....validators.kvm import valid_atx_power_action from ....validators.kvm import valid_atx_button @@ -64,7 +65,7 @@ class SwitchApi: @exposed_http("POST", "/switch/set_active") async def __set_active_port_handler(self, req: Request) -> Response: - port = valid_int_f0(req.query.get("port")) + port = valid_float_f0(req.query.get("port")) await self.__switch.set_active_port(port) return make_json_response() @@ -72,7 +73,7 @@ class SwitchApi: async def __set_beacon_handler(self, req: Request) -> Response: on = valid_bool(req.query.get("state")) if "port" in req.query: - port = valid_int_f0(req.query.get("port")) + port = valid_float_f0(req.query.get("port")) await self.__switch.set_port_beacon(port, on) elif "uplink" in req.query: unit = valid_int_f0(req.query.get("uplink")) @@ -84,7 +85,7 @@ class SwitchApi: @exposed_http("POST", "/switch/set_port_params") async def __set_port_params(self, req: Request) -> Response: - port = valid_int_f0(req.query.get("port")) + port = valid_float_f0(req.query.get("port")) params = { param: validator(req.query.get(param)) for (param, validator) in [ @@ -153,7 +154,7 @@ class SwitchApi: @exposed_http("POST", "/switch/atx/power") async def __power_handler(self, req: Request) -> Response: - port = valid_int_f0(req.query.get("port")) + port = valid_float_f0(req.query.get("port")) action = valid_atx_power_action(req.query.get("action")) await ({ "on": self.__switch.atx_power_on, @@ -165,7 +166,7 @@ class SwitchApi: @exposed_http("POST", "/switch/atx/click") async def __click_handler(self, req: Request) -> Response: - port = valid_int_f0(req.query.get("port")) + port = valid_float_f0(req.query.get("port")) button = valid_atx_button(req.query.get("button")) await ({ "power": self.__switch.atx_click_power, diff --git a/kvmd/apps/kvmd/switch/__init__.py b/kvmd/apps/kvmd/switch/__init__.py index 33f96e4c..85eca48a 100644 --- a/kvmd/apps/kvmd/switch/__init__.py +++ b/kvmd/apps/kvmd/switch/__init__.py @@ -147,13 +147,13 @@ class Switch: # pylint: disable=too-many-public-methods async def set_active_next(self) -> None: self.__chain.set_active_next() - async def set_active_port(self, port: int) -> None: - self.__chain.set_active_port(port) + async def set_active_port(self, port: float) -> None: + self.__chain.set_active_port(self.__chain.translate_port(port)) # ===== - async def set_port_beacon(self, port: int, on: bool) -> None: - self.__chain.set_port_beacon(port, on) + async def set_port_beacon(self, port: float, on: bool) -> None: + self.__chain.set_port_beacon(self.__chain.translate_port(port), on) async def set_uplink_beacon(self, unit: int, on: bool) -> None: self.__chain.set_uplink_beacon(unit, on) @@ -163,33 +163,35 @@ class Switch: # pylint: disable=too-many-public-methods # ===== - async def atx_power_on(self, port: int) -> None: + async def atx_power_on(self, port: float) -> None: self.__inner_atx_cp(port, False, self.__X_ATX_CP_DELAYS) - async def atx_power_off(self, port: int) -> None: + async def atx_power_off(self, port: float) -> None: self.__inner_atx_cp(port, True, self.__X_ATX_CP_DELAYS) - async def atx_power_off_hard(self, port: int) -> None: + async def atx_power_off_hard(self, port: float) -> None: self.__inner_atx_cp(port, True, self.__X_ATX_CPL_DELAYS) - async def atx_power_reset_hard(self, port: int) -> None: + async def atx_power_reset_hard(self, port: float) -> None: self.__inner_atx_cr(port, True) - async def atx_click_power(self, port: int) -> None: + async def atx_click_power(self, port: float) -> None: self.__inner_atx_cp(port, None, self.__X_ATX_CP_DELAYS) - async def atx_click_power_long(self, port: int) -> None: + async def atx_click_power_long(self, port: float) -> None: self.__inner_atx_cp(port, None, self.__X_ATX_CPL_DELAYS) - async def atx_click_reset(self, port: int) -> None: + async def atx_click_reset(self, port: float) -> None: self.__inner_atx_cr(port, None) - def __inner_atx_cp(self, port: int, if_powered: (bool | None), x_delay: str) -> None: + def __inner_atx_cp(self, port: float, if_powered: (bool | None), x_delay: str) -> None: assert x_delay in [self.__X_ATX_CP_DELAYS, self.__X_ATX_CPL_DELAYS] + port = self.__chain.translate_port(port) delay = getattr(self.__cache, f"get_{x_delay}")()[port] self.__chain.click_power(port, delay, if_powered) - def __inner_atx_cr(self, port: int, if_powered: (bool | None)) -> None: + def __inner_atx_cr(self, port: float, if_powered: (bool | None)) -> None: + port = self.__chain.translate_port(port) delay = self.__cache.get_atx_cr_delays()[port] self.__chain.click_reset(port, delay, if_powered) @@ -257,6 +259,7 @@ class Switch: # pylint: disable=too-many-public-methods atx_click_reset_delay: (float | None)=None, ) -> None: + port = self.__chain.translate_port(port) async with self.__lock: if edid_id is not None: edids = self.__cache.get_edids() diff --git a/kvmd/apps/kvmd/switch/chain.py b/kvmd/apps/kvmd/switch/chain.py index 386fb7bf..122d6203 100644 --- a/kvmd/apps/kvmd/switch/chain.py +++ b/kvmd/apps/kvmd/switch/chain.py @@ -220,6 +220,18 @@ class Chain: # pylint: disable=too-many-instance-attributes # ===== + def translate_port(self, port: float) -> int: + assert port >= 0 + if int(port) == port: + return int(port) + (unit, ch) = map(int, str(port).split(".")) + unit = min(max(unit, 1), 5) + ch = min(max(ch, 1), 4) + port = min((unit - 1) * 4 + (ch - 1), 19) + return port + + # ===== + def set_active_prev(self) -> None: self.__queue_cmd(_CmdSetActivePrev()) diff --git a/kvmd/clients/kvmd.py b/kvmd/clients/kvmd.py index c6a286f0..84850c28 100644 --- a/kvmd/clients/kvmd.py +++ b/kvmd/clients/kvmd.py @@ -147,7 +147,7 @@ class _SwitchApiPart(_BaseApiPart): async with session.post("/switch/set_active_next") as resp: htclient.raise_not_200(resp) - async def set_active(self, port: int) -> None: + async def set_active(self, port: float) -> None: session = self._ensure_http_session() async with session.post( url="/switch/set_active",