HABApp
HABApp copied to clipboard
spacemanspiff2007
HABApp libraries the second
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
- IDE support, JetBrains PyCharm in particular, which means working code lookup, inspection and overall type-checking
- Full-on OOP (abstract and Rule-implementing classes bequeathing more specialised ones and possibly future mix-ins)
- Dependency files hierarchy
Description of the issue(s)
- 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).
- 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.
- 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).
- 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?
- (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:
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:
ADDITION: I didn't start refactoring yet, but say that I try drawing rules as sources from lib next now: Is it possible to instantiate an imported rule-class in multiple rule-files? E.g. having a lib/milight.py and importing plus instantiating it in rules/lighting.py as well as rules/awning.py? EDIT: It should be, because I instantiated sort of the class MilightDeviceRule(Rule) 50 times in one file via loops beforehand, resulting in MilightDeviceRule.1 to sort of .50. Conformation still appreciated. Cheers!