pico/.gitignore

@@ -1,4 +1,3 @@
-build
 .venv
 typings
 src/secret.py

pico/Makefile

@@ -1,27 +1,6 @@
-SRC_DIR := src
-BUILD_DIR := build
-
-SRC_FILES := $(wildcard $(SRC_DIR)/*.py) $(wildcard $(SRC_DIR)/**/*.py)
-BUILD_FILES := $(patsubst $(SRC_DIR)/%.py,$(BUILD_DIR)/%.mpy, $(SRC_FILES))
-	
-$(BUILD_DIR)/%.mpy: $(SRC_DIR)/%.py
-	@mkdir -p `dirname $@`
-	mpy-cross $< -o $@
-
-.PHONY: build
-build: $(BUILD_FILES)
-
-.PHONY: clean
-clean:
-	rm -rf $(BUILD_DIR)
-
 .PHONY: deploy
-deploy: $(BUILD_FILES)
+deploy:
-	mpremote fs cp -r $(BUILD_DIR)/* :
+	mpremote fs cp src/* :
-	
-.PHONY: purge
-purge:
-	mpremote run purge.py
 
 .PHONY: run
 run: deploy

pico/README.md

@@ -1,20 +1,6 @@
 # Setup:
-1. Set up the virtual environment:
+```
-   ```
+$ python -m venv .venv
-   $ python -m venv .venv
+$ source .venv/bin/activate
-   $ source .venv/bin/activate
+$ pip install -U micropython-rp2-pico_w-stubs --no-user --target typings
-   ```
+```
-
-2. Install the type stubs:
-   ```
-   $ pip install -U micropython-rp2-pico_w-stubs --no-user --target typings
-   ```
-
-3. Ensure `mpy-cross` is available:
-   ```
-   $ git clone https://github.com/micropython/micropython.git /tmp/micropython
-   $ pushd /tmp/micropython/mpy-cross
-   $ make
-   $ mv build/mpy-cross ~/.local/bin/
-   $ popd
-   ```

pico/purge.py

@@ -1,14 +0,0 @@
-import os
-
-def rm(p: str, cwd: str = "/"):
-    path = cwd+p
-    print(f"rm {path=}")
-    try:
-        os.unlink(path)
-    except OSError:
-        for f in os.listdir(path):  # pyright: ignore[reportAny]
-            rm(f, path+"/")  # pyright: ignore[reportAny]
-        os.unlink(path)
-
-for f in os.listdir():  # pyright: ignore[reportAny]
-    rm(f)  # pyright: ignore[reportAny]

pico/src/connection.py

@@ -1,46 +0,0 @@
-import network
-import asyncio
-
-from log import Logger, Color
-from statusled import STATUS_LED
-
-async def ensure_network():
-    from secret import SSID, PASSWORD
-    log = Logger("net", Color.BLUE)
-
-    nic = network.WLAN(network.STA_IF)
-    log.info("setting up WLAN interface...")
-    nic.active(True)
-    while not nic.active():
-        await asyncio.sleep(0.1)
-    log.info("connecting...")
-    nic.connect(SSID, PASSWORD)
-
-    logged = False
-
-    status_name = {
-        network.STAT_CONNECT_FAIL: "CONNECT_FAIL",
-        network.STAT_CONNECTING: "CONNECTING",
-        network.STAT_IDLE: "IDLE",
-        network.STAT_NO_AP_FOUND: "NO_AP_FOUND",
-        network.STAT_WRONG_PASSWORD: "WRONG_PASSWORD"
-    }
-
-    while True:
-        if (status := nic.status()) == network.STAT_GOT_IP:  # pyright: ignore[reportAny]
-            STATUS_LED.status(2)
-            if not logged:
-                logged = True
-                log.success(f"Got IP: {nic.ifconfig()!r}")
-            rssi = nic.status("rssi")  # pyright: ignore[reportAny]
-            log.info(f"{rssi=}")
-        else:
-            logged = False
-            STATUS_LED.status(1)
-            log.warning("status={}".format(status_name.get(status, f"UNKNOWN ({status})")))  # pyright: ignore[reportAny]
-            if status == network.STAT_CONNECT_FAIL or status == network.STAT_NO_AP_FOUND:
-                log.info("retrying in 3s...")
-                await asyncio.sleep(3)
-                nic.connect(SSID, PASSWORD)
-        
-        await asyncio.sleep(1)

pico/src/log.py

@@ -1,45 +0,0 @@
-def _color(n: int):
-    return f"\033[{n}m"
-
-class Color:
-    RED: str = _color(31)
-    GREEN: str = _color(32)
-    YELLOW: str = _color(33)
-    BLUE: str = _color(34)
-    MAGENTA: str = _color(35)
-    CYAN: str = _color(36)
-    RESET: str = _color(0)
-
-
-class Logger:
-    tag: str
-    color: str | None
-
-    def __init__(self, tag: str, color: str | None = None):
-        self.tag = tag
-        self.color = color
-
-    def info(self, message: str):
-        self._impl("*", None, message)
-
-    def success(self, message: str):
-        self._impl("+", Color.GREEN, message)
-
-    def warning(self, message: str):
-        self._impl("w", Color.YELLOW, message)
-
-    def error(self, message: str):
-        self._impl("!", Color.RED, message)
-
-    def _impl(self, symbol: str, color: str | None, message: str):
-        prefix_len = len(symbol) + len(self.tag) + 3  # "[t|s]"
-        
-        tag = f"{self.color}{self.tag}{Color.RESET}" if self.color else self.tag
-        symbol = f"{color}{symbol}{Color.RESET}" if color else symbol
-        prefix = f"[{tag}|{symbol}]"
-        padding = prefix_len * " "
-
-        first = True
-        for line in message.splitlines():
-            print(prefix if first else padding, line)
-            first = False

pico/src/main.py

@@ -1,38 +1,117 @@
+from machine import Pin
+import machine
+import network
+import time
 import asyncio
 
-from vendor.microdot import Microdot, Request
+class StatusLED:
+    led: Pin
+    _status: int
+    _period: float
+    _duty_cycle: float
 
-from statusled import STATUS_LED
+    def __init__(self):
-from log import Logger, Color
+        self.led = Pin("LED", Pin.OUT)
-from sensors.ultrasonic import UltraSonicSensor
+        self._status = 1
-from connection import ensure_network
+        self._period = 1
+        self._duty_cycle = 0.5
 
-app = Microdot()
+    def status(self, status: int):
+        self._status = status
+        
+    
+    async def task(self):
+        while True:
+            t_1 = self._duty_cycle * self._period
+            t_2 = (1 - self._duty_cycle) * self._period
 
-@app.route("/")
+            t_blink = t_1 / self._status / 2
-async def index(_request: Request):
+            for _ in range(self._status):
-    return "Hello from pico"
+                self.led.value(1)
+                await asyncio.sleep(t_blink)
+                self.led.value(0)
+                await asyncio.sleep(t_blink)
+            
+            await asyncio.sleep(t_2)
+
+STATUS_LED = StatusLED()
+
+
+class UltraSonicSensor:
+    """
+    See https://randomnerdtutorials.com/micropython-hc-sr04-ultrasonic-esp32-esp8266/
+    """
+
+    tx: Pin
+    rx: Pin
+
+    def __init__(self, tx: int, rx: int):
+        self.tx = Pin(tx, Pin.OUT)
+        self.rx = Pin(rx, Pin.IN)
+        
+    def query_mm(self):
+        self.tx.value(0)
+        time.sleep_us(5)
+        self.tx.value(1)
+        time.sleep_us(10)
+        self.tx.value(0)
+        
+        pulse_time = machine.time_pulse_us(self.rx, 1, 500*2*30)
+        if pulse_time < 0:
+            return None
+        
+        return pulse_time * 100 // 582
+    
+
+
+async def ensure_network():
+    from secret import SSID, PASSWORD
+
+    nic = network.WLAN(network.STA_IF)
+    print("[net] setting up WLAN interface...")
+    nic.active(True)
+    while not nic.active():
+        await asyncio.sleep(0.1)
+    print("[net] connecting...")
+    nic.connect(SSID, PASSWORD)
+
+    logged = False
+
+    status_name = {
+        network.STAT_CONNECT_FAIL: "CONNECT_FAIL",
+        network.STAT_CONNECTING: "CONNECTING",
+        network.STAT_IDLE: "IDLE",
+        network.STAT_NO_AP_FOUND: "NO_AP_FOUND",
+        network.STAT_WRONG_PASSWORD: "WRONG_PASSWORD"
+    }
 
-async def read_sensor():
-    log = Logger("tank", Color.CYAN)
-    sensor = UltraSonicSensor(22, 21)
     while True:
-        d = sensor.query_mm()
+        if (status := nic.status()) == network.STAT_GOT_IP:  # pyright: ignore[reportAny]
-        if d is None:
+            STATUS_LED.status(2)
-            log.warning("distance unreadable!")
+            if not logged:
+                logged = True
+                print(f"[net] Got IP: {nic.ifconfig()!r}")
+            rssi = nic.status("rssi")  # pyright: ignore[reportAny]
+            print(f"[net] {rssi=}")
         else:
-            log.info(f"distance: {d}mm")
+            STATUS_LED.status(1)
-        await asyncio.sleep(0.5)
+            print("[net] status={}".format(status_name.get(status, f"UNKNOWN ({status})")))  # pyright: ignore[reportAny]
+            if status == network.STAT_CONNECT_FAIL or status == network.STAT_NO_AP_FOUND:
+                print("[net] retrying in 3s...")
+                await asyncio.sleep(3)
+                nic.connect(SSID, PASSWORD)
+        
+        await asyncio.sleep(1)
 
 
 async def _main():
-    await asyncio.gather(
+    asyncio.create_task(STATUS_LED.task())
-        STATUS_LED.task(),
+    asyncio.create_task(ensure_network())
-        ensure_network(),
+    sensor = UltraSonicSensor(22, 21)
-        read_sensor(),
+    while True:
-        app.start_server(port=80)
+        d = sensor.query_mm()
-    )
+        print(f"d: {d} mm")
-    
+        await asyncio.sleep(0.5)
 
 def main():
     asyncio.run(_main())

pico/src/sensors/ultrasonic.py

@@ -1,28 +0,0 @@
-import machine
-import time
-
-class UltraSonicSensor:
-    """
-    See https://randomnerdtutorials.com/micropython-hc-sr04-ultrasonic-esp32-esp8266/
-    """
-
-    tx: machine.Pin
-    rx: machine.Pin
-
-    def __init__(self, tx: int, rx: int):
-        self.tx = machine.Pin(tx, machine.Pin.OUT)
-        self.rx = machine.Pin(rx, machine.Pin.IN)
-        
-    def query_mm(self):
-        self.tx.value(0)
-        time.sleep_us(5)
-        self.tx.value(1)
-        time.sleep_us(10)
-        self.tx.value(0)
-        
-        pulse_time = machine.time_pulse_us(self.rx, 1, 500*2*30)
-        if pulse_time < 0:
-            return None
-        
-        return pulse_time * 100 // 582
-    

pico/src/statusled.py

@@ -1,34 +0,0 @@
-from machine import Pin
-import asyncio
-
-class StatusLED:
-    led: Pin
-    _status: int
-    _period: float
-    _duty_cycle: float
-
-    def __init__(self):
-        self.led = Pin("LED", Pin.OUT)
-        self._status = 1
-        self._period = 1
-        self._duty_cycle = 0.5
-
-    def status(self, status: int):
-        self._status = status
-        
-    
-    async def task(self):
-        while True:
-            t_1 = self._duty_cycle * self._period
-            t_2 = (1 - self._duty_cycle) * self._period
-
-            t_blink = t_1 / self._status / 2
-            for _ in range(self._status):
-                self.led.value(1)
-                await asyncio.sleep(t_blink)
-                self.led.value(0)
-                await asyncio.sleep(t_blink)
-            
-            await asyncio.sleep(t_2)
-
-STATUS_LED = StatusLED()