Use IntEnum for stream orientation (#81835)

* Use IntEnum for stream orientation

* Rename enum values

* Add comments

* Fix import

homeassistant/components/camera/__init__.py

@@ -30,6 +30,7 @@ from homeassistant.components.media_player import (
 from homeassistant.components.stream import (
     FORMAT_CONTENT_TYPE,
     OUTPUT_FORMATS,
+    Orientation,
     Stream,
     create_stream,
 )
@@ -869,7 +870,7 @@ async def websocket_get_prefs(
         vol.Required("type"): "camera/update_prefs",
         vol.Required("entity_id"): cv.entity_id,
         vol.Optional(PREF_PRELOAD_STREAM): bool,
-        vol.Optional(PREF_ORIENTATION): vol.All(int, vol.Range(min=1, max=8)),
+        vol.Optional(PREF_ORIENTATION): vol.Coerce(Orientation),
     }
 )
 @websocket_api.async_response

homeassistant/components/camera/prefs.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 from typing import Final, Union, cast
 
+from homeassistant.components.stream import Orientation
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import entity_registry as er
@@ -18,11 +19,11 @@ STORAGE_VERSION: Final = 1
 class CameraEntityPreferences:
     """Handle preferences for camera entity."""
 
-    def __init__(self, prefs: dict[str, bool | int]) -> None:
+    def __init__(self, prefs: dict[str, bool | Orientation]) -> None:
         """Initialize prefs."""
         self._prefs = prefs
 
-    def as_dict(self) -> dict[str, bool | int]:
+    def as_dict(self) -> dict[str, bool | Orientation]:
         """Return dictionary version."""
         return self._prefs
 
@@ -32,9 +33,11 @@ class CameraEntityPreferences:
         return cast(bool, self._prefs.get(PREF_PRELOAD_STREAM, False))
 
     @property
-    def orientation(self) -> int:
+    def orientation(self) -> Orientation:
         """Return the current stream orientation settings."""
-        return self._prefs.get(PREF_ORIENTATION, 1)
+        return cast(
+            Orientation, self._prefs.get(PREF_ORIENTATION, Orientation.NO_TRANSFORM)
+        )
 
 
 class CameraPreferences:
@@ -45,11 +48,11 @@ class CameraPreferences:
         self._hass = hass
         # The orientation prefs are stored in in the entity registry options
         # The preload_stream prefs are stored in this Store
-        self._store = Store[dict[str, dict[str, Union[bool, int]]]](
+        self._store = Store[dict[str, dict[str, Union[bool, Orientation]]]](
             hass, STORAGE_VERSION, STORAGE_KEY
         )
         # Local copy of the preload_stream prefs
-        self._prefs: dict[str, dict[str, bool | int]] | None = None
+        self._prefs: dict[str, dict[str, bool | Orientation]] | None = None
 
     async def async_initialize(self) -> None:
         """Finish initializing the preferences."""
@@ -63,9 +66,9 @@ class CameraPreferences:
         entity_id: str,
         *,
         preload_stream: bool | UndefinedType = UNDEFINED,
-        orientation: int | UndefinedType = UNDEFINED,
+        orientation: Orientation | UndefinedType = UNDEFINED,
         stream_options: dict[str, str] | UndefinedType = UNDEFINED,
-    ) -> dict[str, bool | int]:
+    ) -> dict[str, bool | Orientation]:
         """Update camera preferences.
 
         Returns a dict with the preferences on success.

homeassistant/components/stream/__init__.py

@@ -63,6 +63,7 @@ from .core import (
     STREAM_SETTINGS_NON_LL_HLS,
     IdleTimer,
     KeyFrameConverter,
+    Orientation,
     StreamOutput,
     StreamSettings,
 )
@@ -82,6 +83,7 @@ __all__ = [
     "SOURCE_TIMEOUT",
     "Stream",
     "create_stream",
+    "Orientation",
 ]
 
 _LOGGER = logging.getLogger(__name__)
@@ -229,7 +231,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
             part_target_duration=conf[CONF_PART_DURATION],
             hls_advance_part_limit=max(int(3 / conf[CONF_PART_DURATION]), 3),
             hls_part_timeout=2 * conf[CONF_PART_DURATION],
-            orientation=1,
+            orientation=Orientation.NO_TRANSFORM,
         )
     else:
         hass.data[DOMAIN][ATTR_SETTINGS] = STREAM_SETTINGS_NON_LL_HLS
@@ -292,12 +294,12 @@ class Stream:
         self._diagnostics = Diagnostics()
 
     @property
-    def orientation(self) -> int:
+    def orientation(self) -> Orientation:
         """Return the current orientation setting."""
         return self._stream_settings.orientation
 
     @orientation.setter
-    def orientation(self, value: int) -> None:
+    def orientation(self, value: Orientation) -> None:
         """Set the stream orientation setting."""
         self._stream_settings.orientation = value
 

homeassistant/components/stream/core.py

@@ -5,6 +5,7 @@ import asyncio
 from collections import deque
 from collections.abc import Callable, Coroutine, Iterable
 import datetime
+from enum import IntEnum
 import logging
 from typing import TYPE_CHECKING, Any
 
@@ -35,6 +36,19 @@ _LOGGER = logging.getLogger(__name__)
 PROVIDERS: Registry[str, type[StreamOutput]] = Registry()
 
 
+class Orientation(IntEnum):
+    """Orientations for stream transforms. These are based on EXIF orientation tags."""
+
+    NO_TRANSFORM = 1
+    MIRROR = 2
+    ROTATE_180 = 3
+    FLIP = 4
+    ROTATE_LEFT_AND_FLIP = 5
+    ROTATE_LEFT = 6
+    ROTATE_RIGHT_AND_FLIP = 7
+    ROTATE_RIGHT = 8
+
+
 @attr.s(slots=True)
 class StreamSettings:
     """Stream settings."""
@@ -44,7 +58,7 @@ class StreamSettings:
     part_target_duration: float = attr.ib()
     hls_advance_part_limit: int = attr.ib()
     hls_part_timeout: float = attr.ib()
-    orientation: int = attr.ib()
+    orientation: Orientation = attr.ib()
 
 
 STREAM_SETTINGS_NON_LL_HLS = StreamSettings(
@@ -53,7 +67,7 @@ STREAM_SETTINGS_NON_LL_HLS = StreamSettings(
     part_target_duration=TARGET_SEGMENT_DURATION_NON_LL_HLS,
     hls_advance_part_limit=3,
     hls_part_timeout=TARGET_SEGMENT_DURATION_NON_LL_HLS,
-    orientation=1,
+    orientation=Orientation.NO_TRANSFORM,
 )
 
 

homeassistant/components/stream/fmp4utils.py

@@ -6,6 +6,8 @@ from typing import TYPE_CHECKING
 
 from homeassistant.exceptions import HomeAssistantError
 
+from .core import Orientation
+
 if TYPE_CHECKING:
     from io import BufferedIOBase
 
@@ -179,22 +181,24 @@ ROTATE_LEFT_FLIP = (ZERO32 + NEGONE32 + ZERO32) + (NEGONE32 + ZERO32 + ZERO32)
 ROTATE_RIGHT_FLIP = (ZERO32 + ONE32 + ZERO32) + (ONE32 + ZERO32 + ZERO32)
 
 TRANSFORM_MATRIX_TOP = (
-    # The first two entries are just to align the indices with the EXIF orientation tags
-    b"",
-    b"",
-    MIRROR,
-    ROTATE_180,
-    FLIP,
-    ROTATE_LEFT_FLIP,
-    ROTATE_LEFT,
-    ROTATE_RIGHT_FLIP,
-    ROTATE_RIGHT,
+    # The index into this tuple corresponds to the EXIF orientation tag
+    # Only index values of 2 through 8 are used
+    # The first two entries are just to keep everything aligned
+    b"",  # 0
+    b"",  # 1
+    MIRROR,  # 2
+    ROTATE_180,  # 3
+    FLIP,  # 4
+    ROTATE_LEFT_FLIP,  # 5
+    ROTATE_LEFT,  # 6
+    ROTATE_RIGHT_FLIP,  # 7
+    ROTATE_RIGHT,  # 8
 )
 
 
-def transform_init(init: bytes, orientation: int) -> bytes:
+def transform_init(init: bytes, orientation: Orientation) -> bytes:
     """Change the transformation matrix in the header."""
-    if orientation == 1:
+    if orientation == Orientation.NO_TRANSFORM:
         return init
     # Find moov
     moov_location = next(find_box(init, b"moov"))

tests/components/camera/test_init.py

@@ -367,7 +367,7 @@ async def test_websocket_update_orientation_prefs(hass, hass_ws_client, mock_cam
     assert response["success"]
 
     er_camera_prefs = registry.async_get("camera.demo_uniquecamera").options[DOMAIN]
-    assert er_camera_prefs[PREF_ORIENTATION] == 3
+    assert er_camera_prefs[PREF_ORIENTATION] == camera.Orientation.ROTATE_180
     assert response["result"][PREF_ORIENTATION] == er_camera_prefs[PREF_ORIENTATION]
     # Check that the preference was saved
     await client.send_json(
@@ -375,7 +375,7 @@ async def test_websocket_update_orientation_prefs(hass, hass_ws_client, mock_cam
     )
     msg = await client.receive_json()
     # orientation entry for this camera should have been added
-    assert msg["result"]["orientation"] == 3
+    assert msg["result"]["orientation"] == camera.Orientation.ROTATE_180
 
 
 async def test_play_stream_service_no_source(hass, mock_camera, mock_stream):