Skip to content

Commit

Permalink
Use more tuple instead of list / Improve usage of value_lists (#1238)
Browse files Browse the repository at this point in the history
* Add more checks to get value from value_list

* improve usage of value_lists

* Use more tuple instead of list

* Fix some comments
  • Loading branch information
SukramJ authored Oct 6, 2023
1 parent f3eda60 commit a70ac2c
Show file tree
Hide file tree
Showing 33 changed files with 270 additions and 215 deletions.
5 changes: 5 additions & 0 deletions changelog.md
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
# Version 2023.10.5 (2023-10-05)

- Add started property to central
- Rename:
- value_list -> values
- effect_list -> effects
- Add more checks to get/set value from/tp values
- Use more tuple instead of list

# Version 2023.10.4 (2023-10-03)

Expand Down
12 changes: 7 additions & 5 deletions hahomematic/caches/persistent.py
Original file line number Diff line number Diff line change
Expand Up @@ -221,7 +221,7 @@ def _convert_device_descriptions(
def _convert_device_description(
self, interface_id: str, device_description: dict[str, Any]
) -> None:
"""Convert provided list of device descriptions."""
"""Convert provided dict of device descriptions."""
if interface_id not in self._addresses:
self._addresses[interface_id] = {}
if interface_id not in self._device_descriptions:
Expand Down Expand Up @@ -301,9 +301,11 @@ def has_interface_id(self, interface_id: str) -> bool:
"""Return if interface is in paramset_descriptions cache."""
return interface_id in self._raw_paramset_descriptions

def get_paramset_keys(self, interface_id: str, channel_address: str) -> list[str]:
def get_paramset_keys(self, interface_id: str, channel_address: str) -> tuple[str, ...]:
"""Get paramset_keys from paramset descriptions cache."""
return list(self._raw_paramset_descriptions.get(interface_id, {}).get(channel_address, []))
return tuple(
self._raw_paramset_descriptions.get(interface_id, {}).get(channel_address, [])
)

def get_paramset_descriptions(
self, interface_id: str, channel_address: str, paramset_key: str
Expand Down Expand Up @@ -336,7 +338,7 @@ def is_in_multiple_channels(self, channel_address: str, parameter: str) -> bool:
return len(set(channels)) > 1
return False

def get_all_readable_parameters(self) -> list[str]:
def get_all_readable_parameters(self) -> tuple[str, ...]:
"""Return all readable, eventing parameters from VALUES paramset."""
parameters: set[str] = set()
for channels in self._raw_paramset_descriptions.values():
Expand All @@ -346,7 +348,7 @@ def get_all_readable_parameters(self) -> list[str]:
if operations & Operations.READ and operations & Operations.EVENT:
parameters.add(parameter)

return sorted(parameters)
return tuple(sorted(parameters))

def get_channel_addresses_by_paramset_key(
self, interface_id: str, device_address: str
Expand Down
4 changes: 2 additions & 2 deletions hahomematic/caches/visibility.py
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,7 @@ def parameter_is_ignored(
).get(channel_no, {}).get(ParamsetKey.MASTER, []):
return False # pragma: no cover

dt_short = list(
dt_short = tuple(
filter(
device_type_l.startswith,
self._un_ignore_parameters_by_device_paramset_key,
Expand Down Expand Up @@ -453,7 +453,7 @@ def parameter_is_un_ignored(
Additionally to _parameter_is_un_ignored these parameters
from _RELEVANT_MASTER_PARAMSETS_BY_DEVICE are unignored.
"""
dt_short = list(
dt_short = tuple(
filter(
device_type.lower().startswith,
self._un_ignore_parameters_by_device_paramset_key,
Expand Down
46 changes: 24 additions & 22 deletions hahomematic/central/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -205,14 +205,14 @@ def _has_active_threads(self) -> bool:
return False

@property
def interface_ids(self) -> list[str]:
def interface_ids(self) -> tuple[str, ...]:
"""Return all associated interface ids."""
return list(self._clients)
return tuple(self._clients)

@property
def interfaces(self) -> list[str]:
def interfaces(self) -> tuple[str, ...]:
"""Return all associated interfaces."""
return [client.interface for client in self._clients.values()]
return tuple(client.interface for client in self._clients.values())

@property
def is_alive(self) -> bool:
Expand Down Expand Up @@ -394,7 +394,7 @@ async def _create_clients(self) -> bool:
)
return False

local_ip = await self._identify_callback_ip(list(self.config.interface_configs)[0].port)
local_ip = await self._identify_callback_ip(tuple(self.config.interface_configs)[0].port)
for interface_config in self.config.interface_configs:
try:
if client := await hmcl.create_client(
Expand Down Expand Up @@ -530,7 +530,7 @@ async def validate_config_and_get_system_information(self) -> SystemInformation:
raise NoClients("validate_config: No clients defined.")

local_ip = await self._identify_callback_ip(
list(self.config.interface_configs)[0].port
tuple(self.config.interface_configs)[0].port
)
system_information = SystemInformation()
for interface_config in self.config.interface_configs:
Expand Down Expand Up @@ -561,34 +561,34 @@ def get_device(self, address: str) -> HmDevice | None:

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

return [
return tuple(
be
for be in self._entities.values()
if (
be.unique_identifier not in existing_unique_ids
and be.usage != EntityUsage.NO_CREATE
and be.platform == platform
)
]
)

def get_readable_generic_entities(
self, paramset_key: str | None = None
) -> list[GenericEntity]:
"""Return a list of readable generic entities."""
return [
) -> tuple[GenericEntity, ...]:
"""Return the readable generic entities."""
return tuple(
ge
for ge in self._entities.values()
if (
isinstance(ge, GenericEntity)
and ge.is_readable
and ((paramset_key and ge.paramset_key == paramset_key) or paramset_key is None)
)
]
)

def _get_primary_client(self) -> hmcl.Client | None:
"""Return the client by interface_id or the first with a virtual remote."""
Expand All @@ -603,24 +603,26 @@ def _get_primary_client(self) -> hmcl.Client | None:

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

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

def get_virtual_remotes(self) -> list[HmDevice]:
def get_virtual_remotes(self) -> tuple[HmDevice, ...]:
"""Get the virtual remote for the Client."""
return [
return tuple(
cl.get_virtual_remote() # type: ignore[misc]
for cl in self._clients.values()
if cl.get_virtual_remote() is not None
]
)

def has_client(self, interface_id: str) -> bool:
"""Check if client exists in central."""
Expand Down Expand Up @@ -739,15 +741,15 @@ async def delete_devices(self, interface_id: str, addresses: list[str]) -> None:

@callback_system_event(system_event=SystemEvent.NEW_DEVICES)
async def add_new_devices(
self, interface_id: str, device_descriptions: list[dict[str, Any]]
self, interface_id: str, device_descriptions: tuple[dict[str, Any], ...]
) -> None:
"""Add new devices to central unit."""
await self._add_new_devices(
interface_id=interface_id, device_descriptions=device_descriptions
)

async def _add_new_devices(
self, interface_id: str, device_descriptions: list[dict[str, Any]]
self, interface_id: str, device_descriptions: tuple[dict[str, Any], ...]
) -> None:
"""Add new devices to central unit."""
_LOGGER.debug(
Expand Down
34 changes: 20 additions & 14 deletions hahomematic/client/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -266,11 +266,13 @@ async def get_system_variable(self, name: str) -> str:
"""Get single system variable from CCU / Homegear."""

@abstractmethod
async def get_all_system_variables(self, include_internal: bool) -> list[SystemVariableData]:
async def get_all_system_variables(
self, include_internal: bool
) -> tuple[SystemVariableData, ...]:
"""Get all system variables from CCU / Homegear."""

@abstractmethod
async def get_all_programs(self, include_internal: bool) -> list[ProgramData]:
async def get_all_programs(self, include_internal: bool) -> tuple[ProgramData, ...]:
"""Get all programs, if available."""

@abstractmethod
Expand All @@ -294,21 +296,21 @@ def get_virtual_remote(self) -> HmDevice | None:
return None

@measure_execution_time
async def get_all_device_descriptions(self) -> list[dict[str, Any]] | None:
async def get_all_device_descriptions(self) -> tuple[dict[str, Any]] | None:
"""Get device descriptions from CCU / Homegear."""
try:
return await self._proxy.listDevices() # type: ignore[no-any-return]
return tuple(await self._proxy.listDevices()) # type: ignore[return-value]
except BaseHomematicException as ex:
_LOGGER.warning(
"GET_ALL_DEVICE_DESCRIPTIONS failed: %s [%s]", ex.name, reduce_args(args=ex.args)
)
return None

async def get_device_descriptions(self, device_address: str) -> list[dict[str, Any]] | None:
async def get_device_descriptions(self, device_address: str) -> tuple[dict[str, Any]] | None:
"""Get device descriptions from CCU / Homegear."""
try:
if device_descriptions := await self._proxy_read.getDeviceDescription(device_address):
return [device_descriptions]
return (device_descriptions,)
except BaseHomematicException as ex:
_LOGGER.warning(
"GET_DEVICE_DESCRIPTIONS failed: %s [%s]", ex.name, reduce_args(args=ex.args)
Expand Down Expand Up @@ -568,7 +570,7 @@ async def _get_paramset_description(
return None

async def get_all_paramset_descriptions(
self, device_descriptions: list[dict[str, Any]]
self, device_descriptions: tuple[dict[str, Any], ...]
) -> dict[str, dict[str, Any]]:
"""Get all paramset descriptions for provided device descriptions."""
all_paramsets: dict[str, dict[str, Any]] = {}
Expand Down Expand Up @@ -731,13 +733,15 @@ async def get_system_variable(self, name: str) -> Any:
"""Get single system variable from CCU / Homegear."""
return await self._json_rpc_client.get_system_variable(name=name)

async def get_all_system_variables(self, include_internal: bool) -> list[SystemVariableData]:
async def get_all_system_variables(
self, include_internal: bool
) -> tuple[SystemVariableData, ...]:
"""Get all system variables from CCU / Homegear."""
return await self._json_rpc_client.get_all_system_variables(
include_internal=include_internal
)

async def get_all_programs(self, include_internal: bool) -> list[ProgramData]:
async def get_all_programs(self, include_internal: bool) -> tuple[ProgramData, ...]:
"""Get all programs, if available."""
return await self._json_rpc_client.get_all_programs(include_internal=include_internal)

Expand Down Expand Up @@ -864,7 +868,9 @@ async def get_system_variable(self, name: str) -> Any:
"GET_SYSTEM_VARIABLE failed: %s [%s]", ex.name, reduce_args(args=ex.args)
)

async def get_all_system_variables(self, include_internal: bool) -> list[SystemVariableData]:
async def get_all_system_variables(
self, include_internal: bool
) -> tuple[SystemVariableData, ...]:
"""Get all system variables from CCU / Homegear."""
variables: list[SystemVariableData] = []
try:
Expand All @@ -875,11 +881,11 @@ async def get_all_system_variables(self, include_internal: bool) -> list[SystemV
_LOGGER.warning(
"GET_ALL_SYSTEM_VARIABLES failed: %s [%s]", ex.name, reduce_args(args=ex.args)
)
return variables
return tuple(variables)

async def get_all_programs(self, include_internal: bool) -> list[ProgramData]:
async def get_all_programs(self, include_internal: bool) -> tuple[ProgramData, ...]:
"""Get all programs, if available."""
return []
return ()

async def get_all_rooms(self) -> dict[str, set[str]]:
"""Get all rooms from Homegear."""
Expand All @@ -892,7 +898,7 @@ async def get_all_functions(self) -> dict[str, set[str]]:
async def _get_system_information(self) -> SystemInformation:
"""Get system information of the backend."""
return SystemInformation(
available_interfaces=[InterfaceName.BIDCOS_RF], serial=HOMEGEAR_SERIAL
available_interfaces=(InterfaceName.BIDCOS_RF,), serial=HOMEGEAR_SERIAL
)


Expand Down
Loading

0 comments on commit a70ac2c

Please sign in to comment.