switch: id/port api

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,

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:
+    async def set_active_port(self, port: float) -> None:
-        self.__chain.set_active_port(port)
+        self.__chain.set_active_port(self.__chain.translate_port(port))
 
     # =====
 
-    async def set_port_beacon(self, port: int, on: bool) -> None:
+    async def set_port_beacon(self, port: float, on: bool) -> None:
-        self.__chain.set_port_beacon(port, on)
+        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()

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())
 

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",