feat: add AiPort support

src/uiprotect/api.py

@@ -57,7 +57,7 @@
     create_from_unifi_dict,
 )
 from .data.base import ProtectModelWithId
-from .data.devices import Chime
+from .data.devices import AiPort, Chime
 from .data.types import IteratorCallback, ProgressCallback
 from .exceptions import BadRequest, NotAuthorized, NvrError
 from .utils import (
@@ -1268,6 +1268,14 @@
         """
         return cast(list[Chime], await self.get_devices(ModelType.CHIME, Chime))
 
+    async def get_aiports(self) -> list[AiPort]:
+        """
+        Gets the list of aiports straight from the NVR.
+
+        The websocket is connected and running, you likely just want to use `self.bootstrap.aiports`
+        """
+        return cast(list[AiPort], await self.get_devices(ModelType.AIPORT, AiPort))
+
     async def get_viewers(self) -> list[Viewer]:
         """
         Gets the list of viewers straight from the NVR.
@@ -1386,6 +1394,14 @@
         """
         return cast(Chime, await self.get_device(ModelType.CHIME, device_id, Chime))
 
+    async def get_aiport(self, device_id: str) -> AiPort:
+        """
+        Gets a AiPort straight from the NVR.
+
+        The websocket is connected and running, you likely just want to use `self.bootstrap.aiport[device_id]`
+        """
+        return cast(AiPort, await self.get_device(ModelType.AIPORT, device_id, AiPort))
+
     async def get_viewer(self, device_id: str) -> Viewer:
         """
         Gets a viewer straight from the NVR.

src/uiprotect/cli/__init__.py

@@ -17,6 +17,7 @@
 from ..test_util import SampleDataGenerator
 from ..utils import RELEASE_CACHE, get_local_timezone, run_async
 from ..utils import profile_ws as profile_ws_job
+from .aiports import app as aiports_app
 from .base import CliContext, OutputFormatEnum
 from .cameras import app as camera_app
 from .chimes import app as chime_app
@@ -128,6 +129,7 @@
 app.add_typer(light_app, name="lights")
 app.add_typer(sensor_app, name="sensors")
 app.add_typer(viewer_app, name="viewers")
+app.add_typer(aiports_app, name="aiports")
 
 if backup_app is not None:
     app.add_typer(backup_app, name="backup")

src/uiprotect/cli/aiports.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+import typer
+
+from ..api import ProtectApiClient
+from ..cli import base
+from ..data import AiPort
+
+app = typer.Typer(rich_markup_mode="rich")
+
+ARG_DEVICE_ID = typer.Argument(None, help="ID of chime to select for subcommands")
+
+
+@dataclass
+class AiPortContext(base.CliContext):
+    devices: dict[str, AiPort]
+    device: AiPort | None = None
+
+
+ALL_COMMANDS, DEVICE_COMMANDS = base.init_common_commands(app)
+
+
+@app.callback(invoke_without_command=True)
+def main(ctx: typer.Context, device_id: Optional[str] = ARG_DEVICE_ID) -> None:
+    """
+    AiPort device CLI.
+
+    Returns full list of AiPorts without any arguments passed.
+    """
+    protect: ProtectApiClient = ctx.obj.protect
+    context = AiPortContext(
+        protect=ctx.obj.protect,
+        device=None,
+        devices=protect.bootstrap.aiports,
+        output_format=ctx.obj.output_format,
+    )
+    ctx.obj = context
+
+    if device_id is not None and device_id not in ALL_COMMANDS:
+        if (device := protect.bootstrap.aiports.get(device_id)) is None:
+            typer.secho("Invalid aiport ID", fg="red")
+            raise typer.Exit(1)
+        ctx.obj.device = device
+
+    if not ctx.invoked_subcommand:
+        if device_id in ALL_COMMANDS:
+            ctx.invoke(ALL_COMMANDS[device_id], ctx)
+            return
+
+        if ctx.obj.device is not None:
+            base.print_unifi_obj(ctx.obj.device, ctx.obj.output_format)
+            return
+
+        base.print_unifi_dict(ctx.obj.devices)

src/uiprotect/data/__init__.py

@@ -10,6 +10,7 @@
 from .bootstrap import Bootstrap
 from .convert import create_from_unifi_dict
 from .devices import (
+    AiPort,
     Bridge,
     Camera,
     CameraChannel,
@@ -85,6 +86,7 @@
     "DEFAULT_TYPE",
     "NVR",
     "WS_HEADER_SIZE",
+    "AiPort",
     "AnalyticsOption",
     "AudioStyle",
     "Bootstrap",

src/uiprotect/data/bootstrap.py

@@ -23,6 +23,7 @@
 )
 from .convert import MODEL_TO_CLASS, create_from_unifi_dict
 from .devices import (
+    AiPort,
     Bridge,
     Camera,
     Chime,
@@ -181,6 +182,7 @@ class Bootstrap(ProtectBaseObject):
     sensors: dict[str, Sensor]
     doorlocks: dict[str, Doorlock]
     chimes: dict[str, Chime]
+    aiports: dict[str, AiPort]
     last_update_id: str
 
     # TODO:

src/uiprotect/data/convert.py

@@ -8,6 +8,7 @@
 
 from ..exceptions import DataDecodeError
 from .devices import (
+    AiPort,
     Bridge,
     Camera,
     Chime,
@@ -40,6 +41,7 @@
     ModelType.SENSOR: Sensor,
     ModelType.DOORLOCK: Doorlock,
     ModelType.CHIME: Chime,
+    ModelType.AIPORT: AiPort,
     ModelType.KEYRING: Keyring,
     ModelType.ULP_USER: UlpUser,
 }

src/uiprotect/data/devices.py

@@ -976,6 +976,8 @@ class Camera(ProtectMotionDeviceModel):
     audio_settings: CameraAudioSettings | None = None
     # requires 5.0.33+
     is_third_party_camera: bool | None = None
+    # requires 5.1.78+
+    is_paired_with_ai_port: bool | None = None
     # TODO: used for adopting
     # apMac read only
     # apRssi read only
@@ -3382,3 +3384,7 @@ def callback() -> None:
                 raise BadRequest("Camera %s is not paired with chime", camera.id)
 
         await self.queue_update(callback)
+
+
+class AiPort(Camera):
+    paired_cameras: list[str]

src/uiprotect/data/types.py

@@ -124,6 +124,7 @@ class ModelType(str, UnknownValuesEnumMixin, enum.Enum):
     DOORLOCK = "doorlock"
     SCHEDULE = "schedule"
     CHIME = "chime"
+    AIPORT = "aiport"
     DEVICE_GROUP = "deviceGroup"
     RECORDING_SCHEDULE = "recordingSchedule"
     ULP_USER = "ulpUser"
@@ -173,6 +174,7 @@ def _bootstrap_model_types(cls) -> tuple[ModelType, ...]:
             ModelType.SENSOR,
             ModelType.DOORLOCK,
             ModelType.CHIME,
+            ModelType.AIPORT,
         )
 
     @classmethod

src/uiprotect/test_util/__init__.py

@@ -131,6 +131,7 @@ async def async_generate(self, close_session: bool = True) -> None:
             "sensor": len(bootstrap["sensors"]),
             "doorlock": len(bootstrap["doorlocks"]),
             "chime": len(bootstrap["chimes"]),
+            "aiport": len(bootstrap["aiports"]),
         }
 
         self.log("Generating event data...")
@@ -283,6 +284,7 @@ async def generate_device_data(
             self.generate_sensor_data(),
             self.generate_lock_data(),
             self.generate_chime_data(),
+            self.generate_aiport_data(),
             self.generate_bridge_data(),
             self.generate_liveview_data(),
         )
@@ -469,6 +471,21 @@ async def generate_chime_data(self) -> None:
         obj = await self.client.api_request_obj(f"chimes/{device_id}")
         await self.write_json_file("sample_chime", obj)
 
+    async def generate_aiport_data(self) -> None:
+        objs = await self.client.api_request_list("aiports")
+        device_id: str | None = None
+        for obj_dict in objs:
+            device_id = obj_dict["id"]
+            if is_online(obj_dict):
+                break
+
+        if device_id is None:
+            self.log("No aiport found. Skipping aiport endpoints...")
+            return
+
+        obj = await self.client.api_request_obj(f"aiports/{device_id}")
+        await self.write_json_file("sample_aiport", obj)
+
     async def generate_bridge_data(self) -> None:
         objs = await self.client.api_request_list("bridges")
         device_id: str | None = None

tests/conftest.py

@@ -53,6 +53,7 @@
 TEST_LIVEVIEW_EXISTS = (SAMPLE_DATA_DIRECTORY / "sample_liveview.json").exists()
 TEST_DOORLOCK_EXISTS = (SAMPLE_DATA_DIRECTORY / "sample_doorlock.json").exists()
 TEST_CHIME_EXISTS = (SAMPLE_DATA_DIRECTORY / "sample_chime.json").exists()
+TEST_AIPORT_EXISTS = (SAMPLE_DATA_DIRECTORY / "sample_aiport.json").exists()
 
 ANY_NONE = [[None], None, []]
 
@@ -89,6 +90,15 @@ def read_camera_json_file():
     return camera
 
 
+def read_aiport_json_file():
+    # tests expect global recording settings to be off
+    aiport = read_json_file("sample_aiport")
+    if aiport.get("useGlobal"):
+        aiport["useGlobal"] = False
+
+    return aiport
+
+
 def get_now():
     return datetime.fromisoformat(CONSTANTS["time"]).replace(microsecond=0)
 
@@ -149,6 +159,8 @@ async def mock_api_request(url: str, *args, **kwargs):
         return [read_json_file("sample_doorlock")]
     if url == "chimes":
         return [read_json_file("sample_chime")]
+    if url == "aiports":
+        return [read_json_file("sample_aiport")]
     if url.endswith("ptz/preset"):
         return {
             "id": "test-id",
@@ -187,6 +199,8 @@ async def mock_api_request(url: str, *args, **kwargs):
         return read_json_file("sample_doorlock")
     if url.startswith("chimes"):
         return read_json_file("sample_chime")
+    if url.startswith("aiports"):
+        return read_json_file("sample_aiport")
     if "smartDetectTrack" in url:
         return read_json_file("sample_event_smart_track")
 
@@ -457,6 +471,14 @@ async def chime_obj_fixture(protect_client: ProtectApiClient):
     return next(iter(protect_client.bootstrap.chimes.values()))
 
 
+@pytest_asyncio.fixture(name="aiport_obj")
+async def aiport_obj_fixture(protect_client: ProtectApiClient):
+    if not TEST_AIPORT_EXISTS:
+        return None
+
+    return next(iter(protect_client.bootstrap.aiports.values()))
+
+
 @pytest_asyncio.fixture
 async def liveview_obj(protect_client: ProtectApiClient):
     if not TEST_LIVEVIEW_EXISTS:
@@ -502,6 +524,14 @@ def camera():
     return read_camera_json_file()
 
 
+@pytest.fixture()
+def aiport():
+    if not TEST_CAMERA_EXISTS:
+        return None
+
+    return read_aiport_json_file()
+
+
 @pytest.fixture()
 def sensor():
     if not TEST_SENSOR_EXISTS:
@@ -781,7 +811,7 @@ def compare_objs(obj_type, expected, actual):
     actual = deepcopy(actual)
 
     # TODO: fields not supported yet
-    if obj_type == ModelType.CAMERA.value:
+    if obj_type in (ModelType.CAMERA.value, ModelType.AIPORT.value):
         # fields does not always exist (G4 Instant)
         expected.pop("apMac", None)
         # field no longer exists on newer cameras