Improve ports

axis/vapix/interfaces/param_cgi.py

@@ -13,11 +13,11 @@
     BrandT,
     ImageParam,
     Param,
-    PortParam,
     PropertyParam,
     PTZParam,
     StreamProfileParam,
 )
+from ..models.port_cgi import GetPortsRequest, ListInputRequest, ListOutputRequest
 from ..models.stream_profile import StreamProfile
 from .api import APIItems
 
@@ -239,17 +239,13 @@ def image_sources(self) -> dict:
 
     async def update_ports(self) -> None:
         """Update port groups of parameters."""
-        await asyncio.gather(
-            self.update(INPUT),
-            self.update(IOPORT),
-            self.update(OUTPUT),
+        bytes_list = await asyncio.gather(
+            self.vapix.new_request(ListInputRequest()),
+            self.vapix.new_request(GetPortsRequest()),
+            self.vapix.new_request(ListOutputRequest()),
         )
-
-    @property
-    def port_params(self) -> PortParam:
-        """Provide port parameters."""
-        data = self[INPUT].raw | self[OUTPUT].raw | self[IOPORT].raw
-        return PortParam.decode(data)
+        for bytes_data in bytes_list:
+            self.process_raw(bytes_data.decode())
 
     @property
     def nbrofinput(self) -> int:
@@ -262,33 +258,15 @@ def nbrofoutput(self) -> int:
         return int(self[OUTPUT]["NbrOfOutputs"])  # type: ignore
 
     @property
-    def ports(self) -> dict:
+    def ports(self) -> bytes:
         """Create a smaller dictionary containing all ports."""
         if IOPORT not in self:
-            return {}
-
-        attributes = (
-            "Usage",
-            "Configurable",
-            "Direction",
-            "Input.Name",
-            "Input.Trig",
-            "Output.Active",
-            "Output.Button",
-            "Output.DelayTime",
-            "Output.Mode",
-            "Output.Name",
-            "Output.PulseTime",
-        )
-
-        ports = self.process_dynamic_group(
-            self[IOPORT],  # type: ignore[arg-type]
-            "I",
-            attributes,
-            range(self.nbrofinput + self.nbrofoutput),
-        )
+            return b""
 
-        return ports
+        io_port = ""
+        for k, v in self[IOPORT].raw.items():
+            io_port += f"root.IOPort.{k}={v}\n"
+        return io_port.encode()
 
     # Properties
 

axis/vapix/interfaces/port_cgi.py

@@ -8,36 +8,46 @@
 Virtual input API.
 """
 
-from ..models.port_cgi import ACTION_HIGH  # noqa: F401
-from ..models.port_cgi import ACTION_LOW  # noqa: F401
-from ..models.port_cgi import DIRECTION_IN  # noqa: F401
-from ..models.port_cgi import DIRECTION_OUT  # noqa: F401
-from ..models.port_cgi import Port
-from ..models.port_cgi import URL  # noqa: F401
-from .api import APIItems
+from ..models.port_cgi import (
+    GetPortsRequest,
+    GetPortsResponse,
+    Port,
+    PortAction,
+    PortActionRequest,
+    PortDirection,
+)
+from .api_handler import ApiHandler
 
 PROPERTY = "Properties.API.HTTP.Version=3"
 
 
-class Ports(APIItems):
+class Ports(ApiHandler[Port]):
     """Represents all ports of io/port.cgi."""
 
-    item_cls = Port
-    path = ""
-
-    def __init__(self, vapix) -> None:
-        """Initialize port cgi manager."""
-        super().__init__(vapix, vapix.params.ports)
-
-    async def update(self) -> None:
-        """Refresh data."""
-        await self.vapix.params.update_ports()
-        self.process_raw(self.vapix.params.ports)
-
-    @staticmethod
-    def pre_process_raw(ports: dict) -> dict:
-        """Pre process ports for process raw.
-
-        Index needs to be a string.
-        """
-        return {str(k): v for k, v in ports.items()}
+    async def _api_request(self) -> dict[str, Port]:
+        """Get API data method defined by subclass."""
+        return await self.get_ports()
+
+    async def get_ports(self) -> dict[str, Port]:
+        """Retrieve privilege rights for current user."""
+        bytes_data = await self.vapix.new_request(GetPortsRequest())
+        return GetPortsResponse.decode(bytes_data).data
+
+    def process_ports(self) -> dict[str, Port]:
+        """Process ports."""
+        assert self.vapix.params is not None
+        return GetPortsResponse.decode(self.vapix.params.ports).data
+
+    async def action(self, id: str, action: PortAction) -> None:
+        """Activate or deactivate an output."""
+        if (port := self[id]) and port.direction != PortDirection.OUT:
+            return
+        await self.vapix.new_request(PortActionRequest(id, action.value))
+
+    async def open(self, id: str) -> None:
+        """Open port."""
+        await self.action(id, PortAction.LOW)
+
+    async def close(self, id: str) -> None:
+        """Close port."""
+        await self.action(id, PortAction.HIGH)

axis/vapix/models/api.py

@@ -61,6 +61,11 @@ def data(self) -> dict[str, str] | None:
         """Request data."""
         return None
 
+    @property
+    def params(self) -> dict[str, str] | None:
+        """Request query parameters."""
+        return None
+
 
 class APIItem:
     """Base class for all end points using APIItems class."""

axis/vapix/models/param_cgi.py

@@ -255,40 +255,6 @@ def decode(cls, data: dict[str, str]) -> Self:
         )
 
 
-@dataclass
-class PortParam(ApiItem):
-    """Port parameters."""
-
-    nbr_of_input: int
-    nbr_of_output: int
-    ports: dict
-
-    @classmethod
-    def decode(cls, data: dict[str, str]) -> Self:
-        """Decode dictionary to class object."""
-        inputs = int(data["NbrOfInputs"])
-        outputs = int(data["NbrOfOutputs"])
-        attributes = (
-            "Usage",
-            "Configurable",
-            "Direction",
-            "Input.Name",
-            "Input.Trig",
-            "Output.Active",
-            "Output.Button",
-            "Output.DelayTime",
-            "Output.Mode",
-            "Output.Name",
-            "Output.PulseTime",
-        )
-        return cls(
-            id="port",
-            nbr_of_input=inputs,
-            nbr_of_output=outputs,
-            ports=process_dynamic_group(data, "I", attributes, range(inputs + outputs)),
-        )
-
-
 @dataclass
 class PropertyParam(ApiItem):
     """Property parameters."""