Option to select what video source Axis camera should use (#45268)

* Fully working proposal of config option to select what video source camera entity should use

* Bump dependency to v43
Reflect dependency changes in how image sources is now a dict

* Fix bdracos comment

homeassistant/components/axis/camera.py

@@ -1,5 +1,7 @@
 """Support for Axis camera streaming."""
 
+from urllib.parse import urlencode
+
 from homeassistant.components.camera import SUPPORT_STREAM
 from homeassistant.components.mjpeg.camera import (
     CONF_MJPEG_URL,
@@ -17,7 +19,7 @@ from homeassistant.const import (
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
 
 from .axis_base import AxisEntityBase
-from .const import DEFAULT_STREAM_PROFILE, DOMAIN as AXIS_DOMAIN
+from .const import DEFAULT_STREAM_PROFILE, DEFAULT_VIDEO_SOURCE, DOMAIN as AXIS_DOMAIN
 
 
 async def async_setup_entry(hass, config_entry, async_add_entities):
@@ -60,38 +62,55 @@ class AxisCamera(AxisEntityBase, MjpegCamera):
         await super().async_added_to_hass()
 
     @property
-    def supported_features(self):
+    def supported_features(self) -> int:
         """Return supported features."""
         return SUPPORT_STREAM
 
-    def _new_address(self):
+    def _new_address(self) -> None:
         """Set new device address for video stream."""
         self._mjpeg_url = self.mjpeg_source
         self._still_image_url = self.image_source
 
     @property
-    def unique_id(self):
+    def unique_id(self) -> str:
         """Return a unique identifier for this device."""
         return f"{self.device.unique_id}-camera"
 
     @property
-    def image_source(self):
+    def image_source(self) -> str:
         """Return still image URL for device."""
-        return f"http://{self.device.host}:{self.device.port}/axis-cgi/jpg/image.cgi"
+        options = self.generate_options(skip_stream_profile=True)
+        return f"http://{self.device.host}:{self.device.port}/axis-cgi/jpg/image.cgi{options}"
 
     @property
-    def mjpeg_source(self):
+    def mjpeg_source(self) -> str:
         """Return mjpeg URL for device."""
-        options = ""
+        options = self.generate_options()
-        if self.device.option_stream_profile != DEFAULT_STREAM_PROFILE:
-            options = f"?&streamprofile={self.device.option_stream_profile}"
-
         return f"http://{self.device.host}:{self.device.port}/axis-cgi/mjpg/video.cgi{options}"
 
-    async def stream_source(self):
+    async def stream_source(self) -> str:
         """Return the stream source."""
-        options = ""
+        options = self.generate_options(add_video_codec_h264=True)
-        if self.device.option_stream_profile != DEFAULT_STREAM_PROFILE:
+        return f"rtsp://{self.device.username}:{self.device.password}@{self.device.host}/axis-media/media.amp{options}"
-            options = f"&streamprofile={self.device.option_stream_profile}"
 
-        return f"rtsp://{self.device.username}:{self.device.password}@{self.device.host}/axis-media/media.amp?videocodec=h264{options}"
+    def generate_options(
+        self, skip_stream_profile: bool = False, add_video_codec_h264: bool = False
+    ) -> str:
+        """Generate options for video stream."""
+        options_dict = {}
+
+        if add_video_codec_h264:
+            options_dict["videocodec"] = "h264"
+
+        if (
+            not skip_stream_profile
+            and self.device.option_stream_profile != DEFAULT_STREAM_PROFILE
+        ):
+            options_dict["streamprofile"] = self.device.option_stream_profile
+
+        if self.device.option_video_source != DEFAULT_VIDEO_SOURCE:
+            options_dict["camera"] = self.device.option_video_source
+
+        if not options_dict:
+            return ""
+        return f"?{urlencode(options_dict)}"

homeassistant/components/axis/config_flow.py

@@ -22,7 +22,9 @@ from homeassistant.util.network import is_link_local
 from .const import (
     CONF_MODEL,
     CONF_STREAM_PROFILE,
+    CONF_VIDEO_SOURCE,
     DEFAULT_STREAM_PROFILE,
+    DEFAULT_VIDEO_SOURCE,
     DOMAIN as AXIS_DOMAIN,
 )
 from .device import get_device
@@ -220,22 +222,44 @@ class AxisOptionsFlowHandler(config_entries.OptionsFlow):
         return await self.async_step_configure_stream()
 
     async def async_step_configure_stream(self, user_input=None):
-        """Manage the Axis device options."""
+        """Manage the Axis device stream options."""
         if user_input is not None:
             self.options.update(user_input)
             return self.async_create_entry(title="", data=self.options)
 
-        profiles = [DEFAULT_STREAM_PROFILE]
+        schema = {}
-        for profile in self.device.api.vapix.streaming_profiles:
-            profiles.append(profile.name)
 
-        return self.async_show_form(
+        vapix = self.device.api.vapix
-            step_id="configure_stream",
+
-            data_schema=vol.Schema(
+        # Stream profiles
-                {
+
+        if vapix.params.stream_profiles_max_groups > 0:
+
+            stream_profiles = [DEFAULT_STREAM_PROFILE]
+            for profile in vapix.streaming_profiles:
+                stream_profiles.append(profile.name)
+
+            schema[
                 vol.Optional(
                     CONF_STREAM_PROFILE, default=self.device.option_stream_profile
-                    ): vol.In(profiles)
+                )
-                }
+            ] = vol.In(stream_profiles)
-            ),
+
+        # Video sources
+
+        if vapix.params.image_nbrofviews > 0:
+            await vapix.params.update_image()
+
+            video_sources = {DEFAULT_VIDEO_SOURCE: DEFAULT_VIDEO_SOURCE}
+            for idx, video_source in vapix.params.image_sources.items():
+                if not video_source["Enabled"]:
+                    continue
+                video_sources[idx + 1] = video_source["Name"]
+
+            schema[
+                vol.Optional(CONF_VIDEO_SOURCE, default=self.device.option_video_source)
+            ] = vol.In(video_sources)
+
+        return self.async_show_form(
+            step_id="configure_stream", data_schema=vol.Schema(schema)
         )

homeassistant/components/axis/const.py

@@ -15,9 +15,11 @@ ATTR_MANUFACTURER = "Axis Communications AB"
 CONF_EVENTS = "events"
 CONF_MODEL = "model"
 CONF_STREAM_PROFILE = "stream_profile"
+CONF_VIDEO_SOURCE = "video_source"
 
 DEFAULT_EVENTS = True
 DEFAULT_STREAM_PROFILE = "No stream profile"
 DEFAULT_TRIGGER_TIME = 0
+DEFAULT_VIDEO_SOURCE = "No video source"
 
 PLATFORMS = [BINARY_SENSOR_DOMAIN, CAMERA_DOMAIN, LIGHT_DOMAIN, SWITCH_DOMAIN]

homeassistant/components/axis/device.py

@@ -34,9 +34,11 @@ from .const import (
     CONF_EVENTS,
     CONF_MODEL,
     CONF_STREAM_PROFILE,
+    CONF_VIDEO_SOURCE,
     DEFAULT_EVENTS,
     DEFAULT_STREAM_PROFILE,
     DEFAULT_TRIGGER_TIME,
+    DEFAULT_VIDEO_SOURCE,
     DOMAIN as AXIS_DOMAIN,
     LOGGER,
     PLATFORMS,
@@ -113,6 +115,11 @@ class AxisNetworkDevice:
         """Config entry option defining minimum number of seconds to keep trigger high."""
         return self.config_entry.options.get(CONF_TRIGGER_TIME, DEFAULT_TRIGGER_TIME)
 
+    @property
+    def option_video_source(self):
+        """Config entry option defining what video source camera platform should use."""
+        return self.config_entry.options.get(CONF_VIDEO_SOURCE, DEFAULT_VIDEO_SOURCE)
+
     # Signals
 
     @property

homeassistant/components/axis/manifest.json

@@ -3,7 +3,7 @@
   "name": "Axis",
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/axis",
-  "requirements": ["axis==42"],
+  "requirements": ["axis==43"],
   "zeroconf": [
     { "type": "_axis-video._tcp.local.", "macaddress": "00408C*" },
     { "type": "_axis-video._tcp.local.", "macaddress": "ACCC8E*" },

requirements_all.txt

@@ -306,7 +306,7 @@ av==8.0.2
 # avion==0.10
 
 # homeassistant.components.axis
-axis==42
+axis==43
 
 # homeassistant.components.azure_event_hub
 azure-eventhub==5.1.0

requirements_test_all.txt

@@ -177,7 +177,7 @@ auroranoaa==0.0.2
 av==8.0.2
 
 # homeassistant.components.axis
-axis==42
+axis==43
 
 # homeassistant.components.azure_event_hub
 azure-eventhub==5.1.0

tests/components/axis/test_camera.py

@@ -64,7 +64,7 @@ async def test_camera_with_stream_profile(hass):
     assert camera_entity.image_source == "http://1.2.3.4:80/axis-cgi/jpg/image.cgi"
     assert (
         camera_entity.mjpeg_source
-        == "http://1.2.3.4:80/axis-cgi/mjpg/video.cgi?&streamprofile=profile_1"
+        == "http://1.2.3.4:80/axis-cgi/mjpg/video.cgi?streamprofile=profile_1"
     )
     assert (
         await camera_entity.stream_source()

tests/components/axis/test_config_flow.py

@@ -9,7 +9,9 @@ from homeassistant.components.axis.const import (
     CONF_EVENTS,
     CONF_MODEL,
     CONF_STREAM_PROFILE,
+    CONF_VIDEO_SOURCE,
     DEFAULT_STREAM_PROFILE,
+    DEFAULT_VIDEO_SOURCE,
     DOMAIN as AXIS_DOMAIN,
 )
 from homeassistant.components.dhcp import HOSTNAME, IP_ADDRESS, MAC_ADDRESS
@@ -530,8 +532,13 @@ async def test_option_flow(hass):
     config_entry = await setup_axis_integration(hass)
     device = hass.data[AXIS_DOMAIN][config_entry.unique_id]
     assert device.option_stream_profile == DEFAULT_STREAM_PROFILE
+    assert device.option_video_source == DEFAULT_VIDEO_SOURCE
 
-    result = await hass.config_entries.options.async_init(device.config_entry.entry_id)
+    with respx.mock:
+        mock_default_vapix_requests(respx)
+        result = await hass.config_entries.options.async_init(
+            device.config_entry.entry_id
+        )
 
     assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
     assert result["step_id"] == "configure_stream"
@@ -540,15 +547,21 @@ async def test_option_flow(hass):
         "profile_1",
         "profile_2",
     }
+    assert set(result["data_schema"].schema[CONF_VIDEO_SOURCE].container) == {
+        DEFAULT_VIDEO_SOURCE,
+        1,
+    }
 
     result = await hass.config_entries.options.async_configure(
         result["flow_id"],
-        user_input={CONF_STREAM_PROFILE: "profile_1"},
+        user_input={CONF_STREAM_PROFILE: "profile_1", CONF_VIDEO_SOURCE: 1},
     )
 
     assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
     assert result["data"] == {
         CONF_EVENTS: True,
         CONF_STREAM_PROFILE: "profile_1",
+        CONF_VIDEO_SOURCE: 1,
     }
     assert device.option_stream_profile == "profile_1"
+    assert device.option_video_source == 1