add Audio API support for axis network speakers

axis/__main__.py

@@ -4,6 +4,7 @@
 import asyncio
 import logging
 
+import ffmpeg
 from httpx import AsyncClient
 
 import axis
@@ -50,7 +51,13 @@ async def axis_device(
 
 
 async def main(
-    host: str, port: int, username: str, password: str, params: bool, events: bool
+    host: str,
+    port: int,
+    username: str,
+    password: str,
+    params: bool,
+    events: bool,
+    audio: str,
 ) -> None:
     """CLI method for library."""
     LOGGER.info("Connecting to Axis device")
@@ -64,6 +71,10 @@ async def main(
     if params:
         await device.vapix.initialize()
 
+    if audio:
+        mulaw = await ffmpeg.to_axis_mulaw(audio)
+        await device.vapix.audio.transmit(mulaw)
+
     if events:
         device.enable_events()
         device.event.subscribe(event_handler)
@@ -90,6 +101,7 @@ async def main(
     parser.add_argument("-p", "--port", type=int, default=80)
     parser.add_argument("--events", action="store_true")
     parser.add_argument("--params", action="store_true")
+    parser.add_argument("--audio", type=str)
     parser.add_argument("-D", "--debug", action="store_true")
     args = parser.parse_args()
 
@@ -99,13 +111,14 @@ async def main(
     logging.basicConfig(format="%(message)s", level=loglevel)
 
     LOGGER.info(
-        "%s, %s, %s, %s, %s, %s",
+        "%s, %s, %s, %s, %s, %s, %s",
         args.host,
         args.username,
         args.password,
         args.port,
         args.events,
         args.params,
+        args.audio,
     )
 
     try:
@@ -117,6 +130,7 @@ async def main(
                 port=args.port,
                 params=args.params,
                 events=args.events,
+                audio=args.audio,
             )
         )
 

axis/ffmpeg.py

@@ -0,0 +1,42 @@
+"""Routines to convert audio into Axis-compatible format with ffmpeg."""
+
+import asyncio
+import shutil
+
+
+async def to_axis_mulaw(audio_path: str, ffmpeg_path: str | None = None) -> bytes:
+    """Convert media to Axis-compatible format."""
+    if not ffmpeg_path:
+        ffmpeg_path = shutil.which("ffmpeg")
+        if not ffmpeg_path:
+            err = "ffmpeg not found in PATH"
+            raise Exception(err)
+
+    args = [
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-i",
+        audio_path,
+        "-vn",
+        "-c:a",
+        "pcm_mulaw",
+        "-b:a",
+        "128k",
+        "-ac",
+        "1",
+        "-ar",
+        "16000",
+        "-f",
+        "mulaw",
+        "-",
+    ]
+
+    process = await asyncio.create_subprocess_exec(
+        ffmpeg_path,
+        *args,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await process.communicate()
+    return stdout

axis/interfaces/audio.py

@@ -0,0 +1,44 @@
+"""Audio api.
+
+https://www.axis.com/vapix-library/subjects/t10100065/section/t10036015/display
+
+The Audio API helps you transmit audio to your Axis device.
+"""
+
+from ..errors import RequestError
+from ..models.api_discovery import ApiId
+from ..models.audio import API_VERSION, TransmitAudioRequest
+from ..models.parameters.audio import AudioParam
+from .api_handler import ApiHandler
+
+
+class AudioHandler(ApiHandler[AudioParam]):
+    """Audio support for Axis devices."""
+
+    api_id = ApiId.AUDIO_STREAMING_CAPABILITIES
+    default_api_version = API_VERSION
+
+    @property
+    def listed_in_parameters(self) -> bool:
+        """Is API listed in parameters."""
+        if prop := self.vapix.params.property_handler.get("0"):
+            return prop.audio
+        return False
+
+    async def _api_request(self) -> dict[str, AudioParam]:
+        """Get API data method defined by subclass."""
+        return await self.get_audio_params()
+
+    async def get_audio_params(self) -> dict[str, AudioParam]:
+        """Retrieve audio params."""
+        await self.vapix.params.audio_handler.update()
+        return dict(self.vapix.params.audio_handler.items())
+
+    async def transmit(self, audio: bytes) -> None:
+        """Transmit audio to play on the speaker."""
+        try:
+            await self.vapix.api_request(TransmitAudioRequest(audio=audio))
+        except RequestError as e:
+            # the transmit.cgi API will not return a HTTP response until the audio has finished playing. therefore, if the audio is longer than the vapix request timeout, it will throw an exception, even when the request is successful. so, instead we set a short httpx read timeout, and then dampen that exception if it occurs, while preserving exceptions for other timeout conditions
+            if str(e) != "Read Timeout":
+                raise

axis/interfaces/audio_device_control.py

@@ -0,0 +1,96 @@
+"""Audio Device Control API.
+
+https://www.axis.com/vapix-library/subjects/t10100065/section/t10169359/display
+
+The Audio Device Control API helps you configure and control your Axis
+audio devices.
+"""
+
+from typing import Any
+
+from ..models.api_discovery import ApiId
+from ..models.audio_device_control import (
+    API_VERSION,
+    AudioDevice,
+    GetDevicesSettingsRequest,
+    GetDevicesSettingsResponse,
+    GetSupportedVersionsRequest,
+    GetSupportedVersionsResponse,
+    SetDevicesSettingsRequest,
+)
+from .api_handler import ApiHandler
+
+
+class AudioDeviceControlHandler(ApiHandler[Any]):
+    """Audio Device Control for Axis devices."""
+
+    api_id = ApiId.AUDIO_DEVICE_CONTROL
+    default_api_version = API_VERSION
+
+    async def get_devices_settings(self) -> list[AudioDevice]:
+        """Get audio devices settings."""
+        bytes_data = await self.vapix.api_request(GetDevicesSettingsRequest())
+        response = GetDevicesSettingsResponse.decode(bytes_data)
+        return response.data
+
+    async def get_gain_mute(self) -> tuple[int, bool]:
+        """Shortcut to get the gain/mute of the output audio device."""
+        settings = await self.get_devices_settings()
+        channel = (
+            settings[0].outputs[0].connection_types[0].signaling_types[0].channels[0]
+        )
+        return (channel.gain, channel.mute)
+
+    async def mute(self) -> None:
+        """Shortcut to mute the output audio device."""
+        await self._set_gain_mute(mute=True)
+
+    async def unmute(self) -> None:
+        """Shortcut to unmute the output audio device."""
+        await self._set_gain_mute(mute=False)
+
+    async def set_gain(self, gain: int) -> None:
+        """Shortcut to set the gain on the output audio device."""
+        await self._set_gain_mute(gain=gain)
+
+    async def _set_gain_mute(
+        self, gain: int | None = None, mute: bool | None = None
+    ) -> None:
+        """Shortcut to control gain/mute state."""
+        mute_param = {"mute": mute} if mute in (True, False) else {}
+        gain_param = {"gain": gain} if gain else {}
+
+        devices = [
+            {
+                "id": "0",
+                "outputs": [
+                    {
+                        "id": "0",
+                        "connectionTypes": [
+                            {
+                                "id": "internal",
+                                "signalingTypes": [
+                                    {
+                                        "id": "unbalanced",
+                                        "channels": [
+                                            {
+                                                "id": "0",
+                                                **(gain_param),
+                                                **(mute_param),
+                                            }
+                                        ],
+                                    }
+                                ],
+                            }
+                        ],
+                    }
+                ],
+            }
+        ]
+        await self.vapix.api_request(SetDevicesSettingsRequest(devices))
+
+    async def get_supported_versions(self) -> list[str]:
+        """List supported API versions."""
+        bytes_data = await self.vapix.api_request(GetSupportedVersionsRequest())
+        response = GetSupportedVersionsResponse.decode(bytes_data)
+        return response.data

axis/interfaces/parameters/audio.py

@@ -0,0 +1,12 @@
+"""Audio parameters."""
+
+from ...models.parameters.audio import AudioParam
+from ...models.parameters.param_cgi import ParameterGroup
+from .param_handler import ParamHandler
+
+
+class AudioParameterHandler(ParamHandler[AudioParam]):
+    """Handler for audio parameters."""
+
+    parameter_group = ParameterGroup.AUDIO
+    parameter_item = AudioParam

axis/interfaces/parameters/param_cgi.py

@@ -6,6 +6,7 @@
 from ...models.api_discovery import ApiId
 from ...models.parameters.param_cgi import ParameterGroup, ParamRequest, params_to_dict
 from ..api_handler import ApiHandler
+from .audio import AudioParameterHandler
 from .brand import BrandParameterHandler
 from .image import ImageParameterHandler
 from .io_port import IOPortParameterHandler
@@ -26,6 +27,7 @@ def __init__(self, vapix: "Vapix") -> None:
         """Initialize parameter classes."""
         super().__init__(vapix)
 
+        self.audio_handler = AudioParameterHandler(self)
         self.brand_handler = BrandParameterHandler(self)
         self.image_handler = ImageParameterHandler(self)
         self.io_port_handler = IOPortParameterHandler(self)

axis/interfaces/vapix.py

@@ -23,6 +23,8 @@
     ObjectAnalyticsHandler,
 )
 from .applications.vmd4 import Vmd4Handler
+from .audio import AudioHandler
+from .audio_device_control import AudioDeviceControlHandler
 from .basic_device_info import BasicDeviceInfoHandler
 from .event_instances import EventInstanceHandler
 from .light_control import LightHandler
@@ -45,8 +47,6 @@
 
 LOGGER = logging.getLogger(__name__)
 
-TIME_OUT = 15
-
 
 class Vapix:
     """Vapix parameter request."""
@@ -62,6 +62,8 @@ def __init__(self, device: AxisDevice) -> None:
         self.api_discovery: ApiDiscoveryHandler = ApiDiscoveryHandler(self)
         self.params: Params = Params(self)
 
+        self.audio = AudioHandler(self)
+        self.audio_device_control = AudioDeviceControlHandler(self)
         self.basic_device_info = BasicDeviceInfoHandler(self)
         self.io_port_management = IoPortManagement(self)
         self.light_control = LightHandler(self)
@@ -153,6 +155,8 @@ async def initialize_api_discovery(self) -> None:
             return
 
         apis: tuple[ApiHandler[Any], ...] = (
+            self.audio,
+            self.audio_device_control,
             self.basic_device_info,
             self.io_port_management,
             self.light_control,
@@ -249,33 +253,41 @@ async def api_request(self, api_request: ApiRequest) -> bytes:
             data=api_request.data,
             headers=api_request.headers,
             params=api_request.params,
+            timeout=api_request.timeout,
         )
 
     async def request(
         self,
         method: str,
         path: str,
+        timeout: int | httpx.Timeout,  # noqa: ASYNC109
         content: bytes | None = None,
         data: dict[str, str] | None = None,
         headers: dict[str, str] | None = None,
         params: dict[str, str] | None = None,
     ) -> bytes:
         """Make a request to the device."""
         url = self.device.config.url + path
-        LOGGER.debug("%s, %s, '%s', '%s', '%s'", method, url, content, data, params)
+        LOGGER.debug(
+            "%s, %s, '%s', '%s', '%s', %s", method, url, content, data, params, timeout
+        )
 
         try:
             response = await self.device.config.session.request(
                 method,
                 url,
+                timeout=timeout,
                 content=content,
                 data=data,
                 headers=headers,
                 params=params,
                 auth=self.auth,
-                timeout=TIME_OUT,
             )
 
+        except httpx.ReadTimeout as errt:
+            message = "Read Timeout"
+            raise RequestError(message) from errt
+
         except httpx.TimeoutException as errt:
             message = "Timeout"
             raise RequestError(message) from errt

axis/models/api.py

@@ -4,7 +4,10 @@
 from dataclasses import dataclass, field
 from typing import Any, Generic, Self, TypeVar
 
+import httpx
+
 CONTEXT = "Axis library"
+DEFAULT_TIMEOUT = 15
 
 
 @dataclass(frozen=True)
@@ -64,6 +67,7 @@ class ApiRequest:
     method: str = field(init=False)
     path: str = field(init=False)
     content_type: str = field(init=False)
+    timeout: int | httpx.Timeout = field(init=False, default=DEFAULT_TIMEOUT)
 
     @property
     def content(self) -> bytes | None: