UG Buttons

ug/button.py

@@ -0,0 +1,320 @@
+import subprocess
+import threading
+import time
+
+import paho.mqtt.client as mqtt
+
+SIM = False
+
+if SIM:
+    import numpy as np
+    from matplotlib import pyplot as plt
+    import os
+    from adafruit_platformdetect.constants import boards, chips
+
+    os.environ["BLINKA_FORCEBOARD"] = boards.OS_AGNOSTIC_BOARD
+    os.environ["BLINKA_FORCECHIP"] = chips.OS_AGNOSTIC
+    from adafruit_blinka.agnostic import detector
+
+    detector.chip.AG = True
+
+import board
+import neopixel
+import digitalio
+from adafruit_led_animation.animation.chase import Chase
+from adafruit_led_animation import color
+
+if SIM:
+    board.D18 = type("Pin", (), {"id": -1})()
+
+
+class NeoPixelSim(neopixel.NeoPixel):
+    @property
+    def pin(self):
+        return type("DummyPin", (), {
+            "deinit": lambda: None,
+        })
+
+    @pin.setter
+    def pin(self, value):
+        pass
+
+    def _transmit(self, buffer: bytearray) -> None:
+        mat = np.frombuffer(buffer, dtype=np.uint8).reshape((self.n, 1, 3)).copy()
+        mat[..., list(range(len(self._byteorder)))] = mat[..., list(self._byteorder)]
+        plt.ion()
+        plt.imshow(mat)
+        fig = plt.gcf()
+        fig.canvas.draw_idle()
+        fig.canvas.flush_events()
+
+
+class SubStrip:
+    def __init__(self, strip, pixel: list[int]):
+        self.strip: neopixel.NeoPixel = strip
+        self._pixel_map = pixel
+
+    def __len__(self):
+        return len(self._pixel_map)
+
+    def __setitem__(self, key, value):
+        if isinstance(key, slice):
+            for i, v in zip(range(*key.indices(len(self._pixel_map))), value):
+                self.strip[self._pixel_map[i]] = v
+        else:
+            self.strip[self._pixel_map[key]] = value
+
+    def show(self):
+        self.strip.show()
+
+    def fill(self, color):
+        for p in self._pixel_map:
+            self.strip[p] = color
+
+
+class Door:
+    def __init__(self, host, user_open, user_close):
+        self.host = host
+        self.user_open = user_open
+        self.user_close = user_close
+        self.inprogress = False
+
+    def _ssh(self, host, name):
+        def _ssh_(host_, name_):
+            self.inprogress = True
+            try:
+                print("[ssh]", f"{name_}@{host_}")
+                subprocess.check_call(["ssh", f"{name_}@{host_}"], stdin=subprocess.DEVNULL)
+            except subprocess.CalledProcessError as ex:
+                print("[ssh]", ex)
+            self.inprogress = False
+
+        threading.Thread(target=_ssh_, args=(host, name)).start()
+
+    def open(self):
+        self._ssh(self.host, self.user_open)
+
+    def close(self):
+        self._ssh(self.host, self.user_close)
+
+
+class DoorLocal(Door):
+    def __init__(self, cmd_open, cmd_close):
+        super().__init__("localhost", cmd_open, cmd_close)
+        self.door_sense_state = None
+
+    def _ssh(self, host, name):
+        def _write_(name_):
+            self.inprogress = True
+            for _ in range(20):
+                time.sleep(1)
+                if self.door_sense_state is False:  # door closed
+                    break
+            else:  # Door is still open
+                self.inprogress = False
+                return
+
+            time.sleep(2)
+            if self.door_sense_state is False:  # door still closed
+                with open("/var/run/foodoord.pipe", "a") as f:
+                    f.write(name_ + "\n")
+            self.inprogress = False
+
+        threading.Thread(target=_write_, args=(name,)).start()
+
+
+class FoodoorMQTT:
+    def __init__(self, areas):
+        self.areas = areas
+
+        self.client = mqtt.Client()
+        self.client.on_connect = self.on_connect
+        self.client.on_message = self.on_message
+
+        self._connect_lock = threading.Condition()
+        self._data = {}
+
+    def _subscribe(self):
+        for area in self.areas:
+            self.client.subscribe(f"foobar/{area}/foodoor/status", 1)
+
+    def connect(self, host="mqtt.chaospott.de"):
+        try:
+            self.client.connect(host)
+            self.client.loop_start()
+            with self._connect_lock:
+                self._connect_lock.wait()
+        except Exception as e:
+            print(f"Verbindungsfehler zu MQTT-Server: {e}")
+
+    def disconnect(self):
+        self.client.loop_stop()
+
+    def on_connect(self, client, userdata, flags, rc):
+        self._subscribe()
+        with self._connect_lock:
+            self._connect_lock.notify()
+
+    def on_message(self, client, userdata, msg):
+        print(f"MQTT-Server Message: {msg.topic} {msg.payload.decode()}")
+        self._data[msg.topic] = msg.payload.decode()
+
+    def get_door(self, area):
+        return self._data.get(f"foobar/{area}/foodoor/status")
+
+    def is_open(self, area):
+        return self.get_door(area) == "open"
+
+
+class StateMenu:
+    OFF = 0
+    SHOW_STATE = 1
+    CHANGE = 2
+    PROGRESS = 3
+
+    def __init__(self):
+        self.state = StateMenu.OFF
+        self.change_target = None
+        self.target_state = None
+        self.anim = []
+        self.timeout = None
+        self.next_state = None
+
+        self.mqtt = FoodoorMQTT(["oben", "unten"])
+        if SIM:
+            self.mqtt.connect("127.0.0.1")
+        else:
+            self.mqtt.connect()
+
+        if SIM:
+            self.strip = NeoPixelSim(board.D18, 24, brightness=1.5, pixel_order=neopixel.GRB, auto_write=False)
+        else:
+            self.strip = neopixel.NeoPixel(board.D18, 24, brightness=0.5, pixel_order=neopixel.GRB, auto_write=False)
+        self.strip_oben = SubStrip(self.strip, pixel=list(range(12)))
+        self.strip_unten = SubStrip(self.strip, pixel=list(range(13, 24)))
+
+        self.doors = {
+            "oben": Door("10.42.1.28", "open", "close"),
+            # "unten": Door("10.42.1.20", "open", "close"),
+            "unten": DoorLocal("open", "close"),
+        }
+
+    def reset_state(self):
+        self.state = StateMenu.OFF
+        self.change_target = None
+        self.target_state = None
+        self.anim.clear()
+        self.timeout = None
+        self.next_state = None
+
+    def button_pressed(self, area):
+        if self.state == StateMenu.OFF:
+            self.set_state(StateMenu.SHOW_STATE, 10)
+        elif self.state == StateMenu.SHOW_STATE:
+            self.set_state(StateMenu.CHANGE, 10)
+            self.change_target = area
+        elif self.state == StateMenu.CHANGE:
+            self.set_state(StateMenu.PROGRESS, 5)
+            self.next_state = StateMenu.SHOW_STATE
+            self.target_state = "open" if area == "oben" else "close"
+
+            print(self.change_target, self.target_state)
+            if self.target_state == "open":
+                self.doors[self.change_target].open()
+            elif self.target_state == "close":
+                self.doors[self.change_target].close()
+
+        elif self.state == StateMenu.PROGRESS:
+            # self.state = StateMenu.SHOW_STATE
+            # self.timeout = time.monotonic() + 10
+            pass
+
+        self.draw()
+
+    def set_state(self, state, timeout):
+        self.state = state
+        self.timeout = time.monotonic() + timeout
+
+    def draw(self):
+        self.anim.clear()
+
+        if self.state == StateMenu.OFF:
+            self.strip.fill(color.BLACK)
+        elif self.state == StateMenu.SHOW_STATE:
+            self.strip_oben.fill(color.GREEN if self.mqtt.is_open("oben") else color.RED)
+            self.strip_unten.fill(color.GREEN if self.mqtt.is_open("unten") else color.RED)
+        elif self.state == StateMenu.CHANGE:
+            self.anim = [
+                Chase(self.strip_oben, 0.5, color=color.GREEN, size=1, spacing=11),
+                Chase(self.strip_unten, 0.5, color=color.RED, size=1, spacing=11),
+            ]
+        elif self.state == StateMenu.PROGRESS:
+            self.strip_oben.fill(color.GREEN if self.mqtt.is_open("oben") else color.RED)
+            if self.change_target == "oben":
+                self.strip_unten.fill(color.GREEN if self.mqtt.is_open("unten") else color.RED)
+                self.anim = [Chase(self.strip_oben, 0.05, size=1,
+                    color=color.GREEN if self.target_state == "open" else color.RED)]
+            elif self.change_target == "unten":
+                self.strip_oben.fill(color.GREEN if self.mqtt.is_open("oben") else color.RED)
+                self.anim = [Chase(self.strip_unten, 0.05, size=1,
+                    color=color.GREEN if self.target_state == "open" else color.RED)]
+
+    def tick(self):
+        if self.timeout and self.timeout < time.monotonic() and not (self.change_target and self.doors[self.change_target].inprogress):
+            if self.next_state is not None:
+                self.set_state(self.next_state, 10)
+                self.next_state = None
+            else:
+                self.reset_state()
+            self.draw()
+
+        for a in self.anim:
+            a.animate(show=False)
+        self.strip.show()
+
+    def deinit(self):
+        self.strip.deinit()
+        self.mqtt.disconnect()
+
+
+def main():
+    btn_oben = digitalio.DigitalInOut(board.D23)
+    btn_oben.switch_to_input()
+
+    btn_unten = digitalio.DigitalInOut(board.D17)
+    btn_unten.switch_to_input()
+
+    door_sense = digitalio.DigitalInOut(board.D22)
+    door_sense.switch_to_input(digitalio.Pull.UP)
+
+    menu = StateMenu()
+    btn_locked = False
+
+    try:
+        while True:
+
+            if not btn_oben.value:
+                if not btn_locked:
+                    menu.button_pressed("oben")
+                btn_locked = True
+            elif not btn_unten.value:
+                if not btn_locked:
+                    menu.button_pressed("unten")
+                btn_locked = True
+            else:
+                btn_locked = False
+
+            if door_sense.value != menu.doors["unten"].door_sense_state:
+                menu.doors["unten"].door_sense_state = door_sense.value
+                menu.mqtt.client.publish("foobar/unten/foodoor/door",
+                    {True: "open", False: "closed"}[menu.doors["unten"].door_sense_state],
+                    retain=True)
+
+            menu.tick()
+            time.sleep(.01)
+    except KeyboardInterrupt:
+        menu.deinit()
+
+
+if __name__ == '__main__':
+    main()