Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

hass-2024.1.x-compat #95

Merged
merged 6 commits into from
Dec 29, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 34 additions & 45 deletions custom_components/hdhomerun/binary_sensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,31 +41,15 @@

# region #-- binary sensor descriptions --#
@dataclasses.dataclass
class OptionalHDHomerunBinarySensorDescription:
"""Represent the required attributes of the binary_sensor description."""
class AdditionalBinarySensorDescription:
"""Represent additional options for the binary sensor entity."""

extra_attributes: Callable | None = None
state_value: Callable[[Any], bool] | None = None


@dataclasses.dataclass
class RequiredHDHomerunBinarySensorDescription:
"""Represent the required attributes of the sensor description."""


@dataclasses.dataclass
class HDHomerunBinarySensorEntityDescription(
OptionalHDHomerunBinarySensorDescription,
BinarySensorEntityDescription,
RequiredHDHomerunBinarySensorDescription,
):
"""Describes binary_sensor entity."""


# endregion

BINARY_SENSORS: tuple[HDHomerunBinarySensorEntityDescription, ...] = ()


async def async_setup_entry(
hass: HomeAssistant,
Expand All @@ -76,27 +60,22 @@ async def async_setup_entry(
coordinator: DataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id][
CONF_DATA_COORDINATOR_GENERAL
]
sensors = [
HDHomerunBinarySensor(
config_entry=config_entry,
coordinator=coordinator,
description=description,
)
for description in BINARY_SENSORS
]
sensors: list[HDHomerunBinarySensor | HDHomeRunRecurringBinarySensor] = []

if coordinator.data.channel_sources:
sensors.append(
HDHomeRunRecurringBinarySensor(
additional_description=AdditionalBinarySensorDescription(
extra_attributes=lambda r: {
"progress": r,
},
),
config_entry=config_entry,
coordinator=hass.data[DOMAIN][config_entry.entry_id][
CONF_DATA_COORDINATOR_GENERAL
],
description=HDHomerunBinarySensorEntityDescription(
description=BinarySensorEntityDescription(
device_class=BinarySensorDeviceClass.RUNNING,
extra_attributes=lambda r: {
"progress": r,
},
key="channel_scanning",
name="Channel Scanning",
translation_key="channel_scanning",
Expand All @@ -112,15 +91,17 @@ async def async_setup_entry(
if UPDATE_DOMAIN is None:
sensors.append(
HDHomerunBinarySensor(
additional_description=AdditionalBinarySensorDescription(
state_value=lambda d: bool(d.latest_firmware),
),
config_entry=config_entry,
coordinator=hass.data[DOMAIN][config_entry.entry_id][
CONF_DATA_COORDINATOR_GENERAL
],
description=HDHomerunBinarySensorEntityDescription(
description=BinarySensorEntityDescription(
key="",
name="Update available",
device_class=BinarySensorDeviceClass.UPDATE,
state_value=lambda d: bool(d.latest_firmware),
),
)
)
Expand All @@ -138,7 +119,7 @@ async def async_setup_entry(
coordinator=hass.data[DOMAIN][config_entry.entry_id][
CONF_DATA_COORDINATOR_GENERAL
],
description=HDHomerunBinarySensorEntityDescription(
description=BinarySensorEntityDescription(
key="",
name="Update available",
),
Expand All @@ -153,15 +134,17 @@ async def async_setup_entry(
class HDHomerunBinarySensor(HDHomerunEntity, BinarySensorEntity):
"""Representation of a binary sensor."""

entity_description: HDHomerunBinarySensorEntityDescription

def __init__(
self,
config_entry: ConfigEntry,
coordinator: DataUpdateCoordinator,
description: HDHomerunBinarySensorEntityDescription,
description: BinarySensorEntityDescription,
additional_description: AdditionalBinarySensorDescription | None = None,
) -> None:
"""Initialise."""
self._additional_description: AdditionalBinarySensorDescription | None = (
additional_description
)
self._attr_entity_category = EntityCategory.DIAGNOSTIC
self.entity_domain = ENTITY_DOMAIN
super().__init__(
Expand All @@ -174,8 +157,8 @@ def __init__(
def is_on(self) -> bool:
"""Return if the service is on."""
if self.coordinator.data:
if self.entity_description.state_value:
return self.entity_description.state_value(self.coordinator.data)
if self._additional_description.state_value:
return self._additional_description.state_value(self.coordinator.data)

return getattr(self.coordinator.data, self.entity_description.key, None)

Expand All @@ -185,21 +168,23 @@ def is_on(self) -> bool:
class HDHomeRunRecurringBinarySensor(HDHomerunEntity, BinarySensorEntity):
"""Representation of a binary sensor that may need out of band updates."""

entity_description: HDHomerunBinarySensorEntityDescription

def __init__(
self,
coordinator: DataUpdateCoordinator,
config_entry: ConfigEntry,
description: HDHomerunBinarySensorEntityDescription,
description: BinarySensorEntityDescription,
recurrence_interval: int,
recurrence_trigger: str,
state_method: str,
state_processor: Callable[..., bool],
recurrence_post_signal: str | None = None,
additional_description: AdditionalBinarySensorDescription | None = None,
) -> None:
"""Initialise."""
self.entity_domain = ENTITY_DOMAIN
self._additional_description: AdditionalBinarySensorDescription | None = (
additional_description
)
self._attr_entity_category = EntityCategory.DIAGNOSTIC

self._state: bool | None = None
Expand Down Expand Up @@ -254,8 +239,10 @@ async def _async_action(self, _: datetime | None = None) -> None:
raise RuntimeError("State processor is not callable") from None

state_method_results: Any = await state_method()
if isinstance(self.entity_description.extra_attributes, Callable):
self._esa = self.entity_description.extra_attributes(state_method_results)
if isinstance(self._additional_description.extra_attributes, Callable):
self._esa = self._additional_description.extra_attributes(
state_method_results
)
temp_state: bool = self._state_processor(state_method_results)
if temp_state:
if self._remove_action_interval is None:
Expand Down Expand Up @@ -297,8 +284,10 @@ def is_on(self) -> bool | None:
"""Get the state of the binary sensor."""
queried_state: bool
ret: bool
if self.entity_description.state_value:
queried_state = self.entity_description.state_value(self.coordinator.data)
if self._additional_description.state_value:
queried_state = self._additional_description.state_value(
self.coordinator.data
)
else:
queried_state = getattr(
self.coordinator.data, self.entity_description.key, None
Expand Down
62 changes: 28 additions & 34 deletions custom_components/hdhomerun/button.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,28 +38,14 @@


# region #-- button entity descriptions --#
@dataclasses.dataclass
class OptionalButtonDescription:
"""Represent the optional attributes of the button description."""

press_action_arguments: Optional[dict] = dataclasses.field(default_factory=dict)


@dataclasses.dataclass
class RequiredButtonDescription:
"""Represent the required attributes of the button description."""
@dataclasses.dataclass(frozen=True)
class AdditionalButtonDescription:
"""Represent additional options for the button entity."""

press_action: str


@dataclasses.dataclass
class HDHomeRunButtonDescription(
OptionalButtonDescription, ButtonEntityDescription, RequiredButtonDescription
):
"""Describes button entity."""

listen_for_signal: str | None = None
listen_for_signal_action: str | None = None
press_action_arguments: Optional[dict] = dataclasses.field(default_factory=dict)


# endregion
Expand All @@ -77,13 +63,15 @@ async def async_setup_entry(

buttons: List[HDHomeRunButton] = [
HDHomeRunButton(
additional_description=AdditionalButtonDescription(
press_action="async_restart",
),
config_entry=config_entry,
coordinator=coordinator,
description=HDHomeRunButtonDescription(
description=ButtonEntityDescription(
device_class=ButtonDeviceClass.RESTART,
key="",
name="Restart",
press_action="async_restart",
translation_key="restart",
),
)
Expand All @@ -92,19 +80,21 @@ async def async_setup_entry(
if coordinator.data.channel_sources:
buttons.append(
HDHomeRunButton(
config_entry=config_entry,
coordinator=coordinator,
description=HDHomeRunButtonDescription(
icon="mdi:text-search",
key="",
additional_description=AdditionalButtonDescription(
listen_for_signal=SIGNAL_HDHOMERUN_CHANNEL_SOURCE_CHANGE,
listen_for_signal_action="_set_channel_source",
name="Channel Scan",
press_action="async_channel_scan_start",
press_action_arguments={
"signal": SIGNAL_HDHOMERUN_CHANNEL_SCANNING_STARTED,
"channel_source": lambda s: getattr(s, "_channel_source", None),
},
),
config_entry=config_entry,
coordinator=coordinator,
description=ButtonEntityDescription(
icon="mdi:text-search",
key="",
name="Channel Scan",
translation_key="channel_scan",
),
)
Expand Down Expand Up @@ -141,16 +131,18 @@ async def _async_button_pressed(
class HDHomeRunButton(HDHomerunEntity, ButtonEntity, ABC):
"""Representation for a button in the Mesh."""

entity_description: HDHomeRunButtonDescription

def __init__(
self,
coordinator: DataUpdateCoordinator,
config_entry: ConfigEntry,
description: HDHomeRunButtonDescription,
description: ButtonEntityDescription,
additional_description: AdditionalButtonDescription | None = None,
) -> None:
"""Initialise."""
self.entity_domain = ENTITY_DOMAIN
self._additional_description: AdditionalButtonDescription | None = (
additional_description
)
super().__init__(
config_entry=config_entry,
coordinator=coordinator,
Expand All @@ -163,13 +155,15 @@ def _set_channel_source(self, channel_source) -> None:

async def async_added_to_hass(self) -> None:
"""Carry out tasks when added to the regstry."""
if self.entity_description.listen_for_signal:
if self._additional_description.listen_for_signal:
self.async_on_remove(
async_dispatcher_connect(
hass=self.hass,
signal=self.entity_description.listen_for_signal,
signal=self._additional_description.listen_for_signal,
target=getattr(
self, self.entity_description.listen_for_signal_action, None
self,
self._additional_description.listen_for_signal_action,
None,
),
)
)
Expand All @@ -179,8 +173,8 @@ async def async_added_to_hass(self) -> None:
async def async_press(self) -> None:
"""Handle the button being pressed."""
await _async_button_pressed(
action=self.entity_description.press_action,
action_arguments=self.entity_description.press_action_arguments.copy(),
action=self._additional_description.press_action,
action_arguments=self._additional_description.press_action_arguments.copy(),
device=self.coordinator.data,
hass=self.hass,
self=self,
Expand Down
37 changes: 14 additions & 23 deletions custom_components/hdhomerun/select.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,30 +27,14 @@

# region #-- select entity descriptions --#
@dataclasses.dataclass
class OptionalHDHomeRunSelectDescription:
"""Represent the optional attributes of the select description."""
class AdditionalSelectDescription:
"""Represent additional options for the select entity."""

extra_attributes_args: dict | None = dataclasses.field(default_factory=dict)
extra_attributes: Callable[[Any], dict] | None = None
custom_options: Callable[[Any], list[str]] | list[str] = dataclasses.field(
default_factory=list
)


@dataclasses.dataclass
class RequiredHDHomeRunSelectDescription:
"""Represent the required attributes of the select description."""


@dataclasses.dataclass
class HDHomeRunSelectDescription(
OptionalHDHomeRunSelectDescription,
SelectEntityDescription,
RequiredHDHomeRunSelectDescription,
):
"""Describes select entity."""


# endregion


Expand All @@ -71,7 +55,7 @@ async def async_setup_entry(
HDHomeRunSelect(
config_entry=config_entry,
coordinator=coordinator,
description=HDHomeRunSelectDescription(
description=SelectEntityDescription(
entity_category=EntityCategory.CONFIG,
key="channel_sources",
name="Channel Sources",
Expand All @@ -90,9 +74,13 @@ def __init__(
self,
coordinator: DataUpdateCoordinator,
config_entry: ConfigEntry,
description: HDHomeRunSelectDescription,
description: SelectEntityDescription,
additional_description: AdditionalSelectDescription | None = None,
) -> None:
"""Initialise."""
self._additional_description: AdditionalSelectDescription | None = (
additional_description
)
self._attr_current_option = None
self.entity_domain = ENTITY_DOMAIN

Expand All @@ -118,7 +106,10 @@ def options(self) -> List[str] | None:
if self.entity_description.key:
return getattr(self.coordinator.data, self.entity_description.key, None)

if isinstance(self.entity_description.custom_options, Callable):
return self.entity_description.custom_options(self.coordinator.data)
if isinstance(self._additional_description.custom_options, Callable):
return self._additional_description.custom_options(self.coordinator.data)

return self.entity_description.custom_options or self.entity_description.options
return (
self._additional_description.custom_options
or self.entity_description.options
)
Loading