Forgot to add the actual asyncgpio.test module. Again. Sigh.

Author: smurfix

asyncgpio/test.py

@@ -0,0 +1,197 @@
+"""
+This module contains helpers for testing async gpio, via the Linux kernel's
+``gpio_mockup`` module (writing) and ``/sys/kernel/debug/cpio`` (monitoring).
+"""
+
+import os
+import re
+import anyio
+import logging
+import errno
+
+from contextlib import asynccontextmanager
+from collections import namedtuple
+
+logger = logging.getLogger(__name__)
+
+_r_chip = re.compile(
+    "^(?P<chip>[a-z0-9]+): GPIOs (?P<base>[0-9]+)-(?:.*, (?P<name>[-_a-zA-Z0-9]+): *$)?"
+)
+_r_pin = re.compile("^gpio-(?P<pin>[0-9]+) \\(.*\\) (?P<dir>in|out) +(?P<val>hi|lo)")
+
+Pin = namedtuple("Pin",["out","level"])
+
+class _GpioPin:
+    """
+    Code representing one GPIO pin.
+    """
+
+    fd = None
+
+    def __init__(self, watcher: "GpioWatcher", chip: str, pin: int):
+        self.watcher = watcher
+        self.chip = chip
+        self.pin = pin
+        self.mon = set()
+        self.state = (None, None)
+        try:
+            self.fd = os.open(
+                os.path.join(watcher.debugfs_path, "gpio-mockup-event", chip, str(pin)),
+                os.O_WRONLY,
+            )
+        except EnvironmentError as exc:
+            if exc.errno != errno.ENOENT:
+                raise
+
+    def __del__(self):
+        if self.fd is not None:
+            os.close(self.fd)
+            del self.fd
+
+    @asynccontextmanager
+    async def watch(self):
+        """
+        An async context manager that returns an iterator for changes of
+        this pin.
+
+        Values are (out,level) tuples of bool, with "out" and "high"
+        represented as True.
+        """
+        q = anyio.create_queue(10)
+        self.mon.add(q)
+        try:
+            yield q
+        finally:
+            self.mon.remove(q)
+
+    async def see(self, write: bool, level: bool):
+        s = (write, level)
+        if self.state == s:
+            return
+        self.state = s
+        logger.debug("SEE %s %d %s", self.chip, self.pin, self.state)
+        for cb in list(self.mon):
+            await cb.put(s)
+
+    def set(self, value: bool):
+        logger.debug("SET %s %d %s", self.chip, self.pin, value)
+        if self.fd is None:
+            raise RuntimeError("Pin %s/%d is not controlled via the 'gpio_mockup' module" % (self.chip,self.pin))
+        os.write(self.fd, b"1" if value else b"0")
+        # os.lseek(self.fd, 0, os.SEEK_SET)
+
+
+class GpioWatcher:
+    """
+    Code which triggers callbacks whenever a GPIO pin changes.
+
+    This class polls `/sys/kernel/debug/gpio` (can be overridden).
+    """
+
+    def __init__(
+        self,
+        interval: float = 0.2,
+        debugfs_path: str = "/sys/kernel/debug",
+        sysfs_path: str = "/sys",
+    ):
+        self.interval = interval
+        self.gpio = open(os.path.join(debugfs_path, "gpio"), "r")
+        self.targets = dict()  # chip > line > _GpioPin
+        #       self.names = {}
+        self.sysfs_path = sysfs_path
+        self.debugfs_path = debugfs_path
+        gpio_dir = os.path.join(sysfs_path, "class", "gpio")
+
+    #       for d in os.listdir(gpio_dir):
+    #           try:
+    #               with open(os.path.join(gpio_dir,d,"label"),"r") as f:
+    #                   n = f.read().strip()
+    #           except EnvironmentError as e:
+    #               if e.errno == errno.ENOTDIR:
+    #                   continue
+    #               raise
+    #           else:
+    #               self.names[d] = n
+
+    def monitor(self, chip: str, pin: int):
+        """
+        Shortcut for 'self.pin(chip, pin).watch()'.
+        """
+        return self.pin(chip, pin).watch()
+
+    def pin(self, chip: str, pin: int, create: bool = True):
+        """
+        Returns a pins corresponding GpioPin
+        """
+        #       chip = self.names[chip]
+        try:
+            c = self.targets[chip]
+        except KeyError:
+            if not create:
+                raise
+            self.targets[chip] = c = dict()
+        try:
+            p = c[pin]
+        except KeyError:
+            if not create:
+                raise
+            c[pin] = p = _GpioPin(self, chip, pin)
+        return p
+
+    async def _watch(self):
+        # The actual monitor.
+        while True:
+            await self.check_pins()
+            await anyio.sleep(self.interval)
+
+    async def check_pins(self):
+        """
+        Read the GPIO debug file and update pin states
+        """
+        chip = None
+        base = None
+
+        for l in self.gpio:
+            l = l.strip()
+            if not l:
+                chip = None
+                continue
+            if chip is None:
+                r = _r_chip.match(l)
+                if not r:
+                    raise ValueError(l)
+                chip = r.group("name")
+                if not chip:
+                    chip = r.group("chip")
+                base = int(r.group("base"))
+            else:
+                r = _r_pin.match(l)
+                if not r:
+                    breakpoint()
+                    raise ValueError(l)
+                pin = int(r.group("pin")) - base
+                out = r.group("dir") == "out"
+                val = r.group("val") == "hi"
+
+                try:
+                    pin = self.pin(chip, pin, create=False)
+                except KeyError:
+                    pass
+                else:
+                    await pin.see(out, val)
+        self.gpio.seek(0)
+
+    @asynccontextmanager
+    async def run(self):
+        """
+        This async context manager controls the monitoring loop.
+        """
+        async with anyio.create_task_group() as tg:
+            self.tg = tg
+            await tg.spawn(self._watch)
+            try:
+                yield self
+            finally:
+                self.tg = None
+                await tg.cancel_scope.cancel()
+