# SPDX-FileCopyrightText: 2026 Andrew Grimberg <tykeal@bardicgrove.org>
# SPDX-License-Identifier: Apache-2.0
"""High-level async interface for Akuvox device operations."""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
import pylocal_akuvox._device_runtime as _runtime
from pylocal_akuvox import (
_device_access,
_device_config_logs,
_device_contacts,
_device_relays,
_device_users,
)
from pylocal_akuvox._capability_probe import probe_capabilities as _probe_capabilities
from pylocal_akuvox._device_profiles import (
_DEVICE_NOT_IN_MATRIX_NOTE,
_merge_probe_with_matrix,
)
from pylocal_akuvox._http import AkuvoxHttpClient
from pylocal_akuvox.relay import open_door_http as _open
if TYPE_CHECKING:
from pylocal_akuvox import models
from pylocal_akuvox._capability_profile import DeviceCapabilities
from pylocal_akuvox._capability_types import Capability
from pylocal_akuvox._device_runtime import _DeviceContext
from pylocal_akuvox.auth import AuthConfig
__all__ = [
"AkuvoxDevice",
"_DEVICE_NOT_IN_MATRIX_NOTE",
"_merge_probe_with_matrix",
]
[docs]
class AkuvoxDevice:
"""Async context manager for communicating with an Akuvox device."""
[docs]
def __init__(
self,
host: str,
auth: AuthConfig | None = None,
timeout: int = 10,
*,
use_ssl: bool = False,
verify_ssl: bool = True,
request_delay: float = 0.25,
) -> None:
"""Initialize the device connection parameters."""
self._http = AkuvoxHttpClient(
host=host,
auth=auth,
timeout=timeout,
use_ssl=use_ssl,
verify_ssl=verify_ssl,
request_delay=request_delay,
)
self._capabilities: DeviceCapabilities | None = None
self._info: models.DeviceInfo | None = None
self.attempt_unknown_capability: bool = False
@property
def capabilities(self) -> DeviceCapabilities | None:
"""Return the effective capability profile for this connection."""
return self._capabilities
def _context(self) -> _DeviceContext:
"""Build the helper context from current facade state."""
return _runtime.make_context(
self._http,
self._capabilities,
allow_unknown=self.attempt_unknown_capability,
)
def _connection_spec(self) -> dict[str, Any]:
"""Return private connection parameters for diagnostic child sessions."""
return _runtime.get_connection_spec(self._http)
def _require_capabilities(self) -> DeviceCapabilities:
"""Return established capabilities or raise the legacy lifecycle error."""
return _runtime.require_capabilities(self._capabilities)
[docs]
async def probe_capabilities(
self,
*,
timeout: float | None = None,
) -> DeviceCapabilities:
"""Run a non-destructive capability probe against the connected device."""
resolved_timeout = 5.0 if timeout is None else timeout
probe_result = await _probe_capabilities(self._http, timeout=resolved_timeout)
merged = _merge_probe_with_matrix(self._capabilities, probe_result)
self._capabilities = merged
return merged
async def __aenter__(self) -> AkuvoxDevice:
"""Open the underlying HTTP session and populate capabilities."""
await _runtime.enter_device(self)
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: object,
) -> None:
"""Close the underlying HTTP session and clear cached state."""
await _runtime.exit_device(self, exc_type, exc_val, exc_tb)
[docs]
async def get_info(self) -> models.DeviceInfo:
"""Retrieve device identification data."""
return await _runtime.get_info(self._http, self._info)
[docs]
async def get_status(self) -> models.DeviceStatus:
"""Retrieve device operational status."""
return await _runtime.get_status(self._http)
[docs]
async def add_user(
self,
*,
name: str,
user_id: str,
web_relay: str | None = None,
schedule_relay: str,
lift_floor_num: str,
private_pin: str | None = None,
card_code: str | None = None,
) -> None:
"""Add a local user to the device."""
await _device_users.add_user(
self._context(),
name=name,
user_id=user_id,
web_relay=web_relay,
schedule_relay=schedule_relay,
lift_floor_num=lift_floor_num,
private_pin=private_pin,
card_code=card_code,
)
[docs]
async def list_users(self, *, page: int | None = None) -> list[models.User]:
"""List users from the device."""
return await _device_users.list_users(self._context(), page=page)
[docs]
async def modify_user(
self,
*,
id: str,
name: str | None = None,
user_id: str | None = None,
private_pin: str | None = None,
card_code: str | None = None,
web_relay: str | None = None,
schedule_relay: str | None = None,
lift_floor_num: str | None = None,
) -> None:
"""Modify an existing user on the device."""
await _device_users.modify_user(
self._context(),
id=id,
name=name,
user_id=user_id,
private_pin=private_pin,
card_code=card_code,
web_relay=web_relay,
schedule_relay=schedule_relay,
lift_floor_num=lift_floor_num,
)
[docs]
async def delete_user(self, *, id: str) -> None:
"""Delete a user from the device."""
await _device_users.delete_user(self._context(), id=id)
[docs]
async def trigger_relay(
self,
*,
num: int,
mode: int = 0,
level: int = 0,
delay: int = 0,
adapter: Capability | None = None,
) -> None:
"""Trigger a relay to unlock a door or gate."""
await _device_relays.trigger_relay(
self._context(),
num=num,
mode=mode,
level=level,
delay=delay,
adapter=adapter,
)
[docs]
async def open_door_http(
self,
*,
user: str,
password: str,
door_num: int = 1,
) -> None:
"""Unlock via clear-URL OpenDoor HTTP using device relay credentials."""
await _open(self._http, user=user, password=password, door_num=door_num)
def _resolve_override_adapter(
self,
caps: DeviceCapabilities,
adapter: Capability,
) -> Capability:
"""Resolve a caller-supplied ``adapter=`` override."""
return _device_relays.resolve_override_adapter(
caps,
adapter,
allow_unknown=self.attempt_unknown_capability,
)
def _resolve_default_adapter(self, caps: DeviceCapabilities) -> Capability:
"""Walk the preference list and pick the first supported variant."""
return _device_relays.resolve_default_adapter(caps)
[docs]
async def get_relay_status(self) -> dict[str, Any]:
"""Retrieve current relay states from the device."""
return await _device_relays.get_relay_status(self._context())
[docs]
async def get_device_config(self) -> models.DeviceConfig:
"""Retrieve full device configuration."""
return await _device_config_logs.get_device_config(self._context())
[docs]
async def set_device_config(self, settings: dict[str, str]) -> None:
"""Update device configuration settings."""
await _device_config_logs.set_device_config(self._context(), settings)
[docs]
async def add_schedule(
self,
*,
schedule_type: str,
name: str | None = None,
week: str | None = None,
daily: str | None = None,
date_start: str | None = None,
date_end: str | None = None,
time_start: str | None = None,
time_end: str | None = None,
sun: str | None = None,
mon: str | None = None,
tue: str | None = None,
wed: str | None = None,
thur: str | None = None,
fri: str | None = None,
sat: str | None = None,
) -> None:
"""Add an access schedule to the device."""
await _device_access.add_schedule(
self._context(),
schedule_type=schedule_type,
name=name,
week=week,
daily=daily,
date_start=date_start,
date_end=date_end,
time_start=time_start,
time_end=time_end,
sun=sun,
mon=mon,
tue=tue,
wed=wed,
thur=thur,
fri=fri,
sat=sat,
)
[docs]
async def list_schedules(
self, *, page: int | None = None
) -> list[models.AccessSchedule]:
"""List schedules from the device."""
return await _device_access.list_schedules(self._context(), page=page)
[docs]
async def modify_schedule(
self,
*,
id: str,
name: str | None = None,
schedule_type: str | None = None,
week: str | None = None,
daily: str | None = None,
date_start: str | None = None,
date_end: str | None = None,
time_start: str | None = None,
time_end: str | None = None,
sun: str | None = None,
mon: str | None = None,
tue: str | None = None,
wed: str | None = None,
thur: str | None = None,
fri: str | None = None,
sat: str | None = None,
) -> None:
"""Modify an existing schedule on the device."""
await _device_access.modify_schedule(
self._context(),
id=id,
name=name,
schedule_type=schedule_type,
week=week,
daily=daily,
date_start=date_start,
date_end=date_end,
time_start=time_start,
time_end=time_end,
sun=sun,
mon=mon,
tue=tue,
wed=wed,
thur=thur,
fri=fri,
sat=sat,
)
[docs]
async def delete_schedule(self, *, id: str) -> None:
"""Delete a schedule from the device."""
await _device_access.delete_schedule(self._context(), id=id)
[docs]
async def list_groups(self, *, page: int | None = None) -> list[models.Group]:
"""List groups from the device."""
return await _device_access.list_groups(self._context(), page=page)
[docs]
async def add_group(self, *, name: str) -> None:
"""Add a group to the device."""
await _device_access.add_group(self._context(), name=name)
[docs]
async def modify_group(self, *, id: str, name: str) -> None:
"""Modify an existing group on the device."""
await _device_access.modify_group(self._context(), id=id, name=name)
[docs]
async def delete_group(self, *, id: str) -> None:
"""Delete a group from the device."""
await _device_access.delete_group(self._context(), id=id)
[docs]
async def get_door_logs(
self, *, page: int | None = None
) -> list[models.DoorLogEntry]:
"""Retrieve door access logs from the device."""
return await _device_config_logs.get_door_logs(self._context(), page=page)
[docs]
async def get_call_logs(
self, *, page: int | None = None
) -> list[models.CallLogEntry]:
"""Retrieve call logs from the device."""
return await _device_config_logs.get_call_logs(self._context(), page=page)