-
-
Notifications
You must be signed in to change notification settings - Fork 20
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
2024 codestyle and modernization (#88)
* Update constants * Add host to gateway device * Add host to gateway device 2 * Add typing to device_tracker * Move coordinator to a new file * Add more typing * Add MAC address * Better type entry data * Bugfix * Add reboot button as entity
- Loading branch information
Showing
5 changed files
with
199 additions
and
165 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,53 @@ | ||
"""Support for Sagencom F@st buttons.""" | ||
from __future__ import annotations | ||
|
||
from homeassistant.components.button import ButtonDeviceClass, ButtonEntity | ||
from homeassistant.config_entries import ConfigEntry | ||
from homeassistant.const import EntityCategory | ||
from homeassistant.core import HomeAssistant | ||
from homeassistant.helpers.device_registry import DeviceInfo | ||
from homeassistant.helpers.entity_platform import AddEntitiesCallback | ||
from sagemcom_api.client import SagemcomClient | ||
from sagemcom_api.models import DeviceInfo as GatewayDeviceInfo | ||
|
||
from . import HomeAssistantSagemcomFastData | ||
from .const import DOMAIN | ||
|
||
|
||
async def async_setup_entry( | ||
hass: HomeAssistant, | ||
entry: ConfigEntry, | ||
async_add_entities: AddEntitiesCallback, | ||
) -> None: | ||
"""Set up the Sagemcom F@st button from a config entry.""" | ||
data: HomeAssistantSagemcomFastData = hass.data[DOMAIN][entry.entry_id] | ||
entities: list[ButtonEntity] = [] | ||
entities.append(SagemcomFastRebootButton(data.gateway, data.coordinator.client)) | ||
|
||
async_add_entities(entities) | ||
|
||
|
||
class SagemcomFastRebootButton(ButtonEntity): | ||
"""Representation of an Sagemcom F@st Button.""" | ||
|
||
_attr_has_entity_name = True | ||
_attr_name = "Reboot" | ||
_attr_device_class = ButtonDeviceClass.RESTART | ||
_attr_entity_category = EntityCategory.CONFIG | ||
|
||
def __init__(self, gateway: GatewayDeviceInfo, client: SagemcomClient) -> None: | ||
"""Initialize the button.""" | ||
self.gateway = gateway | ||
self.client = client | ||
self._attr_unique_id = f"{self.gateway.serial_number}_reboot" | ||
|
||
async def async_press(self) -> None: | ||
"""Handle the button press.""" | ||
await self.client.reboot() | ||
|
||
@property | ||
def device_info(self) -> DeviceInfo: | ||
"""Return device registry information for this entity.""" | ||
return DeviceInfo( | ||
identifiers={(DOMAIN, self.gateway.serial_number)}, | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,14 +1,25 @@ | ||
"""Constants for the Sagemcom integration.""" | ||
DOMAIN = "sagemcom_fast" | ||
"""Constants for the Sagemcom F@st integration.""" | ||
from __future__ import annotations | ||
|
||
CONF_ENCRYPTION_METHOD = "encryption_method" | ||
CONF_TRACK_WIRELESS_CLIENTS = "track_wireless_clients" | ||
CONF_TRACK_WIRED_CLIENTS = "track_wired_clients" | ||
import logging | ||
from typing import Final | ||
|
||
DEFAULT_TRACK_WIRELESS_CLIENTS = True | ||
DEFAULT_TRACK_WIRED_CLIENTS = True | ||
from homeassistant.const import Platform | ||
|
||
ATTR_MANUFACTURER = "Sagemcom" | ||
LOGGER: logging.Logger = logging.getLogger(__package__) | ||
|
||
MIN_SCAN_INTERVAL = 10 | ||
DEFAULT_SCAN_INTERVAL = 10 | ||
DOMAIN: Final = "sagemcom_fast" | ||
|
||
CONF_ENCRYPTION_METHOD: Final = "encryption_method" | ||
CONF_TRACK_WIRELESS_CLIENTS: Final = "track_wireless_clients" | ||
CONF_TRACK_WIRED_CLIENTS: Final = "track_wired_clients" | ||
|
||
DEFAULT_TRACK_WIRELESS_CLIENTS: Final = True | ||
DEFAULT_TRACK_WIRED_CLIENTS: Final = True | ||
|
||
ATTR_MANUFACTURER: Final = "Sagemcom" | ||
|
||
MIN_SCAN_INTERVAL: Final = 10 | ||
DEFAULT_SCAN_INTERVAL: Final = 10 | ||
|
||
PLATFORMS: list[Platform] = [Platform.DEVICE_TRACKER, Platform.BUTTON] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,56 @@ | ||
"""Helpers to help coordinate updates.""" | ||
from __future__ import annotations | ||
|
||
from datetime import timedelta | ||
import logging | ||
|
||
import async_timeout | ||
from homeassistant.core import HomeAssistant | ||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed | ||
from sagemcom_api.client import SagemcomClient | ||
from sagemcom_api.models import Device | ||
|
||
|
||
class SagemcomDataUpdateCoordinator(DataUpdateCoordinator): | ||
"""Class to manage fetching Sagemcom data.""" | ||
|
||
def __init__( | ||
self, | ||
hass: HomeAssistant, | ||
logger: logging.Logger, | ||
*, | ||
name: str, | ||
client: SagemcomClient, | ||
update_interval: timedelta | None = None, | ||
): | ||
"""Initialize update coordinator.""" | ||
super().__init__( | ||
hass, | ||
logger, | ||
name=name, | ||
update_interval=update_interval, | ||
) | ||
self.data = {} | ||
self.hosts: dict[str, Device] = {} | ||
self.client = client | ||
|
||
async def _async_update_data(self) -> dict[str, Device]: | ||
"""Update hosts data.""" | ||
try: | ||
async with async_timeout.timeout(10): | ||
try: | ||
await self.client.login() | ||
hosts = await self.client.get_hosts(only_active=True) | ||
finally: | ||
await self.client.logout() | ||
|
||
"""Mark all device as non-active.""" | ||
for idx, host in self.hosts.items(): | ||
host.active = False | ||
self.hosts[idx] = host | ||
for host in hosts: | ||
self.hosts[host.id] = host | ||
|
||
return self.hosts | ||
except Exception as exception: | ||
raise UpdateFailed(f"Error communicating with API: {exception}") |
Oops, something went wrong.