HABApp libraries the second

[SPF79]

Hey mate,

alright, I did my research on the approaches to clean libraries you mentioned here Yet, they don't provide any solution to my code. Reiterating...

Requirements

1. IDE support, JetBrains PyCharm in particular, which means working code lookup, inspection and overall type-checking
2. Full-on OOP (abstract and Rule-implementing classes bequeathing more specialised ones and possibly future mix-ins)
3. Dependency files hierarchy

Description of the issue(s)

1. When I use the "lib" folder for source files, there shan't be rules therein (docs). So I can't use "lib" as a source to implement more abstract rule mechanics in the "rules" folder (EDIT: I just realised, that this probably only counts for rule instantiation, not imports. Therefore omittable, whereas confirmation would be appreciated). Apart from that, relative/absolute imports don't work as expected and throw PyCharm off the rails ("from lib.xyz import blah" -> error@runtime).
2. When I go by get_rule(), I can't get the typed class to be looked-up/instantiated. I work with static variables in stub classes to achieve productive code-completion and I would never want to change my style here.
3. I'm quite positive, that I'll run into the same issues with option 3 as with option 2 (correct me if I'm wrong).
4. Unrelated, but your (Mqtt)ValueEventFilter implementation is somehow off. If I instantiate one in Item.listen_event(), it wants the event argument, but it should be optional. Thus I have to suppress a ton of PyCharm inspections.

Example code

to be put in a library file

class MilightHub:
    class Topics:
        Milight = "milight"
        Commands = "commands"
        States = "states"
        Updates = "updates"

    _hostname = None  # type: str

    @property
    def hostname(self):
        # type: () -> str
        return self._hostname

    def __init__(self, hostname):
        # type: (str) -> None
        self._hostname = hostname

    @property
    def mqtt_topic(self):
        # type: () -> str
        return "/".join((self.Topics.Milight, self.hostname))

    @property
    def mqtt_commands_topic(self):
        # type: () -> str
        return "/".join((self.mqtt_topic, self.Topics.Commands)) + "/"

    @property
    def mqtt_states_topic(self):
        # type: () -> str
        return "/".join((self.mqtt_topic, self.Topics.States)) + "/"

    @property
    def mqtt_updates_topic(self):
        # type: () -> str
        return "/".join((self.mqtt_topic, self.Topics.Updates)) + "/"

class MilightDevice:
    _hex_string = None  # type: str
    _groups = None  # type: List[MilightDeviceGroup]
    _title = None  # type: str
    _hub = None  # type: MilightHub

    @property
    def log(self):
        return log.getChild("'%s'" % self.title)

    @property
    def title(self):
        # type: () -> str
        return self._title

    @property
    def hub(self):
        # type: () -> MilightHub
        return self._hub

    def __init__(self, title, hex_string, hub):
        # type: (str, str, MilightHub) -> None
        self._title = title
        self._hex_string = hex_string
        self._groups = []
        self._hub = hub
        MilightLightingDeviceGroup(self.title, self, group_no=0)

    @property
    def hex_string(self):
        # type: () -> str
        return self._hex_string

    @property
    def groups(self):
        # type: () -> List[MilightDeviceGroup]
        return self._groups

    @groups.setter
    def groups(self, value):
        # type: (List[MilightDeviceGroup]) -> None
        self._groups = value

    @property
    def mqtt_device_pattern(self):
        # type: () -> str
        return "Device_%s" % self.hex_string

    def __str__(self):
        return "Milight Device '%s'" % self.title

    @property
    def all(self):
        # type: () -> MilightDeviceGroup
        return self.groups[0]

class MilightDeviceGroup(Rule):
    _title = None  # type: str
    _device = None  # type: MilightDevice
    _group_no = None  # type: int

    _command_topic = None  # type: str
    _update_topic = None  # type: str
    _state_topic = None  # type: str
    state_item = None  # type: MqttItem
    command_item = None  # type: MqttItem
    update_item = None  # type: MqttItem

    _state = None  # type: OnOffValue

    @property
    def log(self):
        return log.getChild(self.title)

    def __init__(self, title, device, group_no, kind="rgb_cct"):
        # type: (str, MilightDevice, int, str) -> None
        Rule.__init__(self)
        self._title = title
        self._device = device
        self._group_no = group_no
        self._command_topic = device.hub.mqtt_commands_topic + "0x%s/%s/%s" % (device.hex_string, kind, group_no)
        self._update_topic = device.hub.mqtt_updates_topic + "0x%s/%s/%s" % (device.hex_string, kind, group_no)
        self._state_topic = device.hub.mqtt_states_topic + "0x%s/%s/%s" % (device.hex_string, kind, group_no)
        self.state_item = MqttItem.get_create_item(self._state_topic)
        self.command_item = MqttItem.get_create_item(self._command_topic)
        self.update_item = MqttItem.get_create_item(self._update_topic)
        self.pending_values = {}

    @property
    def title(self):
        # type: () -> str
        return self._title

    @property
    def device(self):
        # type: () -> MilightDevice
        return self._device

    @property
    def group_no(self):
        # type: () -> int
        return self._group_no

    @property
    def state(self):
        # type: () -> OnOffValue
        return self._state

    last_reset = None  # type: datetime
    recent_update_contents = None  # type: List[str]

    remapping_dict = {
            "night_mode"
    }

    pending_values = None  # type: Dict[str, Any]

    def process_update(self, event):
        # type: (MqttValueUpdateEventFilter()) -> Optional[Dict[str, Any]]
        if self.last_reset is None or self.last_reset + timedelta(seconds=2) < datetime.now():
            self.recent_update_contents = []
        if event.value in self.recent_update_contents:
            return
        self.last_reset = datetime.now()
        self.recent_update_contents.append(event.value)
        self.log.debug("Updating: %s" % event.value)
        values = comprehend_json(event.value)
        if not self.pending_values:
            self.pending_values = values
        else:
            for k, v in values.items():
                self.pending_values[k] = v
        if any("night_mode" in v for v in values):
            return
        if "state" in values:
            self._state = OnOffValue.ON if values["state"] == "ON" else OnOffValue.OFF
        return values

    def on(self):
        self.command({"state": "ON"})

    def off(self):
        self.command({"state": "OFF"})

    def publish(self, json_string, delay=None):
        # type: (str, timedelta) -> None
        if delay is None:
            self.command_item.publish(json_string)
        else:
            self.run.at(datetime.now() + delay, self.command_item.publish, json_string)

    def command(self, payload):
        # type: (Dict) -> None
        json_string = json.dumps(payload)
        self.publish(json_string)

    def __str__(self):
        return "%s Group '%s'" % (self.device, self.group_no)

class MilightLightingDeviceGroup(MilightDeviceGroup):
    properties = dict(
            level=0.0,
            mode=0.0,
            hue=0.0,
            saturation=0.0,
            brightness=0.0,
            kelvin=0.0,
            color_temp=0.0,
    )

    def __init__(self, title, device, group_no, kind="rgb_cct"):
        super(MilightLightingDeviceGroup, self).__init__(title, device, group_no, kind=kind)
        self._device.groups.append(self)
        self.update_item.listen_event(self.process_update, ValueUpdateEventFilter())
        self.state_item.listen_event(self.process_update, ValueUpdateEventFilter())

    @property
    def level(self):
        # type: () -> float
        return self.properties["level"]

    @level.setter
    def level(self, value):
        # type: ((float, str)) -> None
        self.properties["level"] = float(value)

    @property
    def hue(self):
        # type: () -> float
        return self.properties["hue"]

    @hue.setter
    def hue(self, value):
        # type: (float) -> None
        self.properties["hue"] = float(value)

    @property
    def saturation(self):
        # type: () -> float
        return self.properties["saturation"]

    @saturation.setter
    def saturation(self, value):
        # type: (float) -> None
        self.properties["saturation"] = float(value)

    @property
    def brightness(self):
        # type: () -> float
        return self.properties["brightness"]

    @brightness.setter
    def brightness(self, value):
        # type: (float) -> None
        self.properties["brightness"] = float(value)

    @property
    def color_temp(self):
        # type: () -> float
        return self.properties["color_temp"]

    @color_temp.setter
    def color_temp(self, value):
        # type: (float) -> None
        self.properties["color_temp"] = float(value)

    def process_update(self, event):
        # type: (MqttValueUpdateEventFilter()) -> None
        values = super(MilightLightingDeviceGroup, self).process_update(event)
        if values is None:
            return
        for k, v in self.properties.items():
            if k in values:
                self.properties[k] = float(values[k])

to be put in an implementation/configuration file

class MilightHubs:
    Over_There = MilightHub("that_one")
    Over_Here = MilightHub("this_one")
    ...

class MilightDevices:
    Some_Lights = MilightDevice("Some Lights", "1", MilightHubs.Over_There)
    Some_More_Lights = MilightDevice("Some More Lights", "2", MilightHubs.Over_Here)
    ...

class SingleLights:
    Some_Lights_First = MilightLightingDeviceGroup(
            "First", MilightDevices.Some_Lights, 1
    )
    Some_Lights_Second = MilightLightingDeviceGroup(
            "Second", MilightDevices.Some_Lights, 2
    )
    ...

What do you suggest?

[spacemanspiff2007]

1. (EDIT: I just realised, that this probably only counts for rule instantiation, not imports. Therefore omittable, whereas confirmation would be appreciated). Apart from that, relative/absolute imports don't work as expected and throw PyCharm off the rails ("from lib.xyz import blah" -> error@runtime).

You can have the class definition there but not the class instantiation. The class instantiation has to happen in a rule file. It then works like any normal 3rd party package. To make it in PyCharm work you just have to add the library folder as an additional source folder. Then you import from lib_a import xyz.

2. When I go by get_rule(), I can't get the typed class to be looked-up/instantiated. I work with static variables in stub classes to achieve productive code-completion and I would never want to change my style here.

get_rule() returns the already created rule instance. I've provided an example how to make it work with type hints.

What do you suggest?

From skimming over your code it seems you mainly want to interact with each device through hsb, on/off, level and color-temp values. I still think the it's easiest when you create four HABApp internal items per device and a rule per device which listens to a command event (you can reuse the openHAB ItemCommandEvent) on the items and then publishes to mqtt accordingly. On device update from mqtt it updates all four items. Grouping then becomes a matter of grouping the items and sending the events accordingly which is easy. The rule is basically a driver and the items represent the (most used) device state. For more exotic use cases you can still use the get_rule() and call functions on the rule directly.

4. If I instantiate one in Item.listen_event(), it wants the event argument, but it should be optional.

For me it's fine without issues:

[SPF79]

Alright then, I'll give it another shot and value the time you took with devotion.

At the same time, these were just demonstrating excerpts, since as I mentioned before, I hitch-hike Milight to control e.g. my awning relays. They have such nice out-of-the-box useful (handheld as well as wall-mounted) remotes (EDIT: Milight that is) to signal my hubs for MQTT-messages to OH. So please don't think that lighting is the only application. That's exactly why I want libraries to feed my implementations (and I don't use OH items as relays to bridge whatsoever, since that is just to slow, but only to keep track of the states - EDIT3: or at least that was a train of thought in between years ago, I dunno any more, but this code-base works sufficiently).

I know, that this is sort of rogue and not as intended. But well, in my sector you have to get creative to keep costs below par.

Cheers for the prompt reply and will keep you in the loop, mate!

EDIT2: And for me it doesn't: