class EventsRegistry(DescriptorRegistry):
"""
A `DescriptorRegistry` for events of a `Thing` class or `Thing` instance.
[UML Diagram](https://docs.hololinked.dev/UML/PDF/DescriptorRegistry.pdf)
"""
@property
def descriptor_object(self):
"""The type of descriptor object that this registry holds, i.e. `Event`."""
return Event
descriptors = property(DescriptorRegistry.get_descriptors) # type: dict[str, Event]
values = property(DescriptorRegistry.get_values, doc=DescriptorRegistry.get_values.__doc__) # type: dict[str, EventDispatcher]
def __getitem__(self, key: str) -> Event | EventDispatcher:
if self.owner_inst is not None:
return self.descriptors[key].__get__(self.owner_inst, self.owner_cls)
return self.descriptors[key]
def __contains__(self, obj: object) -> bool:
return obj in self.descriptors.values() or obj in self.descriptors
def clear(self):
"""Deletes the cached descriptor dictionaries so that they can be recreated, including change events and observables."""
super().clear()
for attr in ["_change_events", "_observables"]:
try:
delattr(
self,
f"_{self._qualified_prefix}_{self.__class__.__name__.lower()}{attr}",
)
except AttributeError:
pass
@property
def plain(self) -> dict[str, Event]:
"""Dictionary of events that are not change events (i.e., not observable)."""
try:
return getattr(
self,
f"_{self._qualified_prefix}_{self.__class__.__name__.lower()}_non_change_events",
)
except AttributeError:
non_change_events = dict()
for name, evt in self.descriptors.items():
if not evt._observable:
non_change_events[name] = evt
setattr(
self,
f"_{self._qualified_prefix}_{self.__class__.__name__.lower()}_non_change_events",
non_change_events,
)
return non_change_events
@property
def change_events(self) -> dict[str, Event]:
"""Dictionary of change events belonging to observable properties."""
try:
return getattr(
self,
f"_{self._qualified_prefix}_{self.__class__.__name__.lower()}_change_events",
)
except AttributeError:
change_events = dict()
for name, evt in self.descriptors.items():
if not evt._observable:
continue
change_events[name] = evt
setattr(
self,
f"_{self._qualified_prefix}_{self.__class__.__name__.lower()}_change_events",
change_events,
)
return change_events
@property
def observables(self) -> dict[str, Property]:
"""Dictionary of all properties that are observable, i.e. that which push change events."""
try:
return getattr(
self,
f"_{self._qualified__prefix}_{self.__class__.__name__.lower()}_observables",
)
except AttributeError:
props = dict()
for name, prop in self.owner_cls.properties.descriptors.items():
if not isinstance(prop, Property) or not prop.observable:
continue
props[name] = prop
setattr(
self,
f"_{self._qualified_prefix}_{self.__class__.__name__.lower()}_observables",
props,
)
return props