Pico Door Alarm

"""Public excerpt from the Pico door alarm firmware.

The full operating sequence, exact timings and wiring details are omitted from
the portfolio. This excerpt shows the parts that are useful to discuss: mode
priority, local feedback and rate-limited REST notifications.
"""

import gc
import time

from lib.buzzer import AlarmBuzzer
from lib.connection import ApiClient
from lib.led import BlinkLed, PauseLed
from lib.mode import Mode
from lib.switch import Switch


door_contact = Switch("DOOR_CONTACT_PIN")
green_led = PauseLed("GREEN_LED_PIN", blink_time=0.1, pause_time=1.5)
blue_led = PauseLed("BLUE_LED_PIN", blink_time=0.2, pause_time=1.0)
red_led = BlinkLed("RED_LED_PIN", blink_time=0.08)
buzzer = AlarmBuzzer("BUZZER_PIN")
api_client = ApiClient()

sleep_mode = Mode("sleep")
check_mode = Mode("check")
safety_mode = Mode("safety", duration=20.0)
alarm_mode = Mode("alarm", init_duration=5.0)

alarm_was_active = False
loop_count = 0


def update_modes(sleep_enabled: bool, safety_button_pressed: bool) -> None:
    sleep_mode.set_active(sleep_enabled)

    if sleep_mode.is_active():
        check_mode.set_active(False)
        safety_mode.set_active(False)
        alarm_mode.cancel_pending()
        return

    if safety_button_pressed:
        safety_mode.start()
        alarm_mode.cancel_pending()

    safety_mode.update()
    check_mode.set_active(not safety_mode.is_active() and not alarm_mode.is_active())

    if safety_mode.is_active():
        alarm_mode.cancel_pending()
    elif door_contact.is_opened() and not alarm_mode.is_init_active():
        alarm_mode.start_pending()

    alarm_mode.update_init()


def update_feedback(silent_enabled: bool) -> None:
    green_led.paused_blink() if check_mode.is_active() else green_led.on(False)
    blue_led.on(safety_mode.is_active())
    red_led.blink() if alarm_mode.is_active() else red_led.on(False)

    if silent_enabled:
        buzzer.stop_alarm()
    elif alarm_mode.is_active():
        buzzer.start_or_update_alarm()
    else:
        buzzer.stop_alarm()


def send_notifications() -> None:
    global alarm_was_active

    if alarm_mode.is_active():
        alarm_was_active = True
        api_client.timed_post("Door alarm state changed", "DANGER")

    if safety_mode.is_active() and alarm_was_active:
        alarm_was_active = False
        api_client.post("Door alarm acknowledged by safety mode", "INFO")


def loop_once(inputs: dict[str, bool]) -> None:
    global loop_count

    loop_count += 1
    if loop_count >= 100000:
        loop_count = 0
        gc.collect()

    update_modes(
        sleep_enabled=inputs["sleep_enabled"],
        safety_button_pressed=inputs["safety_button_pressed"],
    )
    update_feedback(silent_enabled=inputs["silent_enabled"])
    send_notifications()
    time.sleep(0.005)

import time


class Mode:
    """Small MicroPython state helper with optional active and pending timers."""

    def __init__(self, name, duration=None, init_duration=None):
        self._name = name
        self._active = False
        self._init_active = False
        self._time = time.ticks_ms() if (duration is not None or init_duration is not None) else None
        self._dur_ms = int(duration * 1000) if duration is not None else None
        self._init_dur_ms = int(init_duration * 1000) if init_duration is not None else None

    def info(self):
        return self._name

    def is_active(self):
        return self._active

    def is_init_active(self):
        return self._init_active

    def set_active(self, state):
        self._active = state

    def set_init_active(self, state):
        self._init_active = state

    def start(self):
        self.reset_time()
        self.set_active(True)

    def start_pending(self):
        self.reset_time()
        self.set_init_active(True)

    def cancel_pending(self):
        self.set_init_active(False)
        self.set_active(False)

    def reset_time(self):
        self._time = time.ticks_ms()

    def update(self):
        if self._dur_ms is not None and self._time is not None:
            now = time.ticks_ms()
            if time.ticks_diff(now, self._time) >= self._dur_ms:
                self._time = now
                self._active = False

    def update_init(self):
        if self._init_dur_ms is not None and self._time is not None:
            now = time.ticks_ms()
            if time.ticks_diff(now, self._time) >= self._init_dur_ms:
                self._time = now
                if self._init_active:
                    self._active = True
                    self._init_active = False

import time
import network
import usocket as socket
import urequests as requests
import ujson as json
import gc

import settings


class ConnectionException(Exception):
    pass


class WifiClient:
    def __init__(self, hostname: str = "") -> None:
        self.wlan = network.WLAN(network.STA_IF)
        self._hostname: str = hostname if hostname else settings.WIFI_HOSTNAME

    def _apply_hostname(self) -> None:
        if not self._hostname:
            return
        try:
            network.hostname(self._hostname)
            return
        except Exception:
            pass

        try:
            self.wlan.config(hostname=self._hostname)
            return
        except Exception:
            pass

        try:
            self.wlan.config(dhcp_hostname=self._hostname)
            return
        except Exception:
            print("Hostname config not supported on this firmware.")

    def connect(self, timeout: int = 20) -> None:
        self.wlan.active(False)
        self._apply_hostname()
        self.wlan.active(True)
        self.wlan.connect(settings.WIFI_SSID, settings.WIFI_PASSWORD)
        start = time.ticks_ms()
        while time.ticks_diff(time.ticks_ms(), start) < timeout * 1000:
            if self.wlan.isconnected():
                ip = self.wlan.ifconfig()[0]
                print(f"Wi-Fi connected (hostname={self._hostname}), IP: {ip}")
                return
            time.sleep(1)
        raise ConnectionException("Wi-Fi connection FAILED")


class ApiClient:
    def __init__(self) -> None:
        self.wifi = WifiClient()
        self.wifi.connect()
        protocol = settings.API_PROTOCOL if hasattr(settings, "API_PROTOCOL") else "https"
        self._base = f"{protocol}://{settings.API_URL}:{settings.API_PORT}{settings.API_SUFFIX}"
        self._send_ms = int(settings.SEND_FREQUENCY * 1000)
        self._last_ms = time.ticks_ms()
        self._sent = False

    def check_connection(self, timeout: float = 1.0) -> bool:
        try:
            addr = socket.getaddrinfo(settings.API_URL, settings.API_PORT)[0][-1]
            s = socket.socket()
            s.settimeout(timeout)
            s.connect(addr)
            s.close()
            return True
        except Exception as e:
            print("Socket check failed:", e)
            return False

    def post(self, text: str, message_type: str) -> None:
        if not self.check_connection():
            print("Re-connecting Wi-Fi...")
            try:
                self.wifi.connect()
            except Exception as ex:
                print("Connecting to Wi-Fi unsuccessful:", ex)

        payload: dict = {"text": text, "type": message_type}
        body = json.dumps(payload)
        headers = {"Content-Type": "application/json"}

        try:
            response = requests.post(self._base, data=body, headers=headers)
            print("POST status:", response.status_code)
            response.close()
            gc.collect()
            if response.status_code >= 400:
                try:
                    err = response.json()
                except Exception:
                    err = "No exception message..."
                raise ConnectionException(
                    f"Server error {response.status_code}: {err}"
                )
        except Exception as e:
            print("POST failed:", e)

    def timed_post(self, text: str, message_type: str) -> None:
        now = time.ticks_ms()

        if not self._sent:
            self._last_ms = now
            self._sent = True
            self.post(text, message_type)
        elif time.ticks_diff(now, self._last_ms) >= self._send_ms:
            self._sent = False