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

Cleanup code #1239

Merged
merged 5 commits into from
Oct 6, 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
1 change: 1 addition & 0 deletions changelog.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
- effect_list -> effects
- Add more checks to get/set value from/tp values
- Use more tuple instead of list
- Cleanup code

# Version 2023.10.4 (2023-10-03)

Expand Down
88 changes: 54 additions & 34 deletions hahomematic/central/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,9 +146,9 @@ def __init__(self, central_config: CentralConfig) -> None:
# {device_address, device}
self._devices: Final[dict[str, HmDevice]] = {}
# {sysvar_name, sysvar_entity}
self.sysvar_entities: Final[dict[str, GenericSystemVariable]] = {}
self._sysvar_entities: Final[dict[str, GenericSystemVariable]] = {}
# {sysvar_name, program_button}U
self.program_entities: Final[dict[str, HmProgramButton]] = {}
self._program_buttons: Final[dict[str, HmProgramButton]] = {}
# store last event received datetime by interface
self.last_events: Final[dict[str, datetime]] = {}
# Signature: (name, *args)
Expand Down Expand Up @@ -259,6 +259,37 @@ def system_information(self) -> SystemInformation:
return client.system_information
return SystemInformation()

@property
def sysvar_entities(self) -> tuple[GenericSystemVariable, ...]:
"""Return the sysvar entities."""
return tuple(self._sysvar_entities.values())

def add_sysvar_entity(self, sysvar_entity: GenericSystemVariable) -> None:
"""Add new program button."""
if (name := sysvar_entity.name) is not None:
self._sysvar_entities[name] = sysvar_entity

def remove_sysvar_entity(self, name: str) -> None:
"""Remove a sysvar entity."""
if (sysvar_entity := self.get_sysvar_entity(name=name)) is not None:
sysvar_entity.remove_entity()
del self._sysvar_entities[name]

@property
def program_buttons(self) -> tuple[HmProgramButton, ...]:
"""Return the program entities."""
return tuple(self._program_buttons.values())

def add_program_button(self, program_button: HmProgramButton) -> None:
"""Add new program button."""
self._program_buttons[program_button.pid] = program_button

def remove_program_button(self, pid: str) -> None:
"""Remove a program button."""
if (program_button := self.get_program_button(pid=pid)) is not None:
program_button.remove_entity()
del self._program_buttons[pid]

@property
def version(self) -> str | None:
"""Return the version of the backend."""
Expand Down Expand Up @@ -560,11 +591,11 @@ def get_device(self, address: str) -> HmDevice | None:
return self._devices.get(d_address)

def get_entities_by_platform(
self, platform: HmPlatform, existing_unique_ids: list[str] | None = None
self, platform: HmPlatform, existing_unique_ids: tuple[str, ...] | None = None
) -> tuple[BaseEntity, ...]:
"""Return all entities by platform."""
if not existing_unique_ids:
existing_unique_ids = []
existing_unique_ids = ()

return tuple(
be
Expand Down Expand Up @@ -602,17 +633,15 @@ def _get_primary_client(self) -> hmcl.Client | None:
return client

def get_hub_entities_by_platform(
self, platform: HmPlatform, existing_unique_ids: list[str] | None = None
self, platform: HmPlatform, existing_unique_ids: tuple[str, ...] | None = None
) -> tuple[GenericHubEntity, ...]:
"""Return the hub entities by platform."""
if not existing_unique_ids:
existing_unique_ids = []
existing_unique_ids = ()

return tuple(
he
for he in (
tuple(self.program_entities.values()) + tuple(self.sysvar_entities.values())
)
for he in (self.program_buttons + self.sysvar_entities)
if (he.unique_identifier not in existing_unique_ids and he.platform == platform)
)

Expand Down Expand Up @@ -655,7 +684,7 @@ async def _create_devices(self) -> None:
_LOGGER.debug("CREATE_DEVICES: Starting to create devices for %s", self._name)

new_devices = set[HmDevice]()
for interface_id in self._clients:
for interface_id in self.interface_ids:
if not self.paramset_descriptions.has_interface_id(interface_id=interface_id):
_LOGGER.debug(
"CREATE_DEVICES: Skipping interface %s, missing paramsets",
Expand Down Expand Up @@ -715,20 +744,13 @@ async def delete_device(self, interface_id: str, device_address: str) -> None:

if (device := self._devices.get(device_address)) is None:
return
addresses: list[str] = list(device.channels.keys())
addresses.append(device_address)
if len(addresses) == 0:
_LOGGER.debug(
"DELETE_DEVICE: Nothing to delete: interface_id = %s, device_address = %s",
interface_id,
device_address,
)
return
addresses = device.channel_addresses
addresses += (device_address,)

await self.delete_devices(interface_id=interface_id, addresses=addresses)

@callback_system_event(system_event=SystemEvent.DELETE_DEVICES)
async def delete_devices(self, interface_id: str, addresses: list[str]) -> None:
async def delete_devices(self, interface_id: str, addresses: tuple[str, ...]) -> None:
"""Delete devices from central."""
_LOGGER.debug(
"DELETE_DEVICES: interface_id = %s, addresses = %s",
Expand Down Expand Up @@ -766,11 +788,11 @@ async def _add_new_devices(
return

async with self._sema_add_devices:
# We need this list to avoid adding duplicates.
known_addresses = [
# We need this to avoid adding duplicates.
known_addresses = tuple(
dev_desc[Description.ADDRESS]
for dev_desc in self.device_descriptions.get_raw_device_descriptions(interface_id)
]
)
client = self._clients[interface_id]
for dev_desc in device_descriptions:
try:
Expand Down Expand Up @@ -1010,7 +1032,7 @@ async def get_system_variable(self, name: str) -> Any | None:

async def set_system_variable(self, name: str, value: Any) -> None:
"""Set variable value on CCU/Homegear."""
if entity := self.sysvar_entities.get(name):
if entity := self.get_sysvar_entity(name=name):
await entity.send_variable(value=value)
else:
_LOGGER.warning("Variable %s not found on %s", name, self.name)
Expand Down Expand Up @@ -1046,36 +1068,34 @@ def _get_virtual_remote(self, device_address: str) -> HmDevice | None:
def get_generic_entity(self, channel_address: str, parameter: str) -> GenericEntity | None:
"""Get entity by channel_address and parameter."""
if device := self.get_device(address=channel_address):
return device.generic_entities.get((channel_address, parameter))
return device.get_generic_entity(channel_address=channel_address, parameter=parameter)
return None

def get_wrapper_entity(self, channel_address: str, parameter: str) -> WrapperEntity | None:
"""Return the hm wrapper_entity."""
if device := self.get_device(address=channel_address):
return device.wrapper_entities.get((channel_address, parameter))
return device.get_wrapper_entity(channel_address=channel_address, parameter=parameter)
return None

def get_event(self, channel_address: str, parameter: str) -> GenericEvent | None:
"""Return the hm event."""
if device := self.get_device(address=channel_address):
return device.generic_events.get((channel_address, parameter))
return device.get_generic_event(channel_address=channel_address, parameter=parameter)
return None

def get_custom_entity(self, address: str, channel_no: int | None) -> CustomEntity | None:
def get_custom_entity(self, address: str, channel_no: int) -> CustomEntity | None:
"""Return the hm custom_entity."""
if device := self.get_device(address=address):
for custom_entity in device.custom_entities.values():
if custom_entity.channel_no == channel_no:
return custom_entity
return device.get_custom_entity(channel_no=channel_no)
return None

def get_sysvar_entity(self, name: str) -> GenericSystemVariable | None:
"""Return the sysvar entity."""
return self.sysvar_entities.get(name)
return self._sysvar_entities.get(name)

def get_program_button(self, pid: str) -> HmProgramButton | None:
"""Return the program button."""
return self.program_entities.get(pid)
return self._program_buttons.get(pid)

async def clear_caches(self) -> None:
"""Clear all stored data."""
Expand Down Expand Up @@ -1270,7 +1290,7 @@ def __init__(
username: str,
password: str,
central_id: str,
interface_configs: set[hmcl.InterfaceConfig],
interface_configs: tuple[hmcl.InterfaceConfig, ...],
default_callback_port: int,
client_session: ClientSession | None,
tls: bool = DEFAULT_TLS,
Expand Down
4 changes: 2 additions & 2 deletions hahomematic/central/xml_rpc_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ def newDevices(self, interface_id: str, device_descriptions: list[dict[str, Any]
if central := self._xml_rpc_server.get_central(interface_id):
central.create_task(
central.add_new_devices(
interface_id=interface_id, device_descriptions=device_descriptions
interface_id=interface_id, device_descriptions=tuple(device_descriptions)
),
name="newDevices",
)
Expand All @@ -69,7 +69,7 @@ def deleteDevices(self, interface_id: str, addresses: list[str]) -> None:
central: hmcu.CentralUnit | None
if central := self._xml_rpc_server.get_central(interface_id):
central.create_task(
central.delete_devices(interface_id=interface_id, addresses=addresses),
central.delete_devices(interface_id=interface_id, addresses=tuple(addresses)),
name="deleteDevices",
)

Expand Down
4 changes: 2 additions & 2 deletions hahomematic/platforms/custom/entity.py
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@ def _mark_entity_by_custom_un_ignore_parameters(
if not un_ignore_params_by_paramset_key:
return # pragma: no cover
for paramset_key, un_ignore_params in un_ignore_params_by_paramset_key.items():
for entity in self.device.generic_entities.values():
for entity in self.device.generic_entities:
if entity.paramset_key == paramset_key and entity.parameter in un_ignore_params:
entity.set_usage(EntityUsage.ENTITY)

Expand Down Expand Up @@ -268,7 +268,7 @@ class NoneTypeEntity:
min: Any = None
unit: Any = None
value: Any = None
value_list: list[Any] = []
values: list[Any] = []
visible: Any = None
channel_operation_mode: str | None = None
is_hmtype = False
Expand Down
12 changes: 6 additions & 6 deletions hahomematic/platforms/custom/siren.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,12 +62,12 @@ def is_on(self) -> bool:
@value_property
@abstractmethod
def available_tones(self) -> tuple[str, ...] | None:
"""Return a list of available tones."""
"""Return available tones."""

@value_property
@abstractmethod
def available_lights(self) -> tuple[str, ...] | None:
"""Return a list of available lights."""
"""Return available lights."""

@config_property
@abstractmethod
Expand Down Expand Up @@ -132,12 +132,12 @@ def is_on(self) -> bool:

@value_property
def available_tones(self) -> tuple[str, ...] | None:
"""Return a list of available tones."""
"""Return available tones."""
return self._e_acoustic_alarm_selection.values

@value_property
def available_lights(self) -> tuple[str, ...] | None:
"""Return a list of available lights."""
"""Return available lights."""
return self._e_optical_alarm_selection.values

@config_property
Expand Down Expand Up @@ -216,12 +216,12 @@ def is_on(self) -> bool:

@value_property
def available_tones(self) -> tuple[str, ...] | None:
"""Return a list of available tones."""
"""Return available tones."""
return None

@value_property
def available_lights(self) -> tuple[str, ...] | None:
"""Return a list of available lights."""
"""Return available lights."""
return None

@config_property
Expand Down
Loading