Add PowerSwitch sensor sink

mqtt_sensord/mqtt_sensord.py

@@ -33,20 +33,39 @@ def main():
     mqtt = MQTTClient(mcfg['host'], mcfg['user'], mcfg['password'], client_id)
     mqtt_topic = mcfg['sensor_topic']
     print("Connected to", mqtt.host, "with topic", mqtt_topic)
+    mqtt.set_callback(sensorlib.Sensor.run_mqtt_callback)
 
     sensors = []
+    extra_topics = set()
     for scfg in config['sensors']:
         s = sensorlib.Sensor.make_sensor(**scfg)
+        s.register_callbacks()
+        for topic in s.get_topics():
+            extra_topics.add(topic)
         sensors.append(s)
 
+    for topic in extra_topics:
+        print("Subscribe to", topic)
+        mqtt.subscribe(topic)
+
     last_measurement = time.time()
     while True:
+        # handle regular sensor data
         print("Getting values from sensors...")
         for sensor in sensors:
-            data = sensor.gen_datapoint()
-            print(mqtt_topic, data)
-            mqtt.send_data(mqtt_topic, data)
-        time.sleep(max(0.0, time.time() + config['host']['interval'] - last_measurement))
+            if sensor.HAS_SENSOR_DATA:
+                data = sensor.gen_datapoint()
+                print(mqtt_topic, data)
+                mqtt.send_data(mqtt_topic, data)
+            if not mqtt.is_async():
+                mqtt.process_mqtt()
+
+        if mqtt.is_async():
+            time.sleep(max(0.0, time.time() + config['host']['interval'] - last_measurement))
+        else:
+            while time.time() - last_measurement < config['host']['interval']:
+                mqtt.process_mqtt()
+                time.sleep(0.1)
         last_measurement = time.time()
 
 

mqtt_sensord/mqttlib/base.py

@@ -12,3 +12,6 @@ class MQTTClientBase:
 
     def _connect(self):
         raise NotImplementedError("Connect not implemented")
+
+    def is_async(self):
+        return False

mqtt_sensord/mqttlib/paho_mqtt.py

@@ -13,3 +13,12 @@ class MQTTClient(MQTTClientBase):
 
     def send_data(self, topic, data):
         self._mqtt.publish(topic, json.dumps(data))
+
+    def set_callback(self, callback):
+        self._mqtt.on_message = lambda client, userdata, msg: callback(msg.topic, msg.payload.decode())
+
+    def process_mqtt(self):
+        self._mqtt.loop(0.01)
+
+    def subscribe(self, topic):
+        self._mqtt.subscribe(topic)

mqtt_sensord/sensorlib/base.py

@@ -1,3 +1,9 @@
+try:
+    import json
+except ImportError:
+    import ujson as json
+
+
 def sensor(cls):
     Sensor.register(cls)
 
@@ -8,6 +14,8 @@ class Sensor(object):
     hostname = None
     platform = None
     _sensor_classes = {}
+    _mqtt_callbacks = {}
+    HAS_SENSOR_DATA = True
 
     def __init__(self, id, name, type, pin, description, **sensor_conf):
         self.id = id
@@ -45,3 +53,28 @@ class Sensor(object):
             },
             'fields': self.get_data()
         }
+
+    def register_callbacks(self):
+        pass
+
+    def get_mqtt_topics(self):
+        return []
+
+    @classmethod
+    def register_callback(cls, mqtt_topic, callback):
+        cls._mqtt_callbacks[mqtt_topic] = callback
+
+    @classmethod
+    def run_mqtt_callback(cls, mqtt_topic, data):
+        try:
+            data = json.loads(data)
+        except ValueError:
+            print("Could not parse data {}".format(data))
+
+        if mqtt_topic in cls._mqtt_callbacks:
+            cls._mqtt_callbacks[mqtt_topic](data)
+        else:
+            print("Unknown callback for", mqtt_topic)
+
+    def get_topics(self):
+        return []

mqtt_sensord/sensorlib/esp_sensors.py

@@ -3,6 +3,7 @@ import ds18x20
 import machine
 import onewire
 import time
+import ujson
 
 from .base import Sensor, sensor
 
@@ -41,3 +42,76 @@ class DHT(Sensor):
         self.dht.measure()
         time.sleep(0.1)
         return {"temp": self.dht.temperature(), "humidity": self.dht.humidity()}
+
+
+@sensor
+class PowerSwitch(Sensor):
+    sensor_class = 'esp-rcswitch'
+    HAS_SENSOR_DATA = False
+
+    def __init__(self, mqtt_topic, socket_name, socket_count, socket_json, **sensor_conf):
+        super(PowerSwitch, self).__init__(**sensor_conf)
+
+        self.mqtt_topic = mqtt_topic
+        self.socket_name = socket_name
+        self.socket_count = socket_count
+        self.socket_state = ["off"] * socket_count
+        self.socket_code_state = []
+        self._codes_per_call = 2
+        for _ in range(socket_count):
+            self.socket_code_state.append({'on': 0, 'off': 0})
+        self._pin = machine.Pin(self.pin, machine.Pin.OUT)
+        self._load_json(socket_json)
+
+    def _load_json(self, path):
+        with open(path) as f:
+            data = ujson.load(f)
+
+        for entry in data:
+            if entry["name"] == self.socket_name:
+                self._socket_data = entry
+                break
+        else:
+            raise ValueError("Could not find socket codes for {} in {}".format(self.socket_name, path))
+
+    def send_by_compressed_timings(self, buckets, timings, repeats):
+        delays = [buckets[ord(t) - ord('0')] for t in timings]
+        for repeat in range(repeats):
+            self._pin.value(0)
+            state = 0
+            t1 = time.ticks_us()
+            for delay in delays:
+                state ^= 1
+                self._pin.value(state)
+                time.sleep_us(delay - (time.ticks_us() - t1) - 55)
+                t1 = time.ticks_us()
+
+    def switch_light(self, data):
+        try:
+            sock = int(data["socket"])
+            if data["state"] in ("on", "off"):
+                state = data["state"]
+            elif data["state"] == "toggle":
+                state = "on" if self.socket_state[sock] == "off" else "off"
+            else:
+                raise ValueError("Unknown state {}".format(data["state"]))
+            if sock < 0 or sock >= self.socket_count:
+                raise ValueError("Socket {} is out of range".format(sock))
+        except (KeyError, ValueError) as e:
+            print("Could not load data:", e)
+            return
+
+        print("Switch light called with ", data)
+        codes = self._socket_data["codes"]
+        for _ in range(self._codes_per_call):
+            timings = codes[sock][state][self.socket_code_state[sock][state]]
+            self.send_by_compressed_timings(self._socket_data["buckets"], timings, 10)
+            self.socket_state[sock] = state
+            self.socket_code_state[sock][state] = (self.socket_code_state[sock][state] + 1) % len(codes[sock][state])
+        print("Switch socket {} to state {}".format(sock, state))
+
+    def register_callbacks(self):
+        self.register_callback(self.mqtt_topic, self.switch_light)
+
+    def get_topics(self):
+        return [self.mqtt_topic]

mqtt_sensord/sensorlib/misc_sensors.py

@@ -22,6 +22,24 @@ class FakeSensor(Sensor):
         return {'random': random.randint(0, 20)}
 
 
+@sensor
+class TestSink(Sensor):
+    sensor_class = 'testsink'
+    HAS_SENSOR_DATA = False
+
+    def __init__(self, mqtt_topic, **sensor_conf):
+        self.mqtt_topic = mqtt_topic
+
+    def recvd(self, data):
+        print("I got called! Even with data:", data)
+
+    def register_callbacks(self):
+        self.register_callback(self.mqtt_topic, self.recvd)
+
+    def get_topics(self):
+        return [self.mqtt_topic]
+
+
 @sensor
 class CO2Meter(Sensor):
     sensor_class = 'co2meter'