monitor: Main modules formatted with black.

Author: peterhinch

v3/as_demos/monitor/README.md

@@ -122,6 +122,8 @@ The decorator positional args are as follows:
  [Pico Pin mapping](./README.md#3-pico-pin-mapping).
  2. `max_instances=1` Defines the maximum number of concurrent instances of the
  task to be independently monitored (default 1).
+ 3. `verbose=True` If `False` suppress the warning which is printed on the host
+ if the instance count exceeds `max_instances`.
 
 Whenever the coroutine runs, a pin on the Pico will go high, and when the code
 terminates it will go low. This enables the behaviour of the system to be

v3/as_demos/monitor/monitor.py

@@ -14,36 +14,43 @@ def _quit(s):
     print("Monitor " + s)
     exit(0)
 
-_write = lambda _ : _quit("must run set_device")
-_dummy = lambda : None  # If UART do nothing.
+
+_write = lambda _: _quit("must run set_device")
+_ifrst = lambda: None  # Reset interface. If UART do nothing.
 
 # For UART pass initialised UART. Baudrate must be 1_000_000.
 # For SPI pass initialised instance SPI. Can be any baudrate, but
 # must be default in other respects.
 def set_device(dev, cspin=None):
     global _write
-    global _dummy
+    global _ifrst
     if isinstance(dev, UART) and cspin is None:  # UART
         _write = dev.write
     elif isinstance(dev, SPI) and isinstance(cspin, Pin):
         cspin(1)
+
         def spiwrite(data):
             cspin(0)
             dev.write(data)
             cspin(1)
+
         _write = spiwrite
+
         def clear_sm():  # Set Pico SM to its initial state
             cspin(1)
             dev.write(b"\0")  # SM is now waiting for CS low.
-        _dummy = clear_sm
+
+        _ifrst = clear_sm
     else:
         _quit("set_device: invalid args.")
 
+
 # Justification for validation even when decorating a method
 # /mnt/qnap2/data/Projects/Python/AssortedTechniques/decorators
 _available = set(range(0, 22))  # Valid idents are 0..21
 _reserved = set()  # Idents reserved for synchronous monitoring
 
+
 def _validate(ident, num=1):
     if ident >= 0 and ident + num < 22:
         try:
@@ -54,83 +61,90 @@ def _validate(ident, num=1):
     else:
         _quit("error - ident {:02d} out of range.".format(ident))
 
+
 # Reserve ID's to be used for synchronous monitoring
 def reserve(*ids):
     for ident in ids:
         _validate(ident)
         _reserved.add(ident)
 
+
 # Check whether a synchronous ident was reserved
 def _check(ident):
     if ident not in _reserved:
         _quit("error: synchronous ident {:02d} was not reserved.".format(ident))
 
+
 # asynchronous monitor
-def asyn(n, max_instances=1):
+def asyn(n, max_instances=1, verbose=True):
     def decorator(coro):
-        # This code runs before asyncio.run()
         _validate(n, max_instances)
         instance = 0
+
         async def wrapped_coro(*args, **kwargs):
-            # realtime
             nonlocal instance
             d = 0x40 + n + min(instance, max_instances - 1)
             v = int.to_bytes(d, 1, "big")
             instance += 1
-            if instance > max_instances:  # Warning only
-                print("Monitor {:02d} max_instances reached.".format(n))
+            if verbose and instance > max_instances:  # Warning only.
+                print("Monitor ident: {:02d} instances: {}.".format(n, instance))
             _write(v)
             try:
                 res = await coro(*args, **kwargs)
             except asyncio.CancelledError:
-                raise
+                raise  # Other exceptions produce traceback.
             finally:
                 d |= 0x20
                 v = int.to_bytes(d, 1, "big")
                 _write(v)
                 instance -= 1
             return res
+
         return wrapped_coro
+
     return decorator
 
+
 # If SPI, clears the state machine in case prior test resulted in the DUT
 # crashing. It does this by sending a byte with CS\ False (high).
 def init():
-    _dummy()  # Does nothing if UART
+    _ifrst()  # Reset interface. Does nothing if UART.
     _write(b"z")  # Clear Pico's instance counters etc.
 
+
 # Optionally run this to show up periods of blocking behaviour
-async def hog_detect(i=1, s=(b"\x40", b"\x60")):
+async def hog_detect(s=(b"\x40", b"\x60")):
     while True:
-        _write(s[(i := i ^ 1)])
-        await asyncio.sleep_ms(0)
+        for v in s:
+            _write(v)
+            await asyncio.sleep_ms(0)
+
 
 # Monitor a synchronous function definition
 def sync(n):
     def decorator(func):
         _validate(n)
-        dstart = 0x40 + n
-        vstart = int.to_bytes(dstart, 1, "big")
-        dend = 0x60 + n
-        vend = int.to_bytes(dend, 1, "big")
+        vstart = int.to_bytes(0x40 + n, 1, "big")
+        vend = int.to_bytes(0x60 + n, 1, "big")
+
         def wrapped_func(*args, **kwargs):
             _write(vstart)
             res = func(*args, **kwargs)
             _write(vend)
             return res
+
         return wrapped_func
+
     return decorator
 
+
 # Runtime monitoring: can't validate because code may be looping.
 # Monitor a synchronous function call
 class mon_call:
     def __init__(self, n):
         _check(n)
-        self.n = n
-        self.dstart = 0x40 + n
-        self.vstart = int.to_bytes(self.dstart, 1, "big")
-        self.dend = 0x60 + n
-        self.vend = int.to_bytes(self.dend, 1, "big")
+        self.vstart = int.to_bytes(0x40 + n, 1, "big")
+        self.vend = int.to_bytes(0x60 + n, 1, "big")
 
     def __enter__(self):
         _write(self.vstart)
@@ -140,6 +154,7 @@ def __exit__(self, type, value, traceback):
         _write(self.vend)
         return False  # Don't silence exceptions
 
+
 # Cause pico ident n to produce a brief (~80μs) pulse
 def trigger(n):
     _check(n)

v3/as_demos/monitor/monitor_pico.py

@@ -35,19 +35,23 @@ def spi_in():
 
 
 class PIOSPI:
-
     def __init__(self):
-        self._sm = rp2.StateMachine(0, spi_in,
-                                    in_shiftdir=rp2.PIO.SHIFT_LEFT,
-                                    push_thresh=8, in_base=Pin(0),
-                                    jmp_pin=Pin(2, Pin.IN, Pin.PULL_UP))
+        self._sm = rp2.StateMachine(
+            0,
+            spi_in,
+            in_shiftdir=rp2.PIO.SHIFT_LEFT,
+            push_thresh=8,
+            in_base=Pin(0),
+            jmp_pin=Pin(2, Pin.IN, Pin.PULL_UP),
+        )
         self._sm.active(1)
 
     # Blocking read of 1 char. Returns ord(ch). If DUT crashes, worst case
     # is where CS is left low. SM will hang until user restarts. On restart
     # the app
     def read(self):
-        return self._sm.get() & 0xff
+        return self._sm.get() & 0xFF
+
 
 # ****** Define pins ******
 
@@ -64,11 +68,14 @@ def read(self):
 # ****** Timing *****
 
 pin_t = Pin(28, Pin.OUT)
+
+
 def _cb(_):
     pin_t(1)
     print("Timeout.")
     pin_t(0)
 
+
 tim = Timer()
 
 # ****** Monitor ******
@@ -83,38 +90,42 @@ def _cb(_):
 
 # native reduced latency to 10μs but killed the hog detector: timer never timed out.
 # Also locked up Pico so ctrl-c did not interrupt.
-#@micropython.native
+# @micropython.native
 def run(period=100, verbose=(), device="uart", vb=True):
     if isinstance(period, int):
         t_ms = period
         mode = SOON
     else:
         t_ms, mode = period
         if mode not in (SOON, LATE, MAX):
-            raise ValueError('Invalid mode.')
+            raise ValueError("Invalid mode.")
     for x in verbose:
         pins[x][2] = True
     # A device must support a blocking read.
     if device == "uart":
         uart = UART(0, 1_000_000)  # rx on GPIO 1
+
         def read():
             while not uart.any():  # Prevent UART timeouts
                 pass
             return ord(uart.read(1))
+
     elif device == "spi":
         pio = PIOSPI()
+
         def read():
             return pio.read()
+
     else:
         raise ValueError("Unsupported device:", device)
 
-    vb and print('Awaiting communication')
+    vb and print("Awaiting communication.")
     h_max = 0  # Max hog duration (ms)
     h_start = 0  # Absolute hog start time
     while True:
         if x := read():  # Get an initial 0 on UART
-            if x == 0x7a:  # Init: program under test has restarted
-                vb and print('Got communication.')
+            if x == 0x7A:  # Init: program under test has restarted
+                vb and print("Got communication.")
                 h_max = 0  # Restart timing
                 h_start = 0
                 for pin in pins:
@@ -140,7 +151,7 @@ def read():
                                 pin_t(1)
                                 pin_t(0)
                 h_start = t
-            p = pins[x & 0x1f]  # Key: 0x40 (ord('@')) is pin ID 0
+            p = pins[x & 0x1F]  # Key: 0x40 (ord('@')) is pin ID 0
             if x & 0x20:  # Going down
                 p[1] -= 1
                 if not p[1]:  # Instance count is zero
@@ -149,4 +160,4 @@ def read():
                 p[0](1)
                 p[1] += 1
             if p[2]:
-                print(f'ident {i} count {p[1]}')
+                print(f"ident {i} count {p[1]}")