Rework Axis entity loader to have a better internal storage structure (#114114)

* Make one single subscribe containing all topics at once

* Update homeassistant/components/axis/hub/entity_loader.py

Co-authored-by: J. Nick Koston <nick@koston.org>

---------

Co-authored-by: J. Nick Koston <nick@koston.org>
pull/114217/head
Robert Svensson 2024-03-26 07:55:07 +01:00 committed by GitHub
parent d0ecad78ac
commit 3f545cb3d3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 44 additions and 40 deletions

View File

@ -5,7 +5,6 @@ Central point to load entities for the different platforms.
from __future__ import annotations
from functools import partial
from typing import TYPE_CHECKING
from axis.models.event import Event, EventOperation, EventTopic
@ -23,17 +22,20 @@ class AxisEntityLoader:
"""Axis network device integration handling platforms for entity registration."""
def __init__(self, hub: AxisHub) -> None:
"""Initialize the UniFi entity loader."""
"""Initialize the Axis entity loader."""
self.hub = hub
self.registered_events: set[tuple[str, EventTopic, str]] = set()
self.platforms: list[
self.topic_to_entity: dict[
EventTopic,
list[
tuple[
AddEntitiesCallback,
type[AxisEventEntity],
tuple[AxisEventDescription, ...],
AxisEventDescription,
]
] = []
],
] = {}
@callback
def register_platform(
@ -43,37 +45,39 @@ class AxisEntityLoader:
descriptions: tuple[AxisEventDescription, ...],
) -> None:
"""Register Axis entity platforms."""
self.platforms.append((async_add_entities, entity_class, descriptions))
topics: tuple[EventTopic, ...]
for description in descriptions:
if isinstance(description.event_topic, EventTopic):
topics = (description.event_topic,)
else:
topics = description.event_topic
for topic in topics:
self.topic_to_entity.setdefault(topic, []).append(
(async_add_entities, entity_class, description)
)
@callback
def _create_entities_from_event(self, event: Event) -> None:
"""Create Axis entities from event."""
event_id = (event.topic, event.topic_base, event.id)
if event_id in self.registered_events:
# Device has restarted and all events are initialized anew
return
self.registered_events.add(event_id)
for (
async_add_entities,
entity_class,
description,
) in self.topic_to_entity[event.topic_base]:
if not description.supported_fn(self.hub, event):
continue
async_add_entities([entity_class(self.hub, description, event)])
@callback
def initialize_platforms(self) -> None:
"""Prepare event listeners and platforms."""
@callback
def load_entities(
platform_entity: type[AxisEventEntity],
descriptions: tuple[AxisEventDescription, ...],
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up listeners for events."""
@callback
def create_entity(description: AxisEventDescription, event: Event) -> None:
"""Create Axis entity."""
event_id = (event.topic, event.topic_base, event.id)
if event_id in self.registered_events:
# Device has restarted and all events are initiatlized anew
return
self.registered_events.add(event_id)
if description.supported_fn(self.hub, event):
async_add_entities([platform_entity(self.hub, description, event)])
for description in descriptions:
"""Prepare event listener that can populate platform entities."""
self.hub.api.event.subscribe(
partial(create_entity, description),
topic_filter=description.event_topic,
self._create_entities_from_event,
topic_filter=tuple(self.topic_to_entity.keys()),
operation_filter=EventOperation.INITIALIZED,
)
for async_add_entities, entity_class, descriptions in self.platforms:
load_entities(entity_class, descriptions, async_add_entities)