Skip to content

V4.9.0 #466

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

Merged
merged 5 commits into from
Oct 15, 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
16 changes: 16 additions & 0 deletions .coveragerc
Original file line number Diff line number Diff line change
Expand Up @@ -4,3 +4,19 @@ include = pyhap/*
omit =
tests/*
pyhap/accessories/*

[report]
# Regexes for lines to exclude from consideration
exclude_lines =
# Have to re-enable the standard pragma
pragma: no cover

# Don't complain about missing debug-only code:
def __repr__

# Don't complain if tests don't hit defensive assertion code:
raise AssertionError
raise NotImplementedError

# TYPE_CHECKING and @overload blocks are never executed during pytest run
if TYPE_CHECKING:
5 changes: 5 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,11 @@ Sections
### Developers
-->

## [4.9.0] - 2023-10-15

- Hashing of accessories no longer includes their values, resulting in more reliable syncs between
devices. [#464](https://github.com/ikalchev/HAP-python/pull/464)

## [4.8.0] - 2023-10-06

- Add AccessoryInformation:HardwareFinish and NFCAccess characteristics/services.
Expand Down
78 changes: 50 additions & 28 deletions pyhap/accessory.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
"""Module for the Accessory classes."""
import itertools
import logging
from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Optional
from uuid import UUID

from pyhap import SUPPORT_QR_CODE, util
from pyhap.const import (
from . import SUPPORT_QR_CODE, util
from .const import (
CATEGORY_BRIDGE,
CATEGORY_OTHER,
HAP_PROTOCOL_VERSION,
Expand All @@ -14,14 +15,19 @@
HAP_REPR_VALUE,
STANDALONE_AID,
)
from pyhap.iid_manager import IIDManager
from pyhap.service import Service
from .iid_manager import IIDManager
from .service import Service

if SUPPORT_QR_CODE:
import base36
from pyqrcode import QRCode


if TYPE_CHECKING:
from .accessory_driver import AccessoryDriver
from .characteristic import Characteristic


HAP_PROTOCOL_INFORMATION_SERVICE_UUID = UUID("000000A2-0000-1000-8000-0026BB765291")

logger = logging.getLogger(__name__)
Expand All @@ -35,7 +41,13 @@ class Accessory:

category = CATEGORY_OTHER

def __init__(self, driver, display_name, aid=None, iid_manager=None):
def __init__(
self,
driver: "AccessoryDriver",
display_name: Optional[str],
aid: Optional[int] = None,
iid_manager: Optional[IIDManager] = None,
) -> None:
"""Initialise with the given properties.

:param display_name: Name to be displayed in the Home app.
Expand All @@ -47,24 +59,24 @@ def __init__(self, driver, display_name, aid=None, iid_manager=None):
will assign the standalone AID to this `Accessory`.
:type aid: int
"""
self.aid = aid
self.display_name = display_name
self.aid: Optional[int] = aid
self.display_name: Optional[str] = display_name
self.driver = driver
self.services = []
self.services: List[Service] = []
self.iid_manager = iid_manager or IIDManager()
self.setter_callback = None
self.setter_callback: Optional[Callable[[Any], None]] = None

self.add_info_service()
if aid == STANDALONE_AID:
self.add_protocol_version_service()

def __repr__(self):
def __repr__(self) -> str:
"""Return the representation of the accessory."""
services = [s.display_name for s in self.services]
return f"<accessory display_name='{self.display_name}' services={services}>"

@property
def available(self):
def available(self) -> bool:
"""Accessory is available.

If available is False, get_characteristics will return
Expand All @@ -75,7 +87,7 @@ def available(self):
"""
return True

def add_info_service(self):
def add_info_service(self) -> None:
"""Helper method to add the required `AccessoryInformation` service.

Called in `__init__` to be sure that it is the first service added.
Expand Down Expand Up @@ -116,7 +128,12 @@ def set_info_service(
self.display_name,
)

def add_preload_service(self, service, chars=None, unique_id=None):
def add_preload_service(
self,
service: Service,
chars: Optional[Iterable["Characteristic"]] = None,
unique_id: Optional[str] = None,
) -> Service:
"""Create a service with the given name and add it to this acc."""
service = self.driver.loader.get_service(service)
if unique_id is not None:
Expand All @@ -129,12 +146,12 @@ def add_preload_service(self, service, chars=None, unique_id=None):
self.add_service(service)
return service

def set_primary_service(self, primary_service):
def set_primary_service(self, primary_service: Service) -> None:
"""Set the primary service of the acc."""
for service in self.services:
service.is_primary_service = service.type_id == primary_service.type_id

def add_service(self, *servs):
def add_service(self, *servs: Service) -> None:
"""Add the given services to this Accessory.

This also assigns unique IIDS to the services and their Characteristics.
Expand All @@ -153,7 +170,7 @@ def add_service(self, *servs):
c.broker = self
self.iid_manager.assign(c)

def get_service(self, name):
def get_service(self, name: str) -> Optional[Service]:
"""Return a Service with the given name.

A single Service is returned even if more than one Service with the same name
Expand All @@ -168,7 +185,7 @@ def get_service(self, name):
"""
return next((s for s in self.services if s.display_name == name), None)

def xhm_uri(self):
def xhm_uri(self) -> str:
"""Generates the X-HM:// uri (Setup Code URI)

:rtype: str
Expand All @@ -195,7 +212,7 @@ def xhm_uri(self):

return "X-HM://" + encoded_payload + self.driver.state.setup_id

def get_characteristic(self, aid, iid):
def get_characteristic(self, aid: int, iid: int) -> Optional["Characteristic"]:
"""Get the characteristic for the given IID.

The AID is used to verify if the search is in the correct accessory.
Expand All @@ -205,7 +222,7 @@ def get_characteristic(self, aid, iid):

return self.iid_manager.get_obj(iid)

def to_HAP(self):
def to_HAP(self, include_value: bool = True) -> Dict[str, Any]:
"""A HAP representation of this Accessory.

:return: A HAP representation of this accessory. For example:
Expand All @@ -224,7 +241,7 @@ def to_HAP(self):
"""
return {
HAP_REPR_AID: self.aid,
HAP_REPR_SERVICES: [s.to_HAP() for s in self.services],
HAP_REPR_SERVICES: [s.to_HAP(include_value=include_value) for s in self.services],
}

def setup_message(self):
Expand Down Expand Up @@ -325,13 +342,18 @@ class Bridge(Accessory):

category = CATEGORY_BRIDGE

def __init__(self, driver, display_name, iid_manager=None):
def __init__(
self,
driver: "AccessoryDriver",
display_name: Optional[str],
iid_manager: Optional[IIDManager] = None,
) -> None:
super().__init__(
driver, display_name, aid=STANDALONE_AID, iid_manager=iid_manager
)
self.accessories = {} # aid: acc

def add_accessory(self, acc):
def add_accessory(self, acc: "Accessory") -> None:
"""Add the given ``Accessory`` to this ``Bridge``.

Every ``Accessory`` in a ``Bridge`` must have an AID and this AID must be
Expand Down Expand Up @@ -364,14 +386,14 @@ def add_accessory(self, acc):

self.accessories[acc.aid] = acc

def to_HAP(self):
def to_HAP(self, include_value: bool = True) -> List[Dict[str, Any]]:
"""Returns a HAP representation of itself and all contained accessories.

.. seealso:: Accessory.to_HAP
"""
return [acc.to_HAP() for acc in (super(), *self.accessories.values())]
return [acc.to_HAP(include_value=include_value) for acc in (super(), *self.accessories.values())]

def get_characteristic(self, aid, iid):
def get_characteristic(self, aid: int, iid: int) -> Optional["Characteristic"]:
""".. seealso:: Accessory.to_HAP"""
if self.aid == aid:
return self.iid_manager.get_obj(iid)
Expand All @@ -382,17 +404,17 @@ def get_characteristic(self, aid, iid):

return acc.get_characteristic(aid, iid)

async def run(self):
async def run(self) -> None:
"""Schedule tasks for each of the accessories' run method."""
for acc in self.accessories.values():
self.driver.async_add_job(acc.run)

async def stop(self):
async def stop(self) -> None:
"""Calls stop() on all contained accessories."""
await self.driver.async_add_job(super().stop)
for acc in self.accessories.values():
await self.driver.async_add_job(acc.stop)


def get_topic(aid, iid):
def get_topic(aid: int, iid: int) -> str:
return str(aid) + "." + str(iid)
Loading