49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
"""Class to track subscription event statistics."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from soco.data_structures_entry import from_didl_string
|
|
from soco.events_base import Event as SonosEvent, parse_event_xml
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class EventStatistics:
|
|
"""Representation of Sonos event statistics."""
|
|
|
|
def __init__(self, zone_name: str) -> None:
|
|
"""Initialize EventStatistics."""
|
|
self._stats = {}
|
|
self.zone_name = zone_name
|
|
|
|
def receive(self, event: SonosEvent) -> None:
|
|
"""Mark a received event by subscription type."""
|
|
stats_entry = self._stats.setdefault(
|
|
event.service.service_type, {"received": 0, "duplicates": 0, "processed": 0}
|
|
)
|
|
stats_entry["received"] += 1
|
|
|
|
def duplicate(self, event: SonosEvent) -> None:
|
|
"""Mark a duplicate event by subscription type."""
|
|
self._stats[event.service.service_type]["duplicates"] += 1
|
|
|
|
def process(self, event: SonosEvent) -> None:
|
|
"""Mark a fully processed event by subscription type."""
|
|
self._stats[event.service.service_type]["processed"] += 1
|
|
|
|
def report(self) -> dict:
|
|
"""Generate a report for use in diagnostics."""
|
|
payload = self._stats.copy()
|
|
payload["soco:from_didl_string"] = from_didl_string.cache_info()
|
|
payload["soco:parse_event_xml"] = parse_event_xml.cache_info()
|
|
return payload
|
|
|
|
def log_report(self) -> None:
|
|
"""Log event statistics for this speaker."""
|
|
_LOGGER.debug(
|
|
"Event statistics for %s: %s",
|
|
self.zone_name,
|
|
self.report(),
|
|
)
|