Merge pull request #264 from Xyaren/dependabot/pip/homeassistant-b03d…

…bfd2e1

Update to 2025.1.2

.devcontainer/devcontainer.json

@@ -1,6 +1,6 @@
 {
   "name": "xyaren/homeassistant-magentatv",
-  "image": "mcr.microsoft.com/devcontainers/python:3.12",
+  "image": "mcr.microsoft.com/devcontainers/python:3.13",
   "postCreateCommand": "scripts/setup",
   "runArgs": [
     "--network=host"

.github/workflows/lint.yml

@@ -20,7 +20,7 @@ jobs:
         - name: "Set up Python"
           uses: actions/[email protected]
           with:
-            python-version: "3.12"
+            python-version: "3.13"
             cache: "pip"
 
         - name: "Install requirements"

.github/workflows/test.yml

@@ -20,7 +20,7 @@ jobs:
         - name: "Set up Python"
           uses: actions/[email protected]
           with:
-            python-version: "3.12"
+            python-version: "3.13"
             cache: "pip"
 
         - name: "Install requirements"
@@ -55,7 +55,7 @@ jobs:
       - name: "Set up Python"
         uses: actions/[email protected]
         with:
-          python-version: "3.12"
+          python-version: "3.13"
       - name: Install coverage.py
         run: pip install coverage
 
@@ -89,7 +89,7 @@ jobs:
       - name: "Set up Python"
         uses: actions/[email protected]
         with:
-          python-version: "3.12"
+          python-version: "3.13"
       - name: Install coverage.py
         run: pip install coverage
 

custom_components/magentatv/api/client.py

@@ -146,6 +146,7 @@ async def async_get_player_state(self) -> str:
         result = {}
         for child in tree[0][0]:
             result[child.tag] = child.text
+
         return result
 
     async def _async_send_pairing_request(self):

custom_components/magentatv/api/event_model.py

@@ -1,35 +1,26 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 
 class EventModel(BaseModel):
     """Pydantic baseclass with overloaded operator for
     instantiating new objects.
     """
 
-    def __new__(cls, *args, **kwargs):
-        """Be careful when using this method:
-        https://docs.python.org/3/reference/datamodel.html#object.__new__.
-        """
-        # If all args are none -> do nothing
-        if all(v is None for v in args) and all(v is None for v in kwargs.values()):
-            pass
-        else:
-            return super().__new__(cls)
+    model_config = ConfigDict(populate_by_name=False, frozen=True)
 
-    class Config:
-        allow_population_by_field_name = True
-        allow_mutation = False
+    def set_keys(self):
+        return self.model_dump(exclude_unset=True, by_alias=False).keys()
 
 
 class PlayContentEvent(EventModel):
-    new_play_mode: int | None
-    play_back_state: int | None = Field(alias="playBackState")
-    media_type: int | None = Field(alias="mediaType")
-    media_code: str | None = Field(alias="mediaCode")
-    duration: int | None
-    play_position: int | None = Field(alias="playPostion")  # not a typo !
-    fast_speed: int | None = Field(alias="fastSpeed")
-    chan_key: int | None = Field(alias="chanKey")
+    new_play_mode: int | None = None
+    play_back_state: int | None = Field(alias="playBackState", default=None)
+    media_type: int | None = Field(alias="mediaType", default=None)
+    media_code: str | None = Field(alias="mediaCode", default=None)
+    duration: int | None = None
+    play_position: int | None = Field(alias="playPostion", default=None)  # not a typo !
+    fast_speed: int | None = Field(alias="fastSpeed", default=None)
+    chan_key: int | None = Field(alias="chanKey", default=None)
 
 
 class ShortEvent(EventModel):
@@ -50,7 +41,12 @@ class ProgramInfo(EventModel):
 class EitChangedEvent(EventModel):
     type: str
     instance_id: int
-    channel_code: str
-    channel_num: str
+    channel_code: int
+    channel_num: int
     media_id: str = Field(alias="mediaId")
-    program_info: list[ProgramInfo]
+    program_info: list[ProgramInfo | None]
+
+    @field_validator("program_info", mode="before")
+    @classmethod
+    def filter_empty_program_info(cls, value) -> list[ProgramInfo | None]:
+        return [x if x else None for x in value]

custom_components/magentatv/api/state_machine.py

@@ -2,7 +2,7 @@
 from enum import Enum
 from logging import Logger, getLogger
 
-from .event_model import ProgramInfo
+from .event_model import EitChangedEvent, PlayContentEvent, ProgramInfo
 
 LOGGER: Logger = getLogger(__package__ + ".state_machine")
 
@@ -26,7 +26,7 @@ class MediaReceiverStateMachine:
     program_current: ProgramInfo | None = None
     program_next: ProgramInfo | None = None
 
-    _last_poll_player_state = None
+    _last_poll_player_state: PlayContentEvent | None = None
     _last_event_play_content = None
     _last_event_eit_changed = None
 
@@ -37,7 +37,7 @@ class MediaReceiverStateMachine:
     def on_connection_error(self) -> None:
         self._available = False
 
-    def on_event_eit_changed(self, data: dict) -> None:
+    def on_event_eit_changed(self, data: EitChangedEvent) -> None:
         LOGGER.debug("On Event EitChanged: %s", data)
         self._available = True
 
@@ -49,16 +49,15 @@ def on_event_eit_changed(self, data: dict) -> None:
 
         self._last_event_eit_changed = data
 
-    def _on_event_eit_changed_changed(self, data) -> None:
-        if "channel_num" in data:
-            self.chan_key = int(data["channel_num"])
+    def _on_event_eit_changed_changed(self, data: EitChangedEvent) -> None:
+        if "channel_num" in data.set_keys():
+            self.chan_key = data.channel_num
 
-        if "program_info" in data:
-            programm_info = data["program_info"]
-            self.program_current = ProgramInfo(**programm_info[0])
-            self.program_next = ProgramInfo(**programm_info[1])
+        if "program_info" in data.set_keys():
+            self.program_current = data.program_info[0] or None
+            self.program_next = data.program_info[1] or None
 
-    def on_event_play_content(self, data: dict) -> None:
+    def on_event_play_content(self, data: PlayContentEvent) -> None:
         LOGGER.debug("On Event PlayContent: %s", data)
         self._available = True
 
@@ -70,9 +69,9 @@ def on_event_play_content(self, data: dict) -> None:
 
         self._last_event_play_content = data
 
-    def _on_event_play_content_changed(self, data: dict) -> None:
-        if "new_play_mode" in data:
-            if data["new_play_mode"] == 20:
+    def _on_event_play_content_changed(self, data: PlayContentEvent) -> None:
+        if data.new_play_mode is not None:
+            if data.new_play_mode == 20:
                 self.state = State.BUFFERING
                 self.duration = None
                 self.position = None
@@ -82,7 +81,7 @@ def _on_event_play_content_changed(self, data: dict) -> None:
                 self._ignore_next_poll_event = True
                 return
             # [2, 3, 4, 5]
-            elif data["new_play_mode"] == 4:
+            elif data.new_play_mode == 4:
                 self.state = State.PLAYING
                 self.duration = 0
                 self.position = 0
@@ -91,27 +90,27 @@ def _on_event_play_content_changed(self, data: dict) -> None:
                 self._ignore_next_poll_event = True
                 return
 
-            elif data["new_play_mode"] == 2:
+            elif data.new_play_mode == 2:
                 # play after pause -> time shift ?
                 self.state = State.PLAYING
-                self.duration = data["duration"]
-                self.position = data["playPostion"]
+                self.duration = data.duration
+                self.position = data.play_position
                 self.position_last_update = dt.datetime.now()
                 return
 
-            elif data["new_play_mode"] == 1:
+            elif data.new_play_mode == 1:
                 self.state = State.PAUSED
-                self.duration = data["duration"]
-                self.position = data["playPostion"]
+                self.duration = data.duration
+                self.position = data.play_position
                 self.position_last_update = dt.datetime.now()
                 return
 
-            elif data["new_play_mode"] == 0:
+            elif data.new_play_mode == 0:
                 self.state = State.OFF
                 self._clear_non_state_attributes()
                 return
 
-    def on_poll_player_state(self, data: dict) -> None:
+    def on_poll_player_state(self, data: PlayContentEvent) -> None:
         LOGGER.debug("On Poll PlayerState: %s", data)
         self._available = True
 
@@ -123,41 +122,43 @@ def on_poll_player_state(self, data: dict) -> None:
 
         self._last_poll_player_state = data
 
-    def _on_poll_player_state_changed(self, data: dict) -> None:
+    def _on_poll_player_state_changed(self, data: PlayContentEvent) -> None:
+        data_keys = data.set_keys()
+
         # deep sleep ?
-        if {"playBackState"} == set(data.keys()):
-            if data["playBackState"] == "0":
+        if {"play_back_state"} == set(data_keys):
+            if data.play_back_state == 0:
                 self.state = State.OFF
                 self._clear_non_state_attributes()
             return
 
         # tv running
         if {
-            "chanKey",
+            "chan_key",
             "duration",
-            "mediaCode",
-            "mediaType",
-            "playBackState",
-            "playPostion",
-        } <= data.keys():
-            if "fastSpeed" in data and data["fastSpeed"] == "0":
+            "media_code",
+            "media_type",
+            "play_back_state",
+            "play_position",
+        } <= data_keys:
+            if data.fast_speed == 0:
                 self.state = State.PAUSED
             else:
                 if self.state != State.PLAYING:
                     self.state = State.PLAYING
 
-            self.chan_key = int(data["chanKey"])
-            self.duration = int(data["duration"])
-            self.position = int(data["playPostion"])
+            self.chan_key = data.chan_key
+            self.duration = data.duration
+            self.position = data.play_position
             self.position_last_update = dt.datetime.now()
             return
 
         if {
-            "chanKey",
-            "mediaCode",
-            "mediaType",
-            "playBackState",
-        } == set(data.keys()):
+            "chan_key",
+            "media_code",
+            "media_type",
+            "play_back_state",
+        } == set(data_keys):
             if self._last_poll_player_state is not None:
                 # this is an update and NOT the initial poll
                 if not self._ignore_next_poll_event:

custom_components/magentatv/media_player.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import datetime as dt
-import json
 from collections.abc import Mapping
 from datetime import timedelta
 
@@ -29,8 +28,10 @@
 from homeassistant.helpers import entity_platform, instance_id
 from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from pydantic import TypeAdapter
 
 from custom_components.magentatv import async_get_notification_server
+from custom_components.magentatv.api.event_model import EitChangedEvent, PlayContentEvent
 from custom_components.magentatv.api.exceptions import (
     CommunicationException,
     CommunicationTimeoutException,
@@ -152,9 +153,13 @@ def __init__(
     async def _async_on_event(self, changes):
         LOGGER.debug("%s: Event %s", self.entity_id, changes)
         if "STB_playContent" in changes:
-            self._state_machine.on_event_play_content(json.loads(changes["STB_playContent"]))
+            ta = TypeAdapter(PlayContentEvent)
+            parsed = ta.validate_json(changes["STB_playContent"])
+            self._state_machine.on_event_play_content(parsed)
         elif "STB_EitChanged" in changes:
-            self._state_machine.on_event_eit_changed(json.loads(changes["STB_EitChanged"]))
+            ta = TypeAdapter(EitChangedEvent)
+            parsed = ta.validate_json(changes["STB_EitChanged"])
+            self._state_machine.on_event_eit_changed(parsed)
         elif "messageBody" in changes and "X-pairingCheck" in changes["messageBody"]:
             return  # ignore event
         else:
@@ -183,7 +188,10 @@ async def async_update(self) -> None:
             if not self._client.is_paired():
                 await self._client.async_pair()
 
-            self._state_machine.on_poll_player_state(await self._client.async_get_player_state())
+            result = await self._client.async_get_player_state()
+            ta = TypeAdapter(PlayContentEvent)
+            parsed = ta.validate_python(result)
+            self._state_machine.on_poll_player_state(parsed)
         except (PairingTimeoutException, CommunicationTimeoutException, CommunicationException):
             self._state_machine.on_connection_error()
             # raise ex

hacs.json

@@ -1,6 +1,6 @@
 {
   "name": "MagentaTV",
-  "homeassistant": "2024.10.4",
+  "homeassistant": "2025.1.2",
   "render_readme": true,
   "country": "DE"
 }

requirements-dev.txt

@@ -1,5 +1,5 @@
 -r ./requirements.txt
-pytest_homeassistant_custom_component~=0.13.175
+pytest_homeassistant_custom_component~=0.13.203
 pip>=21.0,<24.4
 ruff==0.8.6
 colorlog==6.9.0

requirements.txt

@@ -1,3 +1,3 @@
-homeassistant==2024.10.4
+homeassistant==2025.1.2
 pydantic>=1.10.8
 async-upnp-client>=0.35.0