Source code for juham.base.group

import json
from typing import List
from .base import Base, Object


[docs] class Group(Base): """Group base class that can consist of `Base` and `Group` objects as children. This class can be used for grouping home automation objects into larger logical entities. Example: :: motion_sensors = Group("motionsensors") motion_sensors.add(ShellyMotionSensor("downstairs")) motion_sensors.add(ShellyMotionSensor("upstairs")) """ _class_id = "" def __init__(self, name: str = "group") -> None: super().__init__(name) self.children: List = [] self.role: str = "union"
[docs] def on_connect(self, client, userdata, flags, rc): super().on_connect(client, userdata, flags, rc) if rc == 0: self.mqtt_client.connected_flag = True
[docs] def on_message(self, client, userdata, msg): super().on_message(client, userdata, msg) m = json.loads(msg.payload.decode()) self.warning(f"Message {str(m)} not handled")
[docs] def run(self) -> None: """Start up all automation objects in the group and enter the network event loop to process MQTT messages. The method returns only when the process is terminated by some means e.g. Ctrl+C or kill, or by sending 'quit' event to MQTT message broker. """ self.debug("Starting up mqtt clients...") for h in self.children: h.run() self.debug("Starting up control client, press CTRL+C to exit...") try: self.mqtt_client.loop_forever() self.info(f"Networking loop terminated, terminating children...") except: self.info(f"Networking loop terminated, terminating children...") for h in self.children: self.info(f"Terminating {h.name}") h.shutdown()
[docs] def add(self, h: Object) -> None: """Add new automation object as children. The object to be inserted must be derived from Object base class. Args: h (Object): object to be inserted. """ self.children.append(h)
[docs] def to_dict(self): data = super().to_dict() data["_group"] = { "role": self.role, "children": [child.to_dict() for child in self.children], } return data
[docs] def from_dict(self, data): """Recursively deserialize the group from a dictionary, including it children. Args: data (dict): data to deserialize from. Example: :: from juham.base import Group, Base # create a group with a child group = Group("foo") base = Base("bar") group.add(base) # group to dictionary data = group.to_dict() copy_of_group = Group() copy_of_group.from_dict(data) """ super().from_dict(data) for key, value in data.get("_group", {}).items(): if key == "children": for child_dict in value: child = Object.instantiate(child_dict["_class"]) self.add(child) child.from_dict(child_dict) else: setattr(self, key, value)
[docs] @classmethod def register(cls): if cls._class_id == "": Base.register() cls.initialize_class()