2022-09-28 18:03:22 +01:00
|
|
|
import logging
|
2022-10-06 18:07:41 +01:00
|
|
|
from typing import Iterable, Union
|
2022-09-28 18:03:22 +01:00
|
|
|
|
2023-06-25 13:51:47 +01:00
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
2022-09-28 18:03:22 +01:00
|
|
|
|
2022-08-02 09:28:32 +01:00
|
|
|
class Event:
|
2022-09-29 11:48:30 +01:00
|
|
|
"""Keeps track of event subscriptions"""
|
|
|
|
|
2022-08-02 09:28:32 +01:00
|
|
|
def __init__(self, subs: dict):
|
|
|
|
self.subs = subs
|
2023-06-25 13:51:47 +01:00
|
|
|
self.logger = logger.getChild(self.__class__.__name__)
|
2022-08-02 09:28:32 +01:00
|
|
|
|
2022-09-29 11:48:30 +01:00
|
|
|
def info(self, msg=None):
|
|
|
|
info = (f"{msg} events",) if msg else ()
|
2022-09-28 18:03:22 +01:00
|
|
|
if self.any():
|
|
|
|
info += (f"now listening for {', '.join(self.get())} events",)
|
|
|
|
else:
|
|
|
|
info += (f"not listening for any events",)
|
|
|
|
self.logger.info(", ".join(info))
|
2022-08-02 09:28:32 +01:00
|
|
|
|
|
|
|
@property
|
2022-10-06 16:45:15 +01:00
|
|
|
def pdirty(self) -> bool:
|
2022-08-02 09:28:32 +01:00
|
|
|
return self.subs["pdirty"]
|
|
|
|
|
2022-10-06 16:45:15 +01:00
|
|
|
@pdirty.setter
|
|
|
|
def pdirty(self, val: bool):
|
|
|
|
self.subs["pdirty"] = val
|
2022-10-06 16:50:03 +01:00
|
|
|
self.info(f"pdirty {'added to' if val else 'removed from'}")
|
2022-10-06 16:45:15 +01:00
|
|
|
|
2022-08-02 09:28:32 +01:00
|
|
|
@property
|
2022-10-06 16:45:15 +01:00
|
|
|
def ldirty(self) -> bool:
|
2022-08-02 09:28:32 +01:00
|
|
|
return self.subs["ldirty"]
|
|
|
|
|
2022-10-06 16:45:15 +01:00
|
|
|
@ldirty.setter
|
|
|
|
def ldirty(self, val: bool):
|
|
|
|
self.subs["ldirty"] = val
|
2022-10-06 16:50:03 +01:00
|
|
|
self.info(f"ldirty {'added to' if val else 'removed from'}")
|
2022-10-06 16:45:15 +01:00
|
|
|
|
2022-08-02 09:28:32 +01:00
|
|
|
def get(self) -> list:
|
|
|
|
return [k for k, v in self.subs.items() if v]
|
|
|
|
|
|
|
|
def any(self) -> bool:
|
|
|
|
return any(self.subs.values())
|
|
|
|
|
2022-10-06 18:07:41 +01:00
|
|
|
def add(self, events: Union[str, Iterable[str]]):
|
|
|
|
if isinstance(events, str):
|
|
|
|
events = [events]
|
|
|
|
for event in events:
|
|
|
|
setattr(self, event, True)
|
|
|
|
|
|
|
|
def remove(self, events: Union[str, Iterable[str]]):
|
|
|
|
if isinstance(events, str):
|
|
|
|
events = [events]
|
|
|
|
for event in events:
|
|
|
|
setattr(self, event, False)
|