jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-22 20:57:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve type hints in edl21 (#87267)

Browse Source
This commit is contained in:
epenet 2023-02-03 17:09:16 +01:00 committed by GitHub
parent b092698418
commit e40df09098
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 9 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
homeassistant/components/edl21/sensor.py
View File

@ -292,7 +292,12 @@ class EDL21:
"129-129:199.130.5*255", # Iskraemeco: Public Key "129-129:199.130.5*255", # Iskraemeco: Public Key
} }
def __init__(self, hass, config, async_add_entities) -> None: def __init__(
self,
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Initialize an EDL21 object.""" """Initialize an EDL21 object."""
self._registered_obis: set[tuple[str, str]] = set() self._registered_obis: set[tuple[str, str]] = set()
self._hass = hass self._hass = hass
@ -301,7 +306,7 @@ class EDL21:
self._proto = SmlProtocol(config[CONF_SERIAL_PORT]) self._proto = SmlProtocol(config[CONF_SERIAL_PORT])
self._proto.add_listener(self.event, ["SmlGetListResponse"]) self._proto.add_listener(self.event, ["SmlGetListResponse"])
async def connect(self): async def connect(self) -> None:
"""Connect to an EDL21 reader.""" """Connect to an EDL21 reader."""
await self._proto.connect(self._hass.loop) await self._proto.connect(self._hass.loop)
@ -319,7 +324,7 @@ class EDL21:
return return
electricity_id = electricity_id.replace(" ", "") electricity_id = electricity_id.replace(" ", "")
new_entities = [] new_entities: list[EDL21Entity] = []
for telegram in message_body.get("valList", []): for telegram in message_body.get("valList", []):
if not (obis := telegram.get("objName")): if not (obis := telegram.get("objName")):
continue continue
@ -352,7 +357,7 @@ class EDL21:
if new_entities: if new_entities:
self._hass.loop.create_task(self.add_entities(new_entities)) self._hass.loop.create_task(self.add_entities(new_entities))
async def add_entities(self, new_entities) -> None: async def add_entities(self, new_entities: list[EDL21Entity]) -> None:
"""Migrate old unique IDs, then add entities to hass.""" """Migrate old unique IDs, then add entities to hass."""
registry = er.async_get(self._hass) registry = er.async_get(self._hass)