-
Notifications
You must be signed in to change notification settings - Fork 43
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added leak sensor support. Fixes #6.
- Loading branch information
Showing
9 changed files
with
230 additions
and
46 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| #=========================================================================== | ||
| # | ||
| # Insteon battery powered motion sensor | ||
| # | ||
| #=========================================================================== | ||
| import enum | ||
| from .BatterySensor import BatterySensor | ||
| from .. import log | ||
|
|
||
| LOG = log.get_logger() | ||
|
|
||
|
|
||
| class Leak(BatterySensor): | ||
| """Insteon battery powered water leak sensor. | ||
| A leak sensor is basically an on/off sensor except that it's | ||
| batter powered and only awake when water is detected or the set | ||
| button is pressed. It will broadcast an on command for group 1 | ||
| when dry and on command for group 2 when wet. | ||
| The issue with a battery powered sensor is that we can't download | ||
| the link database without the sensor being on. You can trigger | ||
| the sensor manually and then quickly send an MQTT command with the | ||
| payload 'getdb' to download the database. We also can't test to | ||
| see if the local database is current or what the current motion | ||
| state is - we can really only respond to the sensor when it sends | ||
| out a message. | ||
| The Signal Leak.signal_active(True) will be emitted whenever the | ||
| device senses water and signal_actdive(False) when no water is detected. | ||
| TODO: download the database automatically when motion is seen. | ||
| """ | ||
| # broadcast group ID alert description | ||
| class Type(enum.IntEnum): | ||
| WET = 0x02 | ||
|
|
||
| def __init__(self, protocol, modem, address, name=None): | ||
| """Constructor | ||
| Args: | ||
| protocol: (Protocol) The Protocol object used to communicate | ||
| with the Insteon network. This is needed to allow | ||
| the device to send messages to the PLM modem. | ||
| modem: (Modem) The Insteon modem used to find other devices. | ||
| address: (Address) The address of the device. | ||
| name (str) Nice alias name to use for the device. | ||
| """ | ||
| super().__init__(protocol, modem, address, name) | ||
|
|
||
| # Leak sensor uses group 1 for dry, group 2 for wet. | ||
| self.group_map[0x01] = self.handle_dry | ||
| self.group_map[0x02] = self.handle_wet | ||
|
|
||
| #----------------------------------------------------------------------- | ||
| def handle_dry(self, msg): | ||
| """TODO: doc | ||
| """ | ||
| # off = dry, on == wet | ||
| self._set_is_on(False) | ||
|
|
||
| #----------------------------------------------------------------------- | ||
| def handle_wet(self, msg): | ||
| """TODO: doc | ||
| """ | ||
| # off = dry, on == wet | ||
| self._set_is_on(True) | ||
|
|
||
| #----------------------------------------------------------------------- | ||
| def handle_heartbeat(self, msg): | ||
| """TODO: doc | ||
| """ | ||
| # Update the wet/dry state using the heartbeat if needed. | ||
| is_wet = msg.cmd1 == 0x13 | ||
| if self._is_on != is_wet: | ||
| self._set_is_on(is_wet) | ||
|
|
||
| super().handle_heartbeat(msg) | ||
|
|
||
| #----------------------------------------------------------------------- |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| #=========================================================================== | ||
| # | ||
| # MQTT leak sensor device | ||
| # | ||
| #=========================================================================== | ||
| from .. import log | ||
| from .BatterySensor import BatterySensor | ||
| from .MsgTemplate import MsgTemplate | ||
|
|
||
| LOG = log.get_logger() | ||
|
|
||
|
|
||
| class Leak(BatterySensor): | ||
| """TODO: doc | ||
| This class uses the regular sensor active signal from | ||
| BatterySensor for wet/dry events but we have a different method | ||
| handle_active below to handle the result vs the one in | ||
| BatterySensor. | ||
| """ | ||
| def __init__(self, mqtt, device): | ||
| """TODO: doc | ||
| """ | ||
| super().__init__(mqtt, device) | ||
|
|
||
| self.msg_wet = MsgTemplate( | ||
| topic='insteon/{{address}}/leak', | ||
| payload='{{state.upper()}}', | ||
| ) | ||
|
|
||
| #----------------------------------------------------------------------- | ||
| def load_config(self, config, qos=None): | ||
| """TODO: doc | ||
| """ | ||
| super().load_config(config, qos) | ||
|
|
||
| data = config.get("leak", None) | ||
| if not data: | ||
| return | ||
|
|
||
| self.msg_wet.load_config(data, 'wet_dry_topic', 'wet_dry_payload', qos) | ||
|
|
||
| #----------------------------------------------------------------------- | ||
| # pylint: disable=arguments-differ | ||
| def handle_active(self, device, is_wet): | ||
| """Device wet/dry on/off callback. | ||
| This is triggered via signal when the Insteon device detects | ||
| wet or dry. It will publish an MQTT message with the new | ||
| state. | ||
| Args: | ||
| device: (device.Base) The Insteon device that changed. | ||
| is_wet: (bool) True for wet, False for dry. | ||
| """ | ||
| LOG.info("MQTT received wet/dry change %s '%s' wet= %s", | ||
| device.addr, device.name, is_wet) | ||
|
|
||
| data = { | ||
| "address" : device.addr.hex, | ||
| "name" : device.name if device.name else device.addr.hex, | ||
| "is_wet" : 1 if is_wet else 0, | ||
| "is_wet_str" : "on" if is_wet else "off", | ||
| "is_dry" : 0 if is_wet else 1, | ||
| "is_dry_str" : "off" if is_wet else "on", | ||
| "state" : "wet" if is_wet else "dry", | ||
| } | ||
|
|
||
| self.msg_wet.publish(self.mqtt, data) | ||
|
|
||
| #----------------------------------------------------------------------- |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.