Skip to content

Commit

Permalink
Add new lookup methods for Services and Characteristics
Browse files Browse the repository at this point in the history
- Services.iid_or_none - fetch by service by iid or return None
- Accessories.aid_or_none - fetch accessory by aid or return None
  • Loading branch information
bdraco committed Oct 19, 2023
1 parent c244405 commit 77c0533
Show file tree
Hide file tree
Showing 3 changed files with 85 additions and 35 deletions.
67 changes: 51 additions & 16 deletions aiohomekit/model/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,17 +60,26 @@ class Transport(Enum):


class Services:
"""Represents a list of HomeKit services."""

def __init__(self):
"""Initialize a new list of services."""
self._services: list[Service] = []
self._iid_to_service: dict[int, Service] = {}
self._type_to_service: dict[str, Service] = {}

def __iter__(self) -> Iterator[Service]:
"""Iterate over all services."""
return iter(self._services)

def iid(self, iid: int) -> Service:
"""Return the service with the given iid, raising KeyError if it does not exist."""
return self._iid_to_service[iid]

def iid_or_none(self, iid: int) -> Service | None:
"""Return the service with the given iid or None if it does not exist."""
return self._iid_to_service.get(iid)

def get_char_by_iid(self, iid: int) -> Characteristic | None:
"""Get a characteristic by iid."""
for service in self._services:
Expand All @@ -86,7 +95,8 @@ def filter(
parent_service: Service = None,
child_service: Service = None,
order_by: list[str] | None = None,
) -> Iterable[Service]:
) -> Iterator[Service]:
"""Filter services by type and characteristics."""
matches = iter(self._services)

if service_type:
Expand Down Expand Up @@ -122,7 +132,8 @@ def first(
characteristics: dict[str, str] = None,
parent_service: Service = None,
child_service: Service = None,
) -> Service:
) -> Service | None:
"""Get the first service."""
if (
service_type is not None
and characteristics is None
Expand All @@ -143,22 +154,29 @@ def first(
except StopIteration:
return None

def append(self, service: Service):
def append(self, service: Service) -> None:
"""Add a service to the list of services."""
self._services.append(service)
self._iid_to_service[service.iid] = service
if service.type not in self._type_to_service:
self._type_to_service[service.type] = service


class Characteristics:
"""Represents a list of HomeKit characteristics."""

def __init__(self, services: Services) -> None:
"""Initialize a new list of characteristics."""
self._services = services

def iid(self, iid: int) -> Characteristic | None:
"""Return the characteristic with the given iid, return None if it does not exist."""
return self._services.get_char_by_iid(iid)


class Accessory:
"""Represents a HomeKit accessory."""

def __init__(self) -> None:
"""Initialize a new accessory."""
self.aid = get_id()
Expand Down Expand Up @@ -206,34 +224,41 @@ def accessory_information(self) -> Service:

@property
def name(self) -> str:
"""Return the name of the accessory."""
return self.accessory_information.value(CharacteristicsTypes.NAME, "")

@property
def manufacturer(self) -> str:
"""Return the manufacturer of the accessory."""
return self.accessory_information.value(CharacteristicsTypes.MANUFACTURER, "")

@property
def model(self) -> str | None:
"""Return the model of the accessory."""
return self.accessory_information.value(CharacteristicsTypes.MODEL, "")

@property
def serial_number(self) -> str:
"""Return the serial number of the accessory."""
return self.accessory_information.value(CharacteristicsTypes.SERIAL_NUMBER, "")

@property
def firmware_revision(self) -> str:
"""Return the firmware revision of the accessory."""
return self.accessory_information.value(
CharacteristicsTypes.FIRMWARE_REVISION, ""
)

@property
def hardware_revision(self) -> str:
"""Return the hardware revision of the accessory."""
return self.accessory_information.value(
CharacteristicsTypes.HARDWARE_REVISION, ""
)

@property
def available(self) -> bool:
"""Return True if the accessory is available."""
return all(s.available for s in self.services)

@property
Expand All @@ -251,6 +276,7 @@ def needs_polling(self) -> bool:

@classmethod
def create_from_dict(cls, data: dict[str, Any]) -> Accessory:
"""Create an accessory from a dict."""
accessory = cls()
accessory.aid = data["aid"]

Expand Down Expand Up @@ -303,6 +329,7 @@ def create_from_dict(cls, data: dict[str, Any]) -> Accessory:
return accessory

def get_next_id(self) -> int:
"""Return the next available id for a service."""
self._next_id += 1
return self._next_id

Expand All @@ -313,31 +340,35 @@ def add_service(
add_required: bool = False,
iid: int | None = None,
) -> Service:
"""Add a service to the accessory."""
service = Service(
self, service_type, name=name, add_required=add_required, iid=iid
)
self.services.append(service)
return service

def to_accessory_and_service_list(self):
services_list = []
for s in self.services:
services_list.append(s.to_accessory_and_service_list())
d = {"aid": self.aid, "services": services_list}
return d
def to_accessory_and_service_list(self) -> dict[str, Any]:
"""Serialize the accessory to a dict."""
return {
"aid": self.aid,
"services": [s.to_accessory_and_service_list() for s in self.services],
}


class Accessories:
"""Represents a list of HomeKit accessories."""

accessories: list[Accessory]

def __init__(self) -> None:
"""Initialize a new list of accessories."""
self.accessories = []
self._aid_to_accessory: dict[int, Accessory] = {}

def __iter__(self) -> Iterator[Accessory]:
return iter(self.accessories)

def __getitem__(self, idx) -> Accessory:
def __getitem__(self, idx: int) -> Accessory:
return self.accessories[idx]

@classmethod
Expand All @@ -353,23 +384,27 @@ def from_list(cls, accessories: entity_map.Accesories) -> Accessories:
return self

def add_accessory(self, accessory: Accessory) -> None:
"""Add an accessory to the list of accessories."""
self.accessories.append(accessory)
self._aid_to_accessory[accessory.aid] = accessory

def serialize(self) -> entity_map.Accesories:
accessories_list = []
for a in self.accessories:
accessories_list.append(a.to_accessory_and_service_list())
return accessories_list
"""Serialize the accessories to a list of dicts."""
return [a.to_accessory_and_service_list() for a in self.accessories]

def to_accessory_and_service_list(self) -> dict[str, entity_map.Accesories]:
d = {"accessories": self.serialize()}
return hkjson.dumps(d)
return hkjson.dumps({"accessories": self.serialize()})

def aid(self, aid: int) -> Accessory:
"""Return the accessory with the given aid, raising KeyError if it does not exist."""
return self._aid_to_accessory[aid]

def aid_or_none(self, aid: int) -> Accessory | None:
"""Return the accessory with the given aid or None if it does not exist."""
return self._aid_to_accessory.get(aid)

def has_aid(self, aid: int) -> bool:
"""Return True if the given aid exists."""
return aid in self._aid_to_accessory

def process_changes(self, changes: dict[tuple[int, int], Any]) -> None:
Expand Down
1 change: 0 additions & 1 deletion aiohomekit/model/characteristics/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -267,6 +267,5 @@ class ThreadStatus(enum.IntFlag):


class TemperatureDisplayUnits(enum.IntEnum):

CELSIUS = 0
FAHRENHEIT = 1
52 changes: 34 additions & 18 deletions aiohomekit/model/services/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,39 +29,51 @@


class Characteristics:
"""Represents a collection of characteristics."""

_characteristics: list[Characteristic]

def __init__(self) -> None:
"""Initialise a collection of characteristics."""
self._characteristics = []
self._iid_to_characteristic: dict[int, Characteristic] = {}

def append(self, char: Characteristic) -> None:
"""Add a characteristic."""
self._characteristics.append(char)
self._iid_to_characteristic[char.iid] = char

def get(self, iid: int) -> Characteristic:
"""Get a characteristic by iid."""
return self._iid_to_characteristic.get(iid)

def __iter__(self) -> Iterator[Characteristic]:
"""Iterate over characteristics."""
return iter(self._characteristics)

def filter(self, char_types=None) -> Iterable[Characteristic]:
def filter(
self, char_types: Iterable[str] | None = None
) -> Iterator[Characteristic]:
"""Filter characteristics by type."""
matches = iter(self)

if char_types:
char_types = [normalize_uuid(c) for c in char_types]
char_types = {normalize_uuid(c) for c in char_types}
matches = filter(lambda char: char.type in char_types, matches)

return matches

def first(self, char_types=None) -> Characteristic:
def first(self, char_types: Iterable[str] | None = None) -> Characteristic:
"""Get the first characteristic."""
return next(self.filter(char_types=char_types))


class Service:
"""Represents a service on an accessory."""

type: str
iid: int
linked: list[Service]
linked: set[Service]

characteristics: Characteristics
characteristics_by_type: dict[str, Characteristic]
Expand All @@ -74,14 +86,15 @@ def __init__(
name: str | None = None,
add_required: bool = False,
iid: int | None = None,
):
) -> None:
"""Initialise a service."""
self.type = normalize_uuid(service_type)

self.accessory = accessory
self.iid = iid or accessory.get_next_id()
self.characteristics = Characteristics()
self.characteristics_by_type = {}
self.linked = []
self.characteristics_by_type: dict[str, Characteristic] = {}
self.linked: list[Service] = []

if name:
char = self.add_char(CharacteristicsTypes.NAME)
Expand All @@ -92,11 +105,12 @@ def __init__(
if required not in self.characteristics_by_type:
self.add_char(required)

def has(self, char_type) -> bool:
char_type = normalize_uuid(char_type)
return char_type in self.characteristics_by_type
def has(self, char_type: str) -> bool:
"""Return True if the service has a characteristic."""
return normalize_uuid(char_type) in self.characteristics_by_type

def value(self, char_type, default_value=None) -> Any:
def value(self, char_type: str, default_value: Any | None = None) -> Any:
"""Return the value of a characteristic."""
char_type = normalize_uuid(char_type)

if char_type not in self.characteristics_by_type:
Expand All @@ -105,10 +119,11 @@ def value(self, char_type, default_value=None) -> Any:
return self.characteristics_by_type[char_type].value

def __getitem__(self, key) -> Characteristic:
key = normalize_uuid(key)
return self.characteristics_by_type[key]
"""Get a characteristic by type."""
return self.characteristics_by_type[normalize_uuid(key)]

def add_char(self, char_type: str, **kwargs: Any) -> Characteristic:
"""Add a characteristic to the service."""
char = Characteristic(self, char_type, **kwargs)
self.characteristics.append(char)
self.characteristics_by_type[char.type] = char
Expand All @@ -119,6 +134,7 @@ def get_char_by_iid(self, iid: int) -> Characteristic | None:
return self.characteristics.get(iid)

def add_linked_service(self, service: Service) -> None:
"""Add a linked service."""
self.linked.append(service)

def build_update(self, payload: dict[str, Any]) -> list[tuple[int, int, Any]]:
Expand All @@ -135,22 +151,22 @@ def build_update(self, payload: dict[str, Any]) -> list[tuple[int, int, Any]]:

return result

def to_accessory_and_service_list(self):
def to_accessory_and_service_list(self) -> dict[str, Any]:
"""Return the service as a dictionary."""
characteristics_list = []
for c in self.characteristics:
characteristics_list.append(c.to_accessory_and_service_list())

d = {
"iid": self.iid,
"type": self.type,
"characteristics": characteristics_list,
}

linked = [service.iid for service in self.linked]
if linked:
if linked := [service.iid for service in self.linked]:
d["linked"] = linked

return d

@property
def available(self) -> bool:
"""Return True if all characteristics are available."""
return all(c.available for c in self.characteristics)

0 comments on commit 77c0533

Please sign in to comment.