test(led-driver): add espnow peer and ap pm0 scripts

Made-with: Cursor

docs/pattern-contract.md

@@ -0,0 +1,51 @@
+# Pattern Contract (Important)
+
+Pattern classes are loaded dynamically by `Presets._load_dynamic_patterns()`.
+
+Patterns must follow this contract exactly.
+
+## Required class shape
+
+- File name is the pattern id (for example `blink.py` -> pattern name `blink`).
+- Module exports a class with:
+  - `__init__(self, driver)` where `driver` is the `Presets` instance.
+  - `run(self, preset)` that returns a generator.
+
+`Presets` binds patterns like this:
+
+- `pattern_class(self).run`
+- then calls `self.patterns[preset.p](preset)` and stores that generator.
+- every frame, `Presets.tick()` does `next(self.generator)`.
+
+## `run()` generator rules
+
+- `run()` must `yield` frequently (normally once per tick loop).
+- Do not block inside `run()`:
+  - no `sleep()` / `sleep_ms()` / long loops without `yield`.
+  - no network or file I/O.
+- Use time checks (`utime.ticks_ms()` + `utime.ticks_diff(...)`) to schedule updates.
+- Keep pattern state inside local variables in `run()` (or object fields if needed).
+
+## Drawing and brightness
+
+- Use `self.driver.apply_brightness(color, preset.b)` for per-preset brightness.
+- Write pixels through `self.driver.n[...]` / `self.driver.n.fill(...)`.
+- Flush frame with `self.driver.n.write()`.
+- If a pattern needs to clear, use black `(0, 0, 0)`.
+
+## Step semantics
+
+- `self.driver.step` is shared pattern state managed by `Presets.select(...)` and patterns.
+- Patterns that use step-based progression should update `self.driver.step` themselves.
+- `select(..., step=...)` may set an explicit starting step.
+
+## Error handling
+
+- Let unexpected errors raise inside the generator.
+- `Presets.tick()` catches exceptions, logs, and stops the active generator.
+- Pattern code should not swallow broad exceptions unless there is a clear recovery path.
+
+## Built-ins
+
+- `off` and `on` are built-in methods on `Presets`, not loaded from this folder.
+- `__init__.py` is ignored by dynamic loader.

lib/microdot/__init__.py

@@ -0,0 +1,2 @@
+from microdot.microdot import Microdot, Request, Response, abort, redirect, \
+    send_file  # noqa: F401

lib/microdot/helpers.py

@@ -0,0 +1,8 @@
+try:
+    from functools import wraps
+except ImportError:  # pragma: no cover
+    # MicroPython does not currently implement functools.wraps
+    def wraps(wrapped):
+        def _(wrapper):
+            return wrapper
+        return _

lib/microdot/session.py

@@ -0,0 +1,225 @@
+try:
+    import jwt
+    HAS_JWT = True
+except ImportError:
+    HAS_JWT = False
+    try:
+        import ubinascii
+    except ImportError:
+        import binascii as ubinascii
+    try:
+        import uhashlib as hashlib
+    except ImportError:
+        import hashlib
+    try:
+        import uhmac as hmac
+    except ImportError:
+        try:
+            import hmac
+        except ImportError:
+            hmac = None
+    import json
+
+from microdot.microdot import invoke_handler
+from microdot.helpers import wraps
+
+
+class SessionDict(dict):
+    """A session dictionary.
+
+    The session dictionary is a standard Python dictionary that has been
+    extended with convenience ``save()`` and ``delete()`` methods.
+    """
+    def __init__(self, request, session_dict):
+        super().__init__(session_dict)
+        self.request = request
+
+    def save(self):
+        """Update the session cookie."""
+        self.request.app._session.update(self.request, self)
+
+    def delete(self):
+        """Delete the session cookie."""
+        self.request.app._session.delete(self.request)
+
+
+class Session:
+    """Session handling
+
+    :param app: The application instance.
+    :param secret_key: The secret key, as a string or bytes object.
+    :param cookie_options: A dictionary with cookie options to pass as
+                           arguments to :meth:`Response.set_cookie()
+                           <microdot.Response.set_cookie>`.
+    """
+    secret_key = None
+
+    def __init__(self, app=None, secret_key=None, cookie_options=None):
+        self.secret_key = secret_key
+        self.cookie_options = cookie_options or {}
+        if app is not None:
+            self.initialize(app)
+
+    def initialize(self, app, secret_key=None, cookie_options=None):
+        if secret_key is not None:
+            self.secret_key = secret_key
+        if cookie_options is not None:
+            self.cookie_options = cookie_options
+        if 'path' not in self.cookie_options:
+            self.cookie_options['path'] = '/'
+        if 'http_only' not in self.cookie_options:
+            self.cookie_options['http_only'] = True
+        app._session = self
+
+    def get(self, request):
+        """Retrieve the user session.
+
+        :param request: The client request.
+
+        The return value is a session dictionary with the data stored in the
+        user's session, or ``{}`` if the session data is not available or
+        invalid.
+        """
+        if not self.secret_key:
+            raise ValueError('The session secret key is not configured')
+        if hasattr(request.g, '_session'):
+            return request.g._session
+        session = request.cookies.get('session')
+        if session is None:
+            request.g._session = SessionDict(request, {})
+            return request.g._session
+        request.g._session = SessionDict(request, self.decode(session))
+        return request.g._session
+
+    def update(self, request, session):
+        """Update the user session.
+
+        :param request: The client request.
+        :param session: A dictionary with the update session data for the user.
+
+        Applications would normally not call this method directly, instead they
+        would use the :meth:`SessionDict.save` method on the session
+        dictionary, which calls this method. For example::
+
+            @app.route('/')
+            @with_session
+            def index(request, session):
+                session['foo'] = 'bar'
+                session.save()
+                return 'Hello, World!'
+
+        Calling this method adds a cookie with the updated session to the
+        request currently being processed.
+        """
+        if not self.secret_key:
+            raise ValueError('The session secret key is not configured')
+
+        encoded_session = self.encode(session)
+
+        @request.after_request
+        def _update_session(request, response):
+            response.set_cookie('session', encoded_session,
+                                **self.cookie_options)
+            return response
+
+    def delete(self, request):
+        """Remove the user session.
+
+        :param request: The client request.
+
+        Applications would normally not call this method directly, instead they
+        would use the :meth:`SessionDict.delete` method on the session
+        dictionary, which calls this method. For example::
+
+            @app.route('/')
+            @with_session
+            def index(request, session):
+                session.delete()
+                return 'Hello, World!'
+
+        Calling this method adds a cookie removal header to the request
+        currently being processed.
+        """
+        @request.after_request
+        def _delete_session(request, response):
+            response.delete_cookie('session', **self.cookie_options)
+            return response
+
+    def encode(self, payload, secret_key=None):
+        """Encode session data using JWT if available, otherwise use simple HMAC."""
+        if HAS_JWT:
+            return jwt.encode(payload, secret_key or self.secret_key,
+                              algorithm='HS256')
+        else:
+            # Simple encoding for MicroPython: base64(json) + HMAC signature
+            key = (secret_key or self.secret_key).encode() if isinstance(secret_key or self.secret_key, str) else (secret_key or self.secret_key)
+            payload_json = json.dumps(payload)
+            payload_b64 = ubinascii.b2a_base64(payload_json.encode()).decode().strip()
+            
+            # Create HMAC signature
+            if hmac:
+                # Use hmac module if available
+                h = hmac.new(key, payload_json.encode(), hashlib.sha256)
+            else:
+                # Fallback: simple SHA256(key + message)
+                h = hashlib.sha256(key + payload_json.encode())
+            signature = ubinascii.b2a_base64(h.digest()).decode().strip()
+            
+            return f"{payload_b64}.{signature}"
+
+    def decode(self, session, secret_key=None):
+        """Decode session data using JWT if available, otherwise use simple HMAC."""
+        if HAS_JWT:
+            try:
+                payload = jwt.decode(session, secret_key or self.secret_key,
+                                     algorithms=['HS256'])
+            except jwt.exceptions.PyJWTError:  # pragma: no cover
+                return {}
+            return payload
+        else:
+            try:
+                # Simple decoding for MicroPython
+                if '.' not in session:
+                    return {}
+                
+                payload_b64, signature = session.rsplit('.', 1)
+                payload_json = ubinascii.a2b_base64(payload_b64).decode()
+                
+                # Verify HMAC signature
+                key = (secret_key or self.secret_key).encode() if isinstance(secret_key or self.secret_key, str) else (secret_key or self.secret_key)
+                if hmac:
+                    # Use hmac module if available
+                    h = hmac.new(key, payload_json.encode(), hashlib.sha256)
+                else:
+                    # Fallback: simple SHA256(key + message)
+                    h = hashlib.sha256(key + payload_json.encode())
+                expected_signature = ubinascii.b2a_base64(h.digest()).decode().strip()
+                
+                if signature != expected_signature:
+                    return {}
+                
+                return json.loads(payload_json)
+            except Exception:
+                return {}
+
+
+def with_session(f):
+    """Decorator that passes the user session to the route handler.
+
+    The session dictionary is passed to the decorated function as an argument
+    after the request object. Example::
+
+        @app.route('/')
+        @with_session
+        def index(request, session):
+            return 'Hello, World!'
+
+    Note that the decorator does not save the session. To update the session,
+    call the :func:`session.save() <microdot.session.SessionDict.save>` method.
+    """
+    @wraps(f)
+    async def wrapper(request, *args, **kwargs):
+        return await invoke_handler(
+            f, request, request.app._session.get(request), *args, **kwargs)
+
+    return wrapper

lib/microdot/utemplate.py

@@ -0,0 +1,70 @@
+from utemplate import recompile
+
+_loader = None
+
+
+class Template:
+    """A template object.
+
+    :param template: The filename of the template to render, relative to the
+                     configured template directory.
+    """
+    @classmethod
+    def initialize(cls, template_dir='templates',
+                   loader_class=recompile.Loader):
+        """Initialize the templating subsystem.
+
+        :param template_dir: the directory where templates are stored. This
+                             argument is optional. The default is to load
+                             templates from a *templates* subdirectory.
+        :param loader_class: the ``utemplate.Loader`` class to use when loading
+                             templates. This argument is optional. The default
+                             is the ``recompile.Loader`` class, which
+                             automatically recompiles templates when they
+                             change.
+        """
+        global _loader
+        _loader = loader_class(None, template_dir)
+
+    def __init__(self, template):
+        if _loader is None:  # pragma: no cover
+            self.initialize()
+        #: The name of the template
+        self.name = template
+        self.template = _loader.load(template)
+
+    def generate(self, *args, **kwargs):
+        """Return a generator that renders the template in chunks, with the
+        given arguments."""
+        return self.template(*args, **kwargs)
+
+    def render(self, *args, **kwargs):
+        """Render the template with the given arguments and return it as a
+        string."""
+        return ''.join(self.generate(*args, **kwargs))
+
+    def generate_async(self, *args, **kwargs):
+        """Return an asynchronous generator that renders the template in
+        chunks, using the given arguments."""
+        class sync_to_async_iter():
+            def __init__(self, iter):
+                self.iter = iter
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                try:
+                    return next(self.iter)
+                except StopIteration:
+                    raise StopAsyncIteration
+
+        return sync_to_async_iter(self.generate(*args, **kwargs))
+
+    async def render_async(self, *args, **kwargs):
+        """Render the template with the given arguments asynchronously and
+        return it as a string."""
+        response = ''
+        async for chunk in self.generate_async(*args, **kwargs):
+            response += chunk
+        return response

lib/microdot/websocket.py

@@ -0,0 +1,231 @@
+import binascii
+import hashlib
+from microdot import Request, Response
+from microdot.microdot import MUTED_SOCKET_ERRORS, print_exception
+from microdot.helpers import wraps
+
+
+class WebSocketError(Exception):
+    """Exception raised when an error occurs in a WebSocket connection."""
+    pass
+
+
+class WebSocket:
+    """A WebSocket connection object.
+
+    An instance of this class is sent to handler functions to manage the
+    WebSocket connection.
+    """
+    CONT = 0
+    TEXT = 1
+    BINARY = 2
+    CLOSE = 8
+    PING = 9
+    PONG = 10
+
+    #: Specify the maximum message size that can be received when calling the
+    #: ``receive()`` method. Messages with payloads that are larger than this
+    #: size will be rejected and the connection closed. Set to 0 to disable
+    #: the size check (be aware of potential security issues if you do this),
+    #: or to -1 to use the value set in
+    #: ``Request.max_body_length``. The default is -1.
+    #:
+    #: Example::
+    #:
+    #:    WebSocket.max_message_length = 4 * 1024  # up to 4KB messages
+    max_message_length = -1
+
+    def __init__(self, request):
+        self.request = request
+        self.closed = False
+
+    async def handshake(self):
+        response = self._handshake_response()
+        await self.request.sock[1].awrite(
+            b'HTTP/1.1 101 Switching Protocols\r\n')
+        await self.request.sock[1].awrite(b'Upgrade: websocket\r\n')
+        await self.request.sock[1].awrite(b'Connection: Upgrade\r\n')
+        await self.request.sock[1].awrite(
+            b'Sec-WebSocket-Accept: ' + response + b'\r\n\r\n')
+
+    async def receive(self):
+        """Receive a message from the client."""
+        while True:
+            opcode, payload = await self._read_frame()
+            send_opcode, data = self._process_websocket_frame(opcode, payload)
+            if send_opcode:  # pragma: no cover
+                await self.send(data, send_opcode)
+            elif data:  # pragma: no branch
+                return data
+
+    async def send(self, data, opcode=None):
+        """Send a message to the client.
+
+        :param data: the data to send, given as a string or bytes.
+        :param opcode: a custom frame opcode to use. If not given, the opcode
+                       is ``TEXT`` or ``BINARY`` depending on the type of the
+                       data.
+        """
+        frame = self._encode_websocket_frame(
+            opcode or (self.TEXT if isinstance(data, str) else self.BINARY),
+            data)
+        await self.request.sock[1].awrite(frame)
+
+    async def close(self):
+        """Close the websocket connection."""
+        if not self.closed:  # pragma: no cover
+            self.closed = True
+            await self.send(b'', self.CLOSE)
+
+    def _handshake_response(self):
+        connection = False
+        upgrade = False
+        websocket_key = None
+        for header, value in self.request.headers.items():
+            h = header.lower()
+            if h == 'connection':
+                connection = True
+                if 'upgrade' not in value.lower():
+                    return self.request.app.abort(400)
+            elif h == 'upgrade':
+                upgrade = True
+                if not value.lower() == 'websocket':
+                    return self.request.app.abort(400)
+            elif h == 'sec-websocket-key':
+                websocket_key = value
+        if not connection or not upgrade or not websocket_key:
+            return self.request.app.abort(400)
+        d = hashlib.sha1(websocket_key.encode())
+        d.update(b'258EAFA5-E914-47DA-95CA-C5AB0DC85B11')
+        return binascii.b2a_base64(d.digest())[:-1]
+
+    @classmethod
+    def _parse_frame_header(cls, header):
+        fin = header[0] & 0x80
+        opcode = header[0] & 0x0f
+        if fin == 0 or opcode == cls.CONT:  # pragma: no cover
+            raise WebSocketError('Continuation frames not supported')
+        has_mask = header[1] & 0x80
+        length = header[1] & 0x7f
+        if length == 126:
+            length = -2
+        elif length == 127:
+            length = -8
+        return fin, opcode, has_mask, length
+
+    def _process_websocket_frame(self, opcode, payload):
+        if opcode == self.TEXT:
+            payload = payload.decode()
+        elif opcode == self.BINARY:
+            pass
+        elif opcode == self.CLOSE:
+            raise WebSocketError('Websocket connection closed')
+        elif opcode == self.PING:
+            return self.PONG, payload
+        elif opcode == self.PONG:  # pragma: no branch
+            return None, None
+        return None, payload
+
+    @classmethod
+    def _encode_websocket_frame(cls, opcode, payload):
+        frame = bytearray()
+        frame.append(0x80 | opcode)
+        if opcode == cls.TEXT:
+            payload = payload.encode()
+        if len(payload) < 126:
+            frame.append(len(payload))
+        elif len(payload) < (1 << 16):
+            frame.append(126)
+            frame.extend(len(payload).to_bytes(2, 'big'))
+        else:
+            frame.append(127)
+            frame.extend(len(payload).to_bytes(8, 'big'))
+        frame.extend(payload)
+        return frame
+
+    async def _read_frame(self):
+        header = await self.request.sock[0].read(2)
+        if len(header) != 2:  # pragma: no cover
+            raise WebSocketError('Websocket connection closed')
+        fin, opcode, has_mask, length = self._parse_frame_header(header)
+        if length == -2:
+            length = await self.request.sock[0].read(2)
+            length = int.from_bytes(length, 'big')
+        elif length == -8:
+            length = await self.request.sock[0].read(8)
+            length = int.from_bytes(length, 'big')
+        max_allowed_length = Request.max_body_length \
+            if self.max_message_length == -1 else self.max_message_length
+        if length > max_allowed_length:
+            raise WebSocketError('Message too large')
+        if has_mask:  # pragma: no cover
+            mask = await self.request.sock[0].read(4)
+        payload = await self.request.sock[0].read(length)
+        if has_mask:  # pragma: no cover
+            payload = bytes(x ^ mask[i % 4] for i, x in enumerate(payload))
+        return opcode, payload
+
+
+async def websocket_upgrade(request):
+    """Upgrade a request handler to a websocket connection.
+
+    This function can be called directly inside a route function to process a
+    WebSocket upgrade handshake, for example after the user's credentials are
+    verified. The function returns the websocket object::
+
+        @app.route('/echo')
+        async def echo(request):
+            if not authenticate_user(request):
+                abort(401)
+            ws = await websocket_upgrade(request)
+            while True:
+                message = await ws.receive()
+                await ws.send(message)
+    """
+    ws = WebSocket(request)
+    await ws.handshake()
+
+    @request.after_request
+    async def after_request(request, response):
+        return Response.already_handled
+
+    return ws
+
+
+def websocket_wrapper(f, upgrade_function):
+    @wraps(f)
+    async def wrapper(request, *args, **kwargs):
+        ws = await upgrade_function(request)
+        try:
+            await f(request, ws, *args, **kwargs)
+        except OSError as exc:
+            if exc.errno not in MUTED_SOCKET_ERRORS:  # pragma: no cover
+                raise
+        except WebSocketError:
+            pass
+        except Exception as exc:
+            print_exception(exc)
+        finally:  # pragma: no cover
+            try:
+                await ws.close()
+            except Exception:
+                pass
+        return Response.already_handled
+    return wrapper
+
+
+def with_websocket(f):
+    """Decorator to make a route a WebSocket endpoint.
+
+    This decorator is used to define a route that accepts websocket
+    connections. The route then receives a websocket object as a second
+    argument that it can use to send and receive messages::
+
+        @app.route('/echo')
+        @with_websocket
+        async def echo(request, ws):
+            while True:
+                message = await ws.receive()
+                await ws.send(message)
+    """
+    return websocket_wrapper(f, websocket_upgrade)

lib/utemplate/compiled.py

@@ -0,0 +1,14 @@
+class Loader:
+
+    def __init__(self, pkg, dir):
+        if dir == ".":
+            dir = ""
+        else:
+            dir = dir.replace("/", ".") + "."
+        if pkg and pkg != "__main__":
+            dir = pkg + "." + dir
+        self.p = dir
+
+    def load(self, name):
+        name = name.replace(".", "_")
+        return __import__(self.p + name, None, None, (name,)).render

lib/utemplate/recompile.py

@@ -0,0 +1,21 @@
+# (c) 2014-2020 Paul Sokolovsky. MIT license.
+try:
+    from uos import stat, remove
+except:
+    from os import stat, remove
+from . import source
+
+
+class Loader(source.Loader):
+
+    def load(self, name):
+        o_path = self.pkg_path + self.compiled_path(name)
+        i_path = self.pkg_path + self.dir + "/" + name
+        try:
+            o_stat = stat(o_path)
+            i_stat = stat(i_path)
+            if i_stat[8] > o_stat[8]:
+                # input file is newer, remove output to force recompile
+                remove(o_path)
+        finally:
+            return super().load(name)

lib/utemplate/source.py

@@ -0,0 +1,188 @@
+# (c) 2014-2019 Paul Sokolovsky. MIT license.
+from . import compiled
+
+
+class Compiler:
+
+    START_CHAR = "{"
+    STMNT = "%"
+    STMNT_END = "%}"
+    EXPR = "{"
+    EXPR_END = "}}"
+
+    def __init__(self, file_in, file_out, indent=0, seq=0, loader=None):
+        self.file_in = file_in
+        self.file_out = file_out
+        self.loader = loader
+        self.seq = seq
+        self._indent = indent
+        self.stack = []
+        self.in_literal = False
+        self.flushed_header = False
+        self.args = "*a, **d"
+
+    def indent(self, adjust=0):
+        if not self.flushed_header:
+            self.flushed_header = True
+            self.indent()
+            self.file_out.write("def render%s(%s):\n" % (str(self.seq) if self.seq else "", self.args))
+            self.stack.append("def")
+        self.file_out.write("    " * (len(self.stack) + self._indent + adjust))
+
+    def literal(self, s):
+        if not s:
+            return
+        if not self.in_literal:
+            self.indent()
+            self.file_out.write('yield """')
+            self.in_literal = True
+        self.file_out.write(s.replace('"', '\\"'))
+
+    def close_literal(self):
+        if self.in_literal:
+            self.file_out.write('"""\n')
+        self.in_literal = False
+
+    def render_expr(self, e):
+        self.indent()
+        self.file_out.write('yield str(' + e + ')\n')
+
+    def parse_statement(self, stmt):
+        tokens = stmt.split(None, 1)
+        if tokens[0] == "args":
+            if len(tokens) > 1:
+                self.args = tokens[1]
+            else:
+                self.args = ""
+        elif tokens[0] == "set":
+            self.indent()
+            self.file_out.write(stmt[3:].strip() + "\n")
+        elif tokens[0] == "include":
+            if not self.flushed_header:
+                # If there was no other output, we still need a header now
+                self.indent()
+            tokens = tokens[1].split(None, 1)
+            args = ""
+            if len(tokens) > 1:
+                args = tokens[1]
+            if tokens[0][0] == "{":
+                self.indent()
+                # "1" as fromlist param is uPy hack
+                self.file_out.write('_ = __import__(%s.replace(".", "_"), None, None, 1)\n' % tokens[0][2:-2])
+                self.indent()
+                self.file_out.write("yield from _.render(%s)\n" % args)
+                return
+
+            with self.loader.input_open(tokens[0][1:-1]) as inc:
+                self.seq += 1
+                c = Compiler(inc, self.file_out, len(self.stack) + self._indent, self.seq)
+                inc_id = self.seq
+                self.seq = c.compile()
+            self.indent()
+            self.file_out.write("yield from render%d(%s)\n" % (inc_id, args))
+        elif len(tokens) > 1:
+            if tokens[0] == "elif":
+                assert self.stack[-1] == "if"
+                self.indent(-1)
+                self.file_out.write(stmt + ":\n")
+            else:
+                self.indent()
+                self.file_out.write(stmt + ":\n")
+                self.stack.append(tokens[0])
+        else:
+            if stmt.startswith("end"):
+                assert self.stack[-1] == stmt[3:]
+                self.stack.pop(-1)
+            elif stmt == "else":
+                assert self.stack[-1] == "if"
+                self.indent(-1)
+                self.file_out.write("else:\n")
+            else:
+                assert False
+
+    def parse_line(self, l):
+        while l:
+            start = l.find(self.START_CHAR)
+            if start == -1:
+                self.literal(l)
+                return
+            self.literal(l[:start])
+            self.close_literal()
+            sel = l[start + 1]
+            #print("*%s=%s=" % (sel, EXPR))
+            if sel == self.STMNT:
+                end = l.find(self.STMNT_END)
+                assert end > 0
+                stmt = l[start + len(self.START_CHAR + self.STMNT):end].strip()
+                self.parse_statement(stmt)
+                end += len(self.STMNT_END)
+                l = l[end:]
+                if not self.in_literal and l == "\n":
+                    break
+            elif sel == self.EXPR:
+    #            print("EXPR")
+                end = l.find(self.EXPR_END)
+                assert end > 0
+                expr = l[start + len(self.START_CHAR + self.EXPR):end].strip()
+                self.render_expr(expr)
+                end += len(self.EXPR_END)
+                l = l[end:]
+            else:
+                self.literal(l[start])
+                l = l[start + 1:]
+
+    def header(self):
+        self.file_out.write("# Autogenerated file\n")
+
+    def compile(self):
+        self.header()
+        for l in self.file_in:
+            self.parse_line(l)
+        self.close_literal()
+        return self.seq
+
+
+class Loader(compiled.Loader):
+
+    def __init__(self, pkg, dir):
+        super().__init__(pkg, dir)
+        self.dir = dir
+        if pkg == "__main__":
+            # if pkg isn't really a package, don't bother to use it
+            # it means we're running from "filesystem directory", not
+            # from a package.
+            pkg = None
+
+        self.pkg_path = ""
+        if pkg:
+            p = __import__(pkg)
+            if isinstance(p.__path__, str):
+                # uPy
+                self.pkg_path = p.__path__
+            else:
+                # CPy
+                self.pkg_path = p.__path__[0]
+            self.pkg_path += "/"
+
+    def input_open(self, template):
+        path = self.pkg_path + self.dir + "/" + template
+        return open(path)
+
+    def compiled_path(self, template):
+        return self.dir + "/" + template.replace(".", "_") + ".py"
+
+    def load(self, name):
+        try:
+            return super().load(name)
+        except (OSError, ImportError):
+            pass
+
+        compiled_path = self.pkg_path + self.compiled_path(name)
+
+        f_in = self.input_open(name)
+        f_out = open(compiled_path, "w")
+        c = Compiler(f_in, f_out, loader=self)
+        c.compile()
+        f_in.close()
+        f_out.close()
+        return super().load(name)

presets.json

@@ -0,0 +1 @@
+{"15": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 0, 0], [0, 255, 0]], "b": 255, "n2": 0, "n1": 0, "p": "blink", "n3": 0, "d": 500}, "40": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 255, 0]], "b": 255, "n2": 2600, "n1": 35, "p": "flame", "n3": 0, "d": 50}, "41": {"n5": 0, "n4": 5, "a": true, "n6": 0, "c": [[120, 200, 255], [80, 140, 255], [180, 120, 255], [100, 220, 232], [160, 200, 255]], "b": 255, "n2": 10, "n1": 72, "p": "twinkle", "n3": 5, "d": 500}, "42": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[166, 0, 255], [0, 10, 10]], "b": 255, "n2": 900, "n1": 30, "p": "radiate", "n3": 4000, "d": 5000}, "6": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[0, 255, 0]], "b": 255, "n2": 500, "n1": 1000, "p": "pulse", "n3": 1000, "d": 500}, "10": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[230, 242, 255]], "b": 200, "n2": 0, "n1": 0, "p": "on", "n3": 0, "d": 100}, "13": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 255, 255]], "b": 255, "n2": 0, "n1": 1, "p": "rainbow", "n3": 0, "d": 150}, "3": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 255, 255]], "b": 255, "n2": 0, "n1": 2, "p": "rainbow", "n3": 0, "d": 100}, "2": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 255, 255]], "b": 0, "n2": 0, "n1": 0, "p": "off", "n3": 0, "d": 100}, "38": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 0, 0], [0, 0, 255]], "b": 255, "n2": 0, "n1": 1, "p": "colour_cycle", "n3": 0, "d": 100}, "11": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 0, 0]], "b": 255, "n2": 0, "n1": 0, "p": "on", "n3": 0, "d": 100}, "12": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[0, 0, 255]], "b": 255, "n2": 0, "n1": 0, "p": "on", "n3": 0, "d": 100}, "1": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 255, 255]], "b": 255, "n2": 0, "n1": 0, "p": "on", "n3": 0, "d": 100}, "9": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 245, 230]], "b": 200, "n2": 0, "n1": 0, "p": "on", "n3": 0, "d": 100}, "8": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 0, 0], [0, 255, 0], [0, 0, 255], [255, 255, 0]], "b": 255, "n2": 0, "n1": 0, "p": "blink", "n3": 0, "d": 1000}, "39": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 184, 77]], "b": 255, "n2": 0, "n1": 30, "p": "flicker", "n3": 0, "d": 80}, "14": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 102, 0]], "b": 255, "n2": 1000, "n1": 2000, "p": "pulse", "n3": 2000, "d": 800}, "5": {"n5": 0, "n4": 1, "a": true, "n6": 0, "c": [[255, 0, 0], [0, 0, 255]], "b": 255, "n2": 5, "n1": 5, "p": "chase", "n3": 1, "d": 200}, "4": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 0, 0], [0, 255, 0], [0, 0, 255], [255, 255, 255], [0, 0, 255], [255, 255, 0]], "b": 255, "n2": 0, "n1": 0, "p": "transition", "n3": 0, "d": 5000}, "7": {"n5": 0, "n4": 5, "a": true, "n6": 0, "c": [[255, 165, 0], [128, 0, 128]], "b": 255, "n2": 10, "n1": 2, "p": "circle", "n3": 2, "d": 200}}

src/controller_messages.py

@@ -0,0 +1,239 @@
+"""Parse controller JSON (v1) and apply brightness, presets, OTA patterns, etc."""
+
+import json
+import socket
+
+from utils import convert_and_reorder_colors
+
+try:
+    import uos as os
+except ImportError:
+    import os
+
+
+def process_data(payload, settings, presets, controller_ip=None):
+    """Read one controller message; json.loads (bytes or str), then apply fields."""
+    try:
+        data = json.loads(payload)
+        print(payload)
+        if data.get("v", "") != "1":
+            return
+    except (ValueError, TypeError):
+        return
+    if "b" in data:
+        apply_brightness(data, settings, presets)
+    if "presets" in data:
+        apply_presets(data, settings, presets)
+    if "clear_presets" in data:
+        apply_clear_presets(data, presets)
+    if "select" in data:
+        apply_select(data, settings, presets)
+    if "default" in data:
+        apply_default(data, settings, presets)
+    if "manifest" in data:
+        apply_patterns_ota(data, presets, controller_ip=controller_ip)
+    if "save" in data and ("presets" in data or "default" in data):
+        presets.save()
+    if "save" in data and "clear_presets" in data:
+        presets.save()
+
+
+def apply_brightness(data, settings, presets):
+    try:
+        presets.b = max(0, min(255, int(data["b"])))
+        settings["brightness"] = presets.b
+    except (TypeError, ValueError):
+        pass
+
+
+def apply_presets(data, settings, presets):
+    presets_map = data["presets"]
+    for id, preset_data in presets_map.items():
+        if not preset_data:
+            continue
+        color_key = "c" if "c" in preset_data else ("colors" if "colors" in preset_data else None)
+        if color_key is not None:
+            try:
+                preset_data[color_key] = convert_and_reorder_colors(
+                    preset_data[color_key], settings
+                )
+            except (TypeError, ValueError, KeyError):
+                continue
+        presets.edit(id, preset_data)
+        print(f"Edited preset {id}: {preset_data.get('name', '')}")
+
+
+def apply_select(data, settings, presets):
+    select_map = data["select"]
+    device_name = settings["name"]
+    select_list = select_map.get(device_name, [])
+    if not select_list:
+        return
+    preset_name = select_list[0]
+    step = select_list[1] if len(select_list) > 1 else None
+    presets.select(preset_name, step=step)
+
+
+def apply_clear_presets(data, presets):
+    clear_value = data.get("clear_presets")
+    if isinstance(clear_value, bool):
+        should_clear = clear_value
+    elif isinstance(clear_value, int):
+        should_clear = bool(clear_value)
+    elif isinstance(clear_value, str):
+        should_clear = clear_value.lower() in ("true", "1", "yes", "on")
+    else:
+        should_clear = False
+    if not should_clear:
+        return
+    presets.delete_all()
+    print("Cleared all presets.")
+
+
+def apply_default(data, settings, presets):
+    targets = data.get("targets") or []
+    default_name = data["default"]
+    if (
+        settings["name"] in targets
+        and isinstance(default_name, str)
+        and default_name in presets.presets
+    ):
+        settings["default"] = default_name
+    settings.save()
+
+
+def _parse_http_url(url):
+    """Parse http://host[:port]/path into (host, port, path)."""
+    if not isinstance(url, str):
+        raise ValueError("url must be a string")
+    if not url.startswith("http://"):
+        raise ValueError("only http:// URLs are supported")
+    remainder = url[7:]
+    slash_idx = remainder.find("/")
+    if slash_idx == -1:
+        host_port = remainder
+        path = "/"
+    else:
+        host_port = remainder[:slash_idx]
+        path = remainder[slash_idx:]
+    if ":" in host_port:
+        host, port_s = host_port.rsplit(":", 1)
+        port = int(port_s)
+    else:
+        host = host_port
+        port = 80
+    if not host:
+        raise ValueError("missing host")
+    return host, port, path
+
+
+def _http_get_raw(url, timeout_s=10.0):
+    host, port, path = _parse_http_url(url)
+    req = (
+        "GET %s HTTP/1.1\r\nHost: %s\r\nConnection: close\r\n\r\n" % (path, host)
+    ).encode("utf-8")
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    try:
+        sock.settimeout(timeout_s)
+        sock.connect((host, int(port)))
+        sock.send(req)
+        data = b""
+        while True:
+            chunk = sock.recv(1024)
+            if not chunk:
+                break
+            data += chunk
+    finally:
+        try:
+            sock.close()
+        except Exception:
+            pass
+    sep = b"\r\n\r\n"
+    if sep not in data:
+        raise OSError("invalid HTTP response")
+    head, body = data.split(sep, 1)
+    status_line = head.split(b"\r\n", 1)[0]
+    if b" 200 " not in status_line:
+        raise OSError("HTTP status not OK: %s" % status_line.decode("utf-8"))
+    return body
+
+
+def _http_get_json(url, timeout_s=10.0):
+    body = _http_get_raw(url, timeout_s=timeout_s)
+    return json.loads(body.decode("utf-8"))
+
+
+def _http_get_text(url, timeout_s=10.0, controller_ip=None):
+    # Support relative URLs from controller messages.
+    if isinstance(url, str) and url.startswith("/"):
+        if not controller_ip:
+            raise OSError("controller IP unavailable for relative URL")
+        url = "http://%s%s" % (controller_ip, url)
+    try:
+        body = _http_get_raw(url, timeout_s=timeout_s)
+        return body.decode("utf-8")
+    except Exception:
+        # Fallback for mDNS/unresolvable host: retry against current controller IP.
+        if not controller_ip or not isinstance(url, str) or not url.startswith("http://"):
+            raise
+        _host, _port, path = _parse_http_url(url)
+        fallback = "http://%s:%d%s" % (controller_ip, _port, path)
+        body = _http_get_raw(fallback, timeout_s=timeout_s)
+        return body.decode("utf-8")
+
+
+def _safe_pattern_filename(name):
+    if not isinstance(name, str):
+        return False
+    if not name.endswith(".py"):
+        return False
+    if "/" in name or "\\" in name or ".." in name:
+        return False
+    return True
+
+
+def apply_patterns_ota(data, presets, controller_ip=None):
+    manifest_payload = data.get("manifest")
+    if not manifest_payload:
+        return
+    try:
+        if isinstance(manifest_payload, dict):
+            manifest = manifest_payload
+        elif isinstance(manifest_payload, str):
+            manifest = _http_get_json(manifest_payload, timeout_s=20.0)
+        else:
+            print("patterns_ota: invalid manifest payload type")
+            return
+        files = manifest.get("files", [])
+        if not isinstance(files, list) or not files:
+            print("patterns_ota: no files in manifest")
+            return
+        try:
+            os.mkdir("patterns")
+        except OSError:
+            pass
+        updated = 0
+        for item in files:
+            if not isinstance(item, dict):
+                continue
+            name = item.get("name")
+            url = item.get("url")
+            inline_code = item.get("code")
+            if not _safe_pattern_filename(name):
+                continue
+            if isinstance(inline_code, str):
+                code = inline_code
+            elif isinstance(url, str):
+                code = _http_get_text(url, timeout_s=20.0, controller_ip=controller_ip)
+            else:
+                continue
+            with open("patterns/" + name, "w") as f:
+                f.write(code)
+            updated += 1
+        if updated > 0:
+            presets.reload_patterns()
+            print("patterns_ota: updated", updated, "pattern file(s)")
+        else:
+            print("patterns_ota: no valid files downloaded")
+    except Exception as e:
+        print("patterns_ota failed:", e)

src/hello.py

@@ -1,4 +1,8 @@
-"""LED hello payload and UDP broadcast discovery (controller IP via echo on port 8766).
+"""LED hello JSON line and UDP broadcast on port 8766.
+
+Used so led-controller can register the device (name, MAC, IP) when ``wait_reply`` is
+false; the controller may then connect to the device's WebSocket. With
+``wait_reply`` true, blocks for an echo and returns the controller IP (legacy discovery).
 
 Wi-Fi must already be connected; this module does not use Settings or call connect().
 """
@@ -40,7 +44,13 @@ def ipv4_broadcast(ip, netmask):
     im = [int(x) for x in netmask.split(".")]
     if len(ia) != 4 or len(im) != 4:
         return None
-    return ".".join(str(ia[i] | (255 - im[i])) for i in range(4))
+    # STA often reports 255.255.255.255; "broadcast" would equal the host IP — useless for LAN.
+    if netmask == "255.255.255.255":
+        return None
+    bcast = ".".join(str(ia[i] | (255 - im[i])) for i in range(4))
+    if bcast == ip:
+        return None
+    return bcast
 
 
 def udp_discovery_targets(ip, mask):
@@ -52,6 +62,14 @@ def udp_discovery_targets(ip, mask):
     return out
 
 
+def _udp_discovery_targets_single(ip, mask):
+    """One destination: subnet broadcast if known, else limited broadcast."""
+    b = ipv4_broadcast(ip, mask)
+    if b:
+        return [(b, DISCOVERY_UDP_PORT)]
+    return [("255.255.255.255", DISCOVERY_UDP_PORT)]
+
+
 def broadcast_hello_udp(
     sta,
     device_name="",
@@ -59,11 +77,17 @@ def broadcast_hello_udp(
     wait_reply=True,
     recv_timeout_s=DEFAULT_RECV_TIMEOUT_S,
     wdt=None,
+    dual_destinations=True,
 ):
     """
-    Send pack_hello_line via directed then 255.255.255.255 on DISCOVERY_UDP_PORT.
+    Send pack_hello_line on DISCOVERY_UDP_PORT.
     STA must already be connected with a valid IPv4 (caller brings up Wi-Fi).
 
+    If dual_destinations (default), send subnet broadcast then 255.255.255.255 so
+    discovery works on awkward APs — the controller may receive two packets.
+    If dual_destinations is False, send only one (subnet broadcast or limited),
+    e.g. after TCP connect so the Pi does not run duplicate resync handlers.
+
     If wait_reply, wait for first UDP echo. Returns controller IP string or None.
     """
     ip, mask, _gw, _dns = sta.ifconfig()
@@ -89,7 +113,12 @@ def broadcast_hello_udp(
             pass
 
     discovered = None
-    for dest_ip, dest_port in udp_discovery_targets(ip, mask):
+    targets = (
+        udp_discovery_targets(ip, mask)
+        if dual_destinations
+        else _udp_discovery_targets_single(ip, mask)
+    )
+    for dest_ip, dest_port in targets:
         if wdt is not None:
             wdt.feed()
         label = "%s:%s" % (dest_ip, dest_port)

src/http_poll.py

@@ -1,68 +0,0 @@
-"""Minimal HTTP/1.1 POST JSON client for driver long-poll (MicroPython)."""
-
-import json
-import socket
-
-
-def _send_all(sock, data):
-    n = 0
-    while n < len(data):
-        m = sock.send(data[n:])
-        if m <= 0:
-            raise OSError("socket send failed")
-        n += m
-
-
-def _read_http_json_body(sock, max_headers=8192):
-    buf = b""
-    while b"\r\n\r\n" not in buf:
-        chunk = sock.recv(256)
-        if not chunk:
-            break
-        buf += chunk
-        if len(buf) > max_headers:
-            raise OSError("response headers too large")
-    if b"\r\n\r\n" not in buf:
-        raise OSError("incomplete response headers")
-    head, rest = buf.split(b"\r\n\r\n", 1)
-    cl = None
-    for line in head.split(b"\r\n"):
-        if line.lower().startswith(b"content-length:"):
-            try:
-                cl = int(line.split(b":", 1)[1].strip())
-            except (ValueError, IndexError):
-                cl = None
-    if cl is None:
-        body = rest
-    else:
-        body = rest
-        while len(body) < cl:
-            chunk = sock.recv(min(2048, cl - len(body)))
-            if not chunk:
-                break
-            body += chunk
-    return json.loads(body.decode("utf-8"))
-
-
-def http_driver_poll(host, port, payload_dict, timeout_s=40.0):
-    """
-    POST ``/driver/v1/poll`` with JSON body; return parsed JSON (expects ``{"lines": [...]}``).
-    """
-    path = "/driver/v1/poll"
-    body_bytes = json.dumps(payload_dict).encode("utf-8")
-    host_s = str(host)
-    req_head = (
-        "POST %s HTTP/1.1\r\nHost: %s\r\nContent-Type: application/json\r\nContent-Length: %d\r\nConnection: close\r\n\r\n"
-        % (path, host_s, len(body_bytes))
-    ).encode("utf-8")
-    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    try:
-        sock.settimeout(timeout_s)
-        sock.connect((host_s, int(port)))
-        _send_all(sock, req_head + body_bytes)
-        return _read_http_json_body(sock)
-    finally:
-        try:
-            sock.close()
-        except Exception:
-            pass

src/main.py

@@ -1,191 +1,65 @@
 from settings import Settings
-from machine import WDT
+import machine
-import utime
 import network
-from presets import Presets
+import utime
-from utils import convert_and_reorder_colors
+import asyncio
 import json
-import time
+import gc
-import select
+from microdot import Microdot
-import socket
+from microdot.websocket import WebSocketError, with_websocket
-import ubinascii
+from presets import Presets
-from hello import discover_controller_udp
+from controller_messages import process_data
+from hello import broadcast_hello_udp
 try:
     import uos as os
 except ImportError:
     import os
 
-BROADCAST_MAC = b"\xff\xff\xff\xff\xff\xff"
+machine.freq(160000000)
-CONTROLLER_TCP_PORT = 8765
+
-controller_ip = None
 
 settings = Settings()
 print(settings)
 
+wdt = machine.WDT(timeout=10000)
+wdt.feed()
+
+gc.collect()
+print("mem before presets:", {"free": gc.mem_free(), "alloc": gc.mem_alloc()})
+
 presets = Presets(settings["led_pin"], settings["num_leds"])
 presets.load(settings)
 presets.b = settings.get("brightness", 255)
+presets.debug = bool(settings.get("debug", False))
+gc.collect()
+print("mem after presets:", {"free": gc.mem_free(), "alloc": gc.mem_alloc()})
+
 default_preset = settings.get("default", "")
 if default_preset and default_preset in presets.presets:
-    presets.select(default_preset)
+    if presets.select(default_preset):
         print(f"Selected startup preset: {default_preset}")
-
-wdt = WDT(timeout=10000)
-wdt.feed()
-
-
-# --- Controller JSON (bytes or str): parse v1, then apply -------------------------
-
-
-def process_data(payload):
-    """Read one controller message; json.loads (bytes or str), then apply fields."""
-    try:
-        data = json.loads(payload)
-        print(payload)
-        if data.get("v", "") != "1":
-            return
-    except (ValueError, TypeError):
-        return
-    if "b" in data:
-        apply_brightness(data)
-    if "presets" in data:
-        apply_presets(data)
-    if "select" in data:
-        apply_select(data)
-    if "default" in data:
-        apply_default(data)
-    if "manifest" in data:
-        apply_patterns_ota(data)
-    if "save" in data and ("presets" in data or "default" in data):
-        presets.save()
-
-
-def apply_brightness(data):
-    try:
-        presets.b = max(0, min(255, int(data["b"])))
-        settings["brightness"] = presets.b
-    except (TypeError, ValueError):
-        pass
-
-
-def apply_presets(data):
-    presets_map = data["presets"]
-    for id, preset_data in presets_map.items():
-        if not preset_data:
-            continue
-        color_key = "c" if "c" in preset_data else ("colors" if "colors" in preset_data else None)
-        if color_key is not None:
-            try:
-                preset_data[color_key] = convert_and_reorder_colors(
-                    preset_data[color_key], settings
-                )
-            except (TypeError, ValueError, KeyError):
-                continue
-        presets.edit(id, preset_data)
-        print(f"Edited preset {id}: {preset_data.get('name', '')}")
-
-
-def apply_select(data):
-    select_map = data["select"]
-    device_name = settings["name"]
-    select_list = select_map.get(device_name, [])
-    if not select_list:
-        return
-    preset_name = select_list[0]
-    step = select_list[1] if len(select_list) > 1 else None
-    presets.select(preset_name, step=step)
-
-
-def apply_default(data):
-    targets = data.get("targets") or []
-    default_name = data["default"]
-    if (
-        settings["name"] in targets
-        and isinstance(default_name, str)
-        and default_name in presets.presets
-    ):
-        settings["default"] = default_name
-
-
-def _parse_http_url(url):
-    """Parse http://host[:port]/path into (host, port, path)."""
-    if not isinstance(url, str):
-        raise ValueError("url must be a string")
-    if not url.startswith("http://"):
-        raise ValueError("only http:// URLs are supported")
-    remainder = url[7:]
-    slash_idx = remainder.find("/")
-    if slash_idx == -1:
-        host_port = remainder
-        path = "/"
     else:
-        host_port = remainder[:slash_idx]
+        print("Startup preset failed (invalid pattern?):", default_preset)
-        path = remainder[slash_idx:]
-    if ":" in host_port:
-        host, port_s = host_port.rsplit(":", 1)
-        port = int(port_s)
-    else:
-        host = host_port
-        port = 80
-    if not host:
-        raise ValueError("missing host")
-    return host, port, path
 
+# On ESP32-C3, soft reboots can leave Wi-Fi driver state allocated.
+# Reset both interfaces and collect before bringing STA up.
+ap_if = network.WLAN(network.AP_IF)
+ap_if.active(False)
+sta_if = network.WLAN(network.STA_IF)
+if sta_if.active():
+    sta_if.active(False)
+utime.sleep_ms(100)
+gc.collect()
+sta_if.active(True)
+sta_if.config(pm=network.WLAN.PM_NONE)
+sta_if.connect(settings["ssid"], settings["password"])
+while not sta_if.isconnected():
+    utime.sleep(1)
+    wdt.feed()
 
-def _http_get_raw(url, timeout_s=10.0):
+print(sta_if.ifconfig())
-    host, port, path = _parse_http_url(url)
-    req = (
-        "GET %s HTTP/1.1\r\nHost: %s\r\nConnection: close\r\n\r\n" % (path, host)
-    ).encode("utf-8")
-    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    try:
-        sock.settimeout(timeout_s)
-        sock.connect((host, int(port)))
-        sock.send(req)
-        data = b""
-        while True:
-            chunk = sock.recv(1024)
-            if not chunk:
-                break
-            data += chunk
-    finally:
-        try:
-            sock.close()
-        except Exception:
-            pass
-    sep = b"\r\n\r\n"
-    if sep not in data:
-        raise OSError("invalid HTTP response")
-    head, body = data.split(sep, 1)
-    status_line = head.split(b"\r\n", 1)[0]
-    if b" 200 " not in status_line:
-        raise OSError("HTTP status not OK: %s" % status_line.decode("utf-8"))
-    return body
 
-
+app = Microdot()
-def _http_get_json(url, timeout_s=10.0):
-    body = _http_get_raw(url, timeout_s=timeout_s)
-    return json.loads(body.decode("utf-8"))
-
-
-def _http_get_text(url, timeout_s=10.0):
-    global controller_ip
-    # Support relative URLs from controller messages.
-    if isinstance(url, str) and url.startswith("/"):
-        if not controller_ip:
-            raise OSError("controller IP unavailable for relative URL")
-        url = "http://%s%s" % (controller_ip, url)
-    try:
-        body = _http_get_raw(url, timeout_s=timeout_s)
-        return body.decode("utf-8")
-    except Exception:
-        # Fallback for mDNS/unresolvable host: retry against current controller IP.
-        if not controller_ip or not isinstance(url, str) or not url.startswith("http://"):
-            raise
-        _host, _port, path = _parse_http_url(url)
-        fallback = "http://%s:%d%s" % (controller_ip, _port, path)
-        body = _http_get_raw(fallback, timeout_s=timeout_s)
-        return body.decode("utf-8")
 
 
 def _safe_pattern_filename(name):
@@ -198,193 +72,137 @@ def _safe_pattern_filename(name):
     return True
 
 
-def apply_patterns_ota(data):
+@app.route("/ws")
-    manifest_payload = data.get("manifest")
+@with_websocket
-    if not manifest_payload:
+async def ws_handler(request, ws):
-        return
+    print("WS client connected")
+    controller_ip = None
     try:
-        if isinstance(manifest_payload, dict):
+        client_addr = getattr(request, "client_addr", None)
-            manifest = manifest_payload
+        if isinstance(client_addr, (tuple, list)) and client_addr:
-        elif isinstance(manifest_payload, str):
+            controller_ip = client_addr[0]
-            manifest = _http_get_json(manifest_payload, timeout_s=20.0)
+        elif isinstance(client_addr, str):
-        else:
+            controller_ip = client_addr
-            print("patterns_ota: invalid manifest payload type")
+    except Exception:
-            return
+        controller_ip = None
-        files = manifest.get("files", [])
+    print("WS controller_ip:", controller_ip)
-        if not isinstance(files, list) or not files:
+    try:
-            print("patterns_ota: no files in manifest")
+        while True:
-            return
+            data = await ws.receive()
+            if not data:
+                print("WS client disconnected (closed)")
+                break
+            print("WS recv bytes:", len(data) if isinstance(data, (bytes, bytearray)) else len(str(data)))
+            print(data)
+            process_data(data, settings, presets, controller_ip=controller_ip)
+    except WebSocketError as e:
+        print("WS client disconnected:", e)
+    except OSError as e:
+        print("WS client dropped (OSError):", e)
+
+
+@app.post("/patterns/upload")
+async def upload_pattern(request):
+    """Receive one pattern file body from led-controller and reload patterns."""
+    raw_name = request.args.get("name")
+    reload_raw = request.args.get("reload", "1")
+    reload_patterns = str(reload_raw).strip().lower() not in ("0", "false", "no", "off")
+    print("patterns/upload request:", {"name": raw_name, "reload": reload_patterns})
+
+    if not isinstance(raw_name, str) or not raw_name.strip():
+        return json.dumps({"error": "name is required"}), 400, {
+            "Content-Type": "application/json"
+        }
+    body = request.body
+    if not isinstance(body, (bytes, bytearray)) or not body:
+        print("patterns/upload rejected: empty body")
+        return json.dumps({"error": "code is required"}), 400, {
+            "Content-Type": "application/json"
+        }
+    print("patterns/upload body_bytes:", len(body))
+    try:
+        code = body.decode("utf-8")
+    except UnicodeError:
+        print("patterns/upload rejected: body not utf-8")
+        return json.dumps({"error": "body must be utf-8 text"}), 400, {
+            "Content-Type": "application/json"
+        }
+    if not code.strip():
+        return json.dumps({"error": "code is required"}), 400, {
+            "Content-Type": "application/json"
+        }
+
+    name = raw_name.strip()
+    if not name.endswith(".py"):
+        name += ".py"
+    if not _safe_pattern_filename(name) or name in ("__init__.py", "main.py"):
+        return json.dumps({"error": "invalid pattern filename"}), 400, {
+            "Content-Type": "application/json"
+        }
+
     try:
         os.mkdir("patterns")
     except OSError:
         pass
-        updated = 0
+
-        for item in files:
+    path = "patterns/" + name
-            if not isinstance(item, dict):
+    try:
-                continue
+        print("patterns/upload writing:", path)
-            name = item.get("name")
+        with open(path, "w") as f:
-            url = item.get("url")
-            inline_code = item.get("code")
-            if not _safe_pattern_filename(name):
-                continue
-            if isinstance(inline_code, str):
-                code = inline_code
-            elif isinstance(url, str):
-                code = _http_get_text(url, timeout_s=20.0)
-            else:
-                continue
-            with open("patterns/" + name, "w") as f:
             f.write(code)
-            updated += 1
+        if reload_patterns:
-        if updated > 0:
+            print("patterns/upload reloading patterns")
             presets.reload_patterns()
-            print("patterns_ota: updated", updated, "pattern file(s)")
+    except OSError as e:
-        else:
+        print("patterns/upload failed:", e)
-            print("patterns_ota: no valid files downloaded")
+        return json.dumps({"error": str(e)}), 500, {
-    except Exception as e:
+            "Content-Type": "application/json"
-        print("patterns_ota failed:", e)
+        }
+    print("patterns/upload success:", {"name": name, "reloaded": reload_patterns})
+
+    return json.dumps({
+        "message": "pattern uploaded",
+        "name": name,
+        "reloaded": reload_patterns,
+    }), 201, {"Content-Type": "application/json"}
 
 
-# --- TCP framing (bytes) → process_data -------------------------------------------
+async def presets_loop():
-
+    last_mem_log = utime.ticks_ms()
-
-def tcp_append_and_drain_lines(buf, chunk):
-    """Return (new_buf, list of non-empty stripped line byte strings)."""
-    buf += chunk
-    lines = []
-    while b"\n" in buf:
-        line, buf = buf.split(b"\n", 1)
-        line = line.strip()
-        if line:
-            lines.append(line)
-    return buf, lines
-
-
-# --- Network + hello --------------------------------------------------------------
-
-sta_if = network.WLAN(network.STA_IF)
-sta_if.active(True)
-sta_if.config(pm=network.WLAN.PM_NONE)
-
-mac = sta_if.config("mac")
-hello_payload = {
-    "v": "1",
-    "device_name": settings.get("name", ""),
-    "mac": ubinascii.hexlify(mac).decode().lower(),
-    "type": "led",
-}
-hello_bytes = json.dumps(hello_payload).encode("utf-8")
-
-if settings["transport_type"] == "espnow":
-    from espnow import ESPNow  # import only in this branch (avoids load when using Wi-Fi)
-
-    sta_if.disconnect()
-    sta_if.config(channel=settings.get("wifi_channel", 1))
-    e = ESPNow()
-    e.active(True)
-    e.add_peer(BROADCAST_MAC)
-    e.add_peer(mac)
-    e.send(BROADCAST_MAC, hello_bytes)
     while True:
-        if e.any():
-            _peer, msg = e.recv()
-            if msg:
-                process_data(msg)
         presets.tick()
         wdt.feed()
-
+        if bool(getattr(presets, "debug", False)):
-elif settings["transport_type"] == "wifi":
-    sta_if.connect(settings["ssid"], settings["password"])
-    while not sta_if.isconnected():
-        time.sleep(1)
-    print(f"WiFi connected {sta_if.ifconfig()[0]}")
-    controller_ip = discover_controller_udp(
-        device_name=settings.get("name", ""),
-        wdt=wdt,
-    )
-    if not controller_ip:
-        raise SystemExit("No controller IP discovered for Wi-Fi transport")
-
-    def pick_controller_ip(current):
-        ip = discover_controller_udp(
-            device_name=settings.get("name", ""),
-            wdt=wdt,
-        )
-        if ip and ip != current:
-            print("Controller IP updated to", ip)
-        return ip if ip else current
-
-    reconnect_ms = 1000
-    next_connect_at = 0
-    client = None
-    poller = None
-    buf = b""
-
-    while True:
             now = utime.ticks_ms()
+            if utime.ticks_diff(now, last_mem_log) >= 5000:
+                gc.collect()
+                print("mem runtime:", {"free": gc.mem_free(), "alloc": gc.mem_alloc()})
+                last_mem_log = now
+        # tick() does not await; yield so UDP hello and HTTP/WebSocket can run.
+        await asyncio.sleep(0)
 
-        if client is None and utime.ticks_diff(now, next_connect_at) >= 0:
+
-            c = None
+async def _udp_hello_after_http_ready():
+    """Hello must run after the HTTP server binds, or discovery clients time out on /ws."""
+    await asyncio.sleep(1)
+    print("UDP hello: broadcasting…")
     try:
-                c = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        broadcast_hello_udp(
-                c.connect((controller_ip, CONTROLLER_TCP_PORT))
+            sta_if,
-                c.setblocking(False)
+            settings.get("name", ""),
-                p = select.poll()
+            wait_reply=False,
-                p.register(c, select.POLLIN)
+            wdt=wdt,
-                client = c
+            dual_destinations=True,
-                poller = p
+        )
-                buf = b""
+    except Exception as ex:
-                print("TCP connected")
+        print("UDP hello broadcast failed:", ex)
-            except Exception:
-                if c is not None:
-                    try:
-                        c.close()
-                    except Exception:
-                        pass
-                controller_ip = pick_controller_ip(controller_ip)
-                next_connect_at = utime.ticks_add(now, reconnect_ms)
 
-        if client is not None and poller is not None:
-            try:
-                events = poller.poll(0)
-            except Exception:
-                events = []
 
-            reconnect_needed = False
+async def main(port=80):
-            for fd, event in events:
+    asyncio.create_task(presets_loop())
-                if (event & select.POLLHUP) or (event & select.POLLERR):
+    asyncio.create_task(_udp_hello_after_http_ready())
-                    reconnect_needed = True
+    await app.start_server(host="0.0.0.0", port=port)
-                    break
-                if event & select.POLLIN:
-                    try:
-                        chunk = client.recv(512)
-                    except OSError:
-                        reconnect_needed = True
-                        break
 
-                    if not chunk:
-                        reconnect_needed = True
-                        break
 
-                    buf, lines = tcp_append_and_drain_lines(buf, chunk)
+if __name__ == "__main__":
-                    for raw_line in lines:
+    asyncio.run(main(port=80))
-                        process_data(raw_line)
-
-            if reconnect_needed:
-                print("TCP disconnected, reconnecting...")
-                try:
-                    poller.unregister(client)
-                except Exception:
-                    pass
-                try:
-                    client.close()
-                except Exception:
-                    pass
-                client = None
-                poller = None
-                buf = b""
-                controller_ip = pick_controller_ip(controller_ip)
-                next_connect_at = utime.ticks_add(now, reconnect_ms)
-
-        presets.tick()
-        wdt.feed()

src/patterns/__init__.py

@@ -1,6 +1,5 @@
-from .blink import Blink
+"""Pattern modules are registered only via Presets._load_dynamic_patterns().
-from .rainbow import Rainbow
+
-from .pulse import Pulse
+This file is ignored as a pattern (see presets.py). Keep it free of imports so
-from .transition import Transition
+adding a pattern does not require editing this package.
-from .chase import Chase
+"""
-from .circle import Circle

src/patterns/blink.py

@@ -28,6 +28,6 @@ class Blink:
                     # "Off" phase: turn all LEDs off
                     self.driver.fill((0, 0, 0))
                 state = not state
-                last_update = current_time
+                last_update = utime.ticks_add(last_update, delay_ms)
             # Yield once per tick so other logic can run
             yield

src/patterns/chase.py

@@ -118,7 +118,8 @@ class Chase:
                 # Increment step
                 step_count += 1
                 self.driver.step = step_count
-                last_update = current_time
+                last_update = utime.ticks_add(last_update, transition_duration)
+                transition_duration = max(10, int(preset.d))
 
             # Yield once per tick so other logic can run
             yield

src/patterns/circle.py

@@ -62,7 +62,9 @@ class Circle:
             # Move head continuously at n1 LEDs per second
             if utime.ticks_diff(current_time, last_head_move) >= head_delay:
                 head = (head + 1) % self.driver.num_leds
-                last_head_move = current_time
+                last_head_move = utime.ticks_add(last_head_move, head_delay)
+                head_rate = max(1, int(preset.n1))
+                head_delay = 1000 // head_rate
 
             # Tail behavior based on phase
             if phase == "growing":
@@ -73,7 +75,9 @@ class Circle:
                 # Shrinking phase: move tail forward at n3 LEDs per second
                 if utime.ticks_diff(current_time, last_tail_move) >= tail_delay:
                     tail = (tail + 1) % self.driver.num_leds
-                    last_tail_move = current_time
+                    last_tail_move = utime.ticks_add(last_tail_move, tail_delay)
+                    tail_rate = max(1, int(preset.n3))
+                    tail_delay = 1000 // tail_rate
 
                     # Check if we've reached min length
                     current_length = (head - tail) % self.driver.num_leds

src/patterns/colour_cycle.py

@@ -0,0 +1,56 @@
+import utime
+
+
+class ColourCycle:
+    def __init__(self, driver):
+        self.driver = driver
+
+    def _render(self, colors, phase, brightness):
+        num_leds = self.driver.num_leds
+        color_count = len(colors)
+        if num_leds <= 0 or color_count <= 0:
+            return
+        if color_count == 1:
+            self.driver.fill(self.driver.apply_brightness(colors[0], brightness))
+            return
+
+        full_span = color_count * 256
+        # Match rainbow behaviour: phase is 0..255 and maps to one full-strip shift.
+        phase_shift = (phase * full_span) // 256
+        for i in range(num_leds):
+            # Position around the colour loop, shifted by phase.
+            pos = ((i * full_span) // num_leds + phase_shift) % full_span
+            idx = pos // 256
+            frac = pos & 255
+
+            c1 = colors[idx]
+            c2 = colors[(idx + 1) % color_count]
+            blended = (
+                c1[0] + ((c2[0] - c1[0]) * frac) // 256,
+                c1[1] + ((c2[1] - c1[1]) * frac) // 256,
+                c1[2] + ((c2[2] - c1[2]) * frac) // 256,
+            )
+            self.driver.n[i] = self.driver.apply_brightness(blended, brightness)
+        self.driver.n.write()
+
+    def run(self, preset):
+        colors = preset.c if preset.c else [(255, 255, 255)]
+        phase = self.driver.step % 256
+        step_amount = max(1, int(preset.n1))
+
+        if not preset.a:
+            self._render(colors, phase, preset.b)
+            self.driver.step = (phase + step_amount) % 256
+            yield
+            return
+
+        last_update = utime.ticks_ms()
+        while True:
+            current_time = utime.ticks_ms()
+            delay_ms = max(1, int(preset.d))
+            if utime.ticks_diff(current_time, last_update) >= delay_ms:
+                self._render(colors, phase, preset.b)
+                phase = (phase + step_amount) % 256
+                self.driver.step = phase
+                last_update = utime.ticks_add(last_update, delay_ms)
+            yield

src/patterns/flame.py

@@ -0,0 +1,210 @@
+import random
+import utime
+
+# Default warm palette: ember → orange → yellow → pale hot (RGB)
+_DEFAULT_PALETTE = (
+    (90, 8, 8),
+    (200, 40, 12),
+    (255, 120, 30),
+    (255, 220, 140),
+)
+
+
+def _clamp(x, lo, hi):
+    if x < lo:
+        return lo
+    if x > hi:
+        return hi
+    return x
+
+
+def _lerp_chan(a, b, t):
+    return a + ((b - a) * t >> 8)
+
+
+def _lerp_rgb(c0, c1, t):
+    return (
+        _lerp_chan(c0[0], c1[0], t),
+        _lerp_chan(c0[1], c1[1], t),
+        _lerp_chan(c0[2], c1[2], t),
+    )
+
+
+def _palette_sample(palette, pos256):
+    n = len(palette)
+    if n == 0:
+        return (255, 160, 60)
+    if n == 1:
+        return palette[0]
+    span = (n - 1) * pos256
+    seg = span >> 8
+    if seg >= n - 1:
+        return palette[n - 1]
+    frac = span & 0xFF
+    return _lerp_rgb(palette[seg], palette[seg + 1], frac)
+
+
+def _triangle_255(elapsed_ms, period_ms):
+    period_ms = max(period_ms, 400)
+    p = elapsed_ms % period_ms
+    half = period_ms >> 1
+    if half <= 0:
+        return 128
+    if p < half:
+        return (p * 255) // half
+    return ((period_ms - p) * 255) // (period_ms - half)
+
+
+class Flame:
+    def __init__(self, driver):
+        self.driver = driver
+
+    def _build_palette(self, preset):
+        colors = preset.c
+        if not colors:
+            return list(_DEFAULT_PALETTE)
+        out = []
+        for c in colors:
+            if isinstance(c, (list, tuple)) and len(c) == 3:
+                out.append(
+                    (
+                        _clamp(int(c[0]), 0, 255),
+                        _clamp(int(c[1]), 0, 255),
+                        _clamp(int(c[2]), 0, 255),
+                    )
+                )
+        return out if out else list(_DEFAULT_PALETTE)
+
+    def _draw_frame(self, preset, palette, ticks_now, breath_el_ms, rise, cluster_jit, breath_ms, lo, hi, spark_state):
+        """spark_state: (active: bool, start_ticks, duration_ms). ticks_now for sparks; breath_el_ms for slow wave."""
+        num = self.driver.num_leds
+        denom = num - 1 if num > 1 else 1
+
+        breathe = _triangle_255(breath_el_ms, breath_ms)
+        base_level = lo + (((hi - lo) * breathe) >> 8)
+        micro = 232 + random.randint(0, 35)
+        level = (base_level * micro) >> 8
+        level = _clamp(level, lo, hi)
+
+        spark_boost = 0
+        spark_white = (0, 0, 0)
+        active, s0, dur = spark_state
+        if active and dur > 0:
+            el = utime.ticks_diff(ticks_now, s0)
+            if el < 0:
+                el = 0
+            if el >= dur:
+                spark_boost = 0
+            else:
+                env = 255 - ((el * 255) // dur)
+                spark_boost = (env * 90) >> 8
+                spark_white = ((env * 55) >> 8, (env * 50) >> 8, (env * 40) >> 8)
+
+        for i in range(num):
+            h = (i * 256) // denom
+            flow = (h + rise + ((i // max(1, num >> 3)) * 17)) & 255
+            pos = (flow + cluster_jit[(i >> 2) & 7]) & 255
+            rgb = _palette_sample(palette, pos)
+            if spark_boost:
+                rgb = (
+                    _clamp(rgb[0] + spark_white[0] + (spark_boost * 3 >> 2), 0, 255),
+                    _clamp(rgb[1] + spark_white[1] + (spark_boost >> 1), 0, 255),
+                    _clamp(rgb[2] + spark_white[2] + (spark_boost >> 2), 0, 255),
+                )
+            self.driver.n[i] = self.driver.apply_brightness(rgb, level)
+
+        self.driver.n.write()
+
+    def run(self, preset):
+        """Salt-lamp / hearth-style flame: warm gradient, breathing, jitter, drift, rare sparks."""
+        palette = self._build_palette(preset)
+        lo = max(0, min(255, int(preset.n1)))
+        hi = max(0, min(255, int(preset.b)))
+        if lo > hi:
+            lo, hi = hi, lo
+
+        bp = int(preset.n2)
+        breath_ms = max(800, bp if bp > 0 else 2500)
+
+        gap_lo = int(preset.n3)
+        gap_hi = int(preset.n4)
+        # n3 < 0 disables sparks; n3=n4=0 uses ~10–30 s gaps (hearth pops).
+        if gap_lo < 0:
+            sparks_on = False
+        else:
+            sparks_on = True
+            if gap_lo == 0 and gap_hi == 0:
+                gap_lo, gap_hi = 10000, 30000
+            else:
+                gap_lo = max(gap_lo, 500)
+            if gap_hi < gap_lo:
+                gap_hi = gap_lo
+
+        delay_ms = max(16, int(preset.d))
+        rise = random.randint(0, 255)
+        cluster_jit = [random.randint(-18, 18) for _ in range(8)]
+        last_draw = utime.ticks_ms()
+        breath_origin = last_draw
+        last_cluster = last_draw
+        spark_active = False
+        spark_start = 0
+        spark_dur = 0
+        next_spark = utime.ticks_add(last_draw, random.randint(gap_lo, gap_hi)) if sparks_on else 0
+
+        if not preset.a:
+            now = utime.ticks_ms()
+            self._draw_frame(
+                preset,
+                palette,
+                now,
+                utime.ticks_diff(now, breath_origin),
+                rise,
+                cluster_jit,
+                breath_ms,
+                lo,
+                hi,
+                (False, 0, 0),
+            )
+            yield
+            return
+
+        while True:
+            now = utime.ticks_ms()
+            if utime.ticks_diff(now, last_draw) < delay_ms:
+                yield
+                continue
+            last_draw = utime.ticks_add(last_draw, delay_ms)
+
+            rise = (rise + random.randint(-10, 12)) & 255
+
+            if utime.ticks_diff(now, last_cluster) >= (delay_ms * 4):
+                last_cluster = now
+                cluster_jit = [random.randint(-18, 18) for _ in range(8)]
+
+            spark_state = (spark_active, spark_start, spark_dur)
+            if sparks_on:
+                if spark_active:
+                    if utime.ticks_diff(now, spark_start) >= spark_dur:
+                        spark_active = False
+                        next_spark = utime.ticks_add(
+                            now,
+                            random.randint(gap_lo, gap_hi),
+                        )
+                elif utime.ticks_diff(now, next_spark) >= 0:
+                    spark_active = True
+                    spark_start = now
+                    spark_dur = random.randint(180, 360)
+
+            self._draw_frame(
+                preset,
+                palette,
+                now,
+                utime.ticks_diff(now, breath_origin),
+                rise,
+                cluster_jit,
+                breath_ms,
+                lo,
+                hi,
+                (spark_active, spark_start, spark_dur),
+            )
+            yield

src/patterns/flicker.py

@@ -0,0 +1,40 @@
+import random
+import utime
+
+
+class Flicker:
+    def __init__(self, driver):
+        self.driver = driver
+
+    def run(self, preset):
+        """Random brightness between n1 (min) and b (max); delay d ms between updates."""
+        colors = preset.c if preset.c else [(255, 255, 255)]
+        color_index = 0
+        last_update = utime.ticks_ms()
+
+        def brightness_bounds():
+            lo = max(0, min(255, int(preset.n1)))
+            hi = max(0, min(255, int(preset.b)))
+            if lo > hi:
+                lo, hi = hi, lo
+            return lo, hi
+
+        if not preset.a:
+            lo, hi = brightness_bounds()
+            level = random.randint(lo, hi)
+            base = colors[color_index % len(colors)]
+            self.driver.fill(self.driver.apply_brightness(base, level))
+            yield
+            return
+
+        while True:
+            current_time = utime.ticks_ms()
+            delay_ms = max(1, int(preset.d))
+            lo, hi = brightness_bounds()
+            if utime.ticks_diff(current_time, last_update) >= delay_ms:
+                level = random.randint(lo, hi)
+                base = colors[color_index % len(colors)]
+                self.driver.fill(self.driver.apply_brightness(base, level))
+                color_index += 1
+                last_update = utime.ticks_add(last_update, delay_ms)
+            yield

src/patterns/radiate.py

@@ -0,0 +1,136 @@
+import utime
+
+_RADIATE_DBG_INTERVAL_MS = 1000
+
+
+class Radiate:
+    def __init__(self, driver):
+        self.driver = driver
+
+    def run(self, preset):
+        """Radiate from nodes every n1 LEDs, retriggering every delay (d).
+
+        - n1: node spacing in LEDs
+        - n2: outbound travel time in ms
+        - n3: return travel time in ms
+        - d: retrigger interval in ms
+        """
+        colors = preset.c if preset.c else [(255, 255, 255)]
+        base_on = colors[0]
+        base_off = colors[1] if len(colors) > 1 else (0, 0, 0)
+
+        spacing = max(1, int(preset.n1))
+        outward_ms = max(1, int(preset.n2))
+        return_ms = max(1, int(preset.n3))
+        max_dist = spacing // 2
+
+        lit_color = self.driver.apply_brightness(base_on, preset.b)
+        off_color = self.driver.apply_brightness(base_off, preset.b)
+
+        now = utime.ticks_ms()
+        last_trigger = now
+        active_pulses = [now]
+        last_dbg = now
+        dbg_banner = False
+
+        if not preset.a:
+            # Single-step render uses only the first instant pulse.
+            active_pulses = [utime.ticks_ms()]
+
+        while True:
+            now = utime.ticks_ms()
+            delay_ms = max(1, int(preset.d))
+            spacing = max(1, int(preset.n1))
+            outward_ms = max(1, int(preset.n2))
+            return_ms = max(1, int(preset.n3))
+            max_dist = spacing // 2
+            lit_color = self.driver.apply_brightness(base_on, preset.b)
+            off_color = self.driver.apply_brightness(base_off, preset.b)
+
+            if preset.a and utime.ticks_diff(now, last_trigger) >= delay_ms:
+                # Keep one pulse train at a time; replacing instead of appending
+                # prevents overlap from keeping color[0] continuously visible.
+                active_pulses = [now]
+                last_trigger = utime.ticks_add(last_trigger, delay_ms)
+                if bool(getattr(self.driver, "debug", False)):
+                    print(
+                        "[radiate] trigger spacing=%d out=%d in=%d delay=%d"
+                        % (spacing, outward_ms, return_ms, delay_ms)
+                    )
+
+            # Drop pulses once their out-and-back lifetime ends.
+            pulse_lifetime = outward_ms + return_ms
+            kept = []
+            for start in active_pulses:
+                age = utime.ticks_diff(now, start)
+                if age < pulse_lifetime:
+                    kept.append(start)
+            active_pulses = kept
+            debug_front = -1
+            lit_count = 0
+
+            for i in range(self.driver.num_leds):
+                # Nearest node distance for a repeating node grid every `spacing` LEDs.
+                offset = i % spacing
+                dist = min(offset, spacing - offset)
+
+                lit = False
+                for start in active_pulses:
+                    age = utime.ticks_diff(now, start)
+                    # Do not render on the exact trigger tick; this avoids
+                    # node LEDs appearing "stuck on" between cycles.
+                    if age <= 0:
+                        continue
+                    if age <= outward_ms:
+                        # Integer-ceiling progression so peak can be reached even
+                        # when tick timing skips the exact outward_ms boundary.
+                        front = (age * max_dist + outward_ms - 1) // outward_ms
+                    elif age <= outward_ms + return_ms:
+                        back_age = age - outward_ms
+                        remaining = return_ms - back_age
+                        front = (remaining * max_dist + return_ms - 1) // return_ms
+                    else:
+                        continue
+
+                    if dist <= front:
+                        lit = True
+                        if front > debug_front:
+                            debug_front = front
+                        break
+
+                self.driver.n[i] = lit_color if lit else off_color
+                if lit:
+                    lit_count += 1
+
+            self.driver.n.write()
+
+            if bool(getattr(self.driver, "debug", False)):
+                if not dbg_banner:
+                    dbg_banner = True
+                    print(
+                        "[radiate] debug on: spacing=%s out=%s in=%s d=%s num=%d"
+                        % (
+                            preset.n1,
+                            preset.n2,
+                            preset.n3,
+                            preset.d,
+                            self.driver.num_leds,
+                        )
+                    )
+                if utime.ticks_diff(now, last_dbg) >= _RADIATE_DBG_INTERVAL_MS:
+                    pulse_age = -1
+                    if active_pulses:
+                        pulse_age = utime.ticks_diff(now, active_pulses[0])
+                    print(
+                        "[radiate] age=%d front=%d max=%d active=%d lit=%d"
+                        % (pulse_age, debug_front, max_dist, len(active_pulses), lit_count)
+                    )
+                    if lit_count == 0:
+                        print("[radiate] fully off")
+                    last_dbg = now
+
+            if not preset.a:
+                yield
+                return
+
+            yield

src/patterns/rainbow.py

@@ -46,6 +46,6 @@ class Rainbow:
                 self.driver.n.write()
                 step = (step + step_amount) % 256
                 self.driver.step = step
-                last_update = current_time
+                last_update = utime.ticks_add(last_update, sleep_ms)
             # Yield once per tick so other logic can run
             yield

src/patterns/twinkle.py

@@ -0,0 +1,227 @@
+import random
+import utime
+
+# Default cool palette (icy blues, violet, mint) when preset has no colours.
+# When `driver.debug` is True, print stats every N twinkle ticks (serial can be slow).
+_TWINKLE_DBG_INTERVAL = 40
+
+_DEFAULT_COOL = (
+    (120, 200, 255),
+    (80, 140, 255),
+    (180, 120, 255),
+    (100, 220, 240),
+    (160, 200, 255),
+    (90, 180, 220),
+)
+
+
+class Twinkle:
+    def __init__(self, driver):
+        self.driver = driver
+
+    def _palette(self, preset):
+        colors = preset.c
+        if not colors:
+            return list(_DEFAULT_COOL)
+        out = []
+        for c in colors:
+            if isinstance(c, (list, tuple)) and len(c) == 3:
+                out.append(
+                    (
+                        max(0, min(255, int(c[0]))),
+                        max(0, min(255, int(c[1]))),
+                        max(0, min(255, int(c[2]))),
+                    )
+                )
+        return out if out else list(_DEFAULT_COOL)
+
+    def run(self, preset):
+        """Twinkle: n1 activity, n2 density; n3/n4 min/max length of adjacent on/off runs."""
+        palette = self._palette(preset)
+        num = self.driver.num_leds
+        if num <= 0:
+            while True:
+                yield
+            return
+
+        def activity_rate():
+            r = int(preset.n1)
+            if r <= 0:
+                r = 48
+            return max(1, min(255, r))
+
+        def density255():
+            """Higher → more LEDs lit on average when a twinkle step fires (0 = default mid)."""
+            d = int(preset.n2)
+            if d <= 0:
+                d = 128
+            return max(0, min(255, d))
+
+        def cluster_len_bounds():
+            """n3 = min adjacent LEDs per twinkle, n4 = max (both 0 → 1..4)."""
+            lo = int(preset.n3)
+            hi = int(preset.n4)
+            if lo <= 0 and hi <= 0:
+                lo, hi = 1, min(4, num)
+            else:
+                if lo <= 0:
+                    lo = 1
+                if hi <= 0:
+                    hi = lo
+                if hi < lo:
+                    lo, hi = hi, lo
+            lo = max(1, min(lo, num))
+            hi = max(lo, min(hi, num))
+            return lo, hi
+
+        def random_cluster_len():
+            lo, hi = cluster_len_bounds()
+            # When min and max match, every lit/dim run is exactly that many LEDs (still capped by strip length).
+            if lo == hi:
+                return lo
+            return random.randint(lo, hi)
+
+        def cluster_base_index(start, k):
+            """Shift run left so a length-k segment fits; keeps full k when num >= k."""
+            k = min(max(0, int(k)), num)
+            if k <= 0:
+                return 0
+            return max(0, min(int(start), num - k))
+
+        dens = density255()
+        on = [random.randint(0, 255) < dens for _ in range(num)]
+        colour_i = [random.randint(0, len(palette) - 1) for _ in range(num)]
+        last_update = utime.ticks_ms()
+        dbg_tick = 0
+        dbg_banner = False
+
+        def on_run_min_max(bits):
+            """Smallest and largest contiguous run of True in bits (0,0 if all off)."""
+            best_min = num + 1
+            best_max = 0
+            cur = 0
+            for j in range(num):
+                if bits[j]:
+                    cur += 1
+                else:
+                    if cur:
+                        if cur < best_min:
+                            best_min = cur
+                        if cur > best_max:
+                            best_max = cur
+                    cur = 0
+            if cur:
+                if cur < best_min:
+                    best_min = cur
+                if cur > best_max:
+                    best_max = cur
+            if best_min == num + 1:
+                return 0, 0
+            return best_min, best_max
+
+        if not preset.a:
+            for i in range(num):
+                if on[i]:
+                    base = palette[colour_i[i] % len(palette)]
+                    self.driver.n[i] = self.driver.apply_brightness(base, preset.b)
+                else:
+                    self.driver.n[i] = (0, 0, 0)
+            self.driver.n.write()
+            yield
+            return
+
+        while True:
+            now = utime.ticks_ms()
+            delay_ms = max(1, int(preset.d))
+            if utime.ticks_diff(now, last_update) >= delay_ms:
+                rate = activity_rate()
+                dens = density255()
+                dbg = bool(getattr(self.driver, "debug", False))
+                dbg_tick += 1
+                # Snapshot for decisions; apply all darks then all lights so
+                # overlaps in the same tick favour lit runs (lights win).
+                prev_on = on[:]
+                prev_ci = colour_i[:]
+                next_on = list(prev_on)
+                next_ci = list(prev_ci)
+                dbg_ops = {"L": 0, "D": 0}
+
+                light_i = []
+                dark_i = []
+                for i in range(num):
+                    if random.randint(0, 255) < rate:
+                        r = random.randint(0, 255)
+                        if not prev_on[i]:
+                            if r < dens:
+                                light_i.append(i)
+                        else:
+                            if r < (255 - dens):
+                                dark_i.append(i)
+
+                def light_adjacent(start):
+                    dbg_ops["L"] += 1
+                    k = random_cluster_len()
+                    b = cluster_base_index(start, k)
+                    for dj in range(k):
+                        idx = b + dj
+                        next_on[idx] = True
+                        next_ci[idx] = random.randint(0, len(palette) - 1)
+
+                def dark_adjacent(start):
+                    dbg_ops["D"] += 1
+                    k = random_cluster_len()
+                    b = cluster_base_index(start, k)
+                    for dj in range(k):
+                        idx = b + dj
+                        next_on[idx] = False
+
+                for i in dark_i:
+                    dark_adjacent(i)
+                for i in light_i:
+                    light_adjacent(i)
+
+                for i in range(num):
+                    if next_on[i]:
+                        base = palette[next_ci[i] % len(palette)]
+                        self.driver.n[i] = self.driver.apply_brightness(base, preset.b)
+                    else:
+                        self.driver.n[i] = (0, 0, 0)
+                self.driver.n.write()
+                on = next_on
+                colour_i = next_ci
+                last_update = utime.ticks_add(last_update, delay_ms)
+
+                if dbg:
+                    lo, hi = cluster_len_bounds()
+                    if not dbg_banner:
+                        dbg_banner = True
+                        print(
+                            "[twinkle] debug on: n1=%s n2=%s n3=%s n4=%s d=%s -> lo=%d hi=%d num=%d"
+                            % (
+                                preset.n1,
+                                preset.n2,
+                                preset.n3,
+                                preset.n4,
+                                preset.d,
+                                lo,
+                                hi,
+                                num,
+                            )
+                        )
+                    rmin, rmax = on_run_min_max(on)
+                    bad = lo > 0 and rmin > 0 and rmin < lo and num >= lo
+                    if bad or (dbg_tick % _TWINKLE_DBG_INTERVAL == 0):
+                        print(
+                            "[twinkle] tick=%d rate=%d dens=%d L=%d D=%d on_runs min=%d max=%d%s"
+                            % (
+                                dbg_tick,
+                                rate,
+                                dens,
+                                dbg_ops["L"],
+                                dbg_ops["D"],
+                                rmin,
+                                rmax,
+                                " **run<lo**" if bad else "",
+                            )
+                        )
+            yield

src/presets.json

@@ -1 +0,0 @@
-{"14": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 102, 0]], "b": 255, "n2": 1000, "n1": 2000, "p": "pulse", "n3": 2000, "d": 800}, "15": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 0, 0], [0, 255, 0]], "b": 255, "n2": 0, "n1": 0, "p": "blink", "n3": 0, "d": 500}, "5": {"n5": 0, "n4": 1, "a": true, "n6": 0, "c": [[255, 0, 0], [0, 0, 255]], "b": 255, "n2": 5, "n1": 5, "p": "chase", "n3": 1, "d": 200}, "4": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 0, 0], [0, 255, 0], [0, 0, 255]], "b": 255, "n2": 0, "n1": 0, "p": "transition", "n3": 0, "d": 500}, "7": {"n5": 0, "n4": 5, "a": true, "n6": 0, "c": [[255, 165, 0], [128, 0, 128]], "b": 255, "n2": 10, "n1": 2, "p": "circle", "n3": 2, "d": 200}, "11": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 0, 0]], "b": 255, "n2": 0, "n1": 0, "p": "on", "n3": 0, "d": 100}, "12": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[0, 0, 255]], "b": 255, "n2": 0, "n1": 0, "p": "on", "n3": 0, "d": 100}, "6": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[0, 255, 0]], "b": 255, "n2": 500, "n1": 1000, "p": "pulse", "n3": 1000, "d": 500}, "3": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 255, 255]], "b": 255, "n2": 0, "n1": 2, "p": "rainbow", "n3": 0, "d": 100}, "2": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 255, 255]], "b": 0, "n2": 0, "n1": 0, "p": "off", "n3": 0, "d": 100}, "1": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 255, 255]], "b": 255, "n2": 0, "n1": 0, "p": "on", "n3": 0, "d": 100}, "10": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[230, 242, 255]], "b": 200, "n2": 0, "n1": 0, "p": "on", "n3": 0, "d": 100}, "13": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 255, 255]], "b": 255, "n2": 0, "n1": 1, "p": "rainbow", "n3": 0, "d": 150}, "9": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 245, 230]], "b": 200, "n2": 0, "n1": 0, "p": "on", "n3": 0, "d": 100}, "8": {"n5": 0, "n4": 0, "a": true, "n6": 0, "c": [[255, 0, 0], [0, 255, 0], [0, 0, 255], [255, 255, 0]], "b": 255, "n2": 0, "n1": 0, "p": "blink", "n3": 0, "d": 1000}}

src/presets.py

@@ -9,6 +9,8 @@ try:
 except ImportError:
     import os
 
+MAX_PRESETS = 32
+
 
 class Presets:
     def __init__(self, pin, num_leds):
@@ -40,7 +42,7 @@ class Presets:
             return loaded
 
         for filename in files:
-            if not filename.endswith(".py") or filename == "__init__.py":
+            if not filename.endswith(".py") or filename in ("__init__.py", "main.py"):
                 continue
             module_basename = filename[:-3]
             module_name = "patterns." + module_basename
@@ -95,6 +97,9 @@ class Presets:
         order = settings if settings is not None else "rgb"
         self.presets = {}
         for name, preset_data in data.items():
+            if len(self.presets) >= MAX_PRESETS:
+                print("Preset limit reached on load:", MAX_PRESETS)
+                break
             color_key = "c" if "c" in preset_data else ("colors" if "colors" in preset_data else None)
             if color_key is not None:
                 preset_data[color_key] = convert_and_reorder_colors(
@@ -113,6 +118,9 @@ class Presets:
             # Update existing preset
             self.presets[name].edit(data)
         else:
+            if len(self.presets) >= MAX_PRESETS and name not in ("on", "off"):
+                print("Preset limit reached:", MAX_PRESETS)
+                return False
             # Create new preset
             self.presets[name] = Preset(data)
         return True
@@ -123,6 +131,12 @@ class Presets:
             return True
         return False
 
+    def delete_all(self):
+        self.presets = {}
+        self.generator = None
+        self.selected = None
+        return True
+
     def tick(self):
         if self.generator is None:
             return
@@ -153,6 +167,9 @@ class Presets:
                 self.generator = self.patterns[preset.p](preset)
                 self.selected = preset_name  # Store the preset name, not the object
                 return True
+            print("select failed: pattern not found for preset", preset_name, "pattern=", preset.p)
+            return False
+        print("select failed: preset not found", preset_name)
         # If preset doesn't exist or pattern not found, indicate failure
         return False
 

src/settings.py

@@ -12,19 +12,25 @@ class Settings(dict):
         self.color_order = self.get_color_order(self["color_order"])
 
     def set_defaults(self):
+
         self["led_pin"] = 10
         self["num_leds"] = 119
       
         self["color_order"] = "rgb"
-        self["name"] = "a"
+
+        sta = network.WLAN(network.STA_IF)
+        sta.active(True)
+        #use led-mac for name
+        mac = sta.config("mac")
+        mac = ubinascii.hexlify(mac).decode().lower()
+        self["name"] = "led-" + mac
        
         self["debug"] = False
         self["default"] = "on"
         self["brightness"] = 32
-        self["transport_type"] = "wifi"
+        self["transport_type"] = "espnow"
         self["wifi_channel"] = 1
-        # Wi-Fi + TCP to controller: set ssid and password. Use transport_type "espnow"
+        # ESP-NOW transport (requires espnow firmware; uses wifi_channel).
-        # for ESP-NOW (requires espnow firmware).
         self["ssid"] = ""
         self["password"] = ""
 

tests/all.py

@@ -7,7 +7,7 @@ import utime
 from machine import WDT
 
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 from utils import convert_and_reorder_colors
 
 
@@ -23,7 +23,7 @@ class _TestContext:
         start = utime.ticks_ms()
         while utime.ticks_diff(utime.ticks_ms(), start) < duration_ms:
             self.wdt.feed()
-            self.presets.tick()
+            run_tick(self.presets)
             utime.sleep_ms(sleep_ms)
 
 

tests/patterns/auto_manual.py

@@ -2,7 +2,7 @@
 import utime
 from machine import WDT
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 
 
 def run_for(p, wdt, duration_ms):
@@ -10,7 +10,7 @@ def run_for(p, wdt, duration_ms):
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < duration_ms:
         wdt.feed()
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(10)
 
 
@@ -52,7 +52,7 @@ def main():
     p.select("rainbow_manual")
     print("Calling tick() 5 times (should advance 5 steps)...")
     for i in range(5):
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(100)  # Small delay to see changes
         print(f"  Tick {i+1}: generator={'active' if p.generator is not None else 'stopped'}")
     
@@ -96,7 +96,7 @@ def main():
     tick_count = 0
     max_ticks = 200  # Safety limit
     while p.generator is not None and tick_count < max_ticks:
-        p.tick()
+        run_tick(p)
         tick_count += 1
         utime.sleep_ms(10)
     
@@ -133,7 +133,7 @@ def main():
     tick_count = 0
     max_ticks = 200
     while p.generator is not None and tick_count < max_ticks:
-        p.tick()
+        run_tick(p)
         tick_count += 1
         utime.sleep_ms(10)
     
@@ -162,7 +162,7 @@ def main():
     
     print("Calling tick() 3 times in manual mode...")
     for i in range(3):
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(100)
         print(f"  Tick {i+1}: generator={'active' if p.generator is not None else 'stopped'}")
     
@@ -178,7 +178,7 @@ def main():
     print("\nCleaning up...")
     p.edit("cleanup_off", {"p": "off"})
     p.select("cleanup_off")
-    p.tick()
+    run_tick(p)
     utime.sleep_ms(100)
     
     print("\n" + "=" * 50)

tests/patterns/blink.py

@@ -2,7 +2,7 @@
 import utime
 from machine import WDT
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 
 
 def main():
@@ -25,7 +25,7 @@ def main():
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < 1500:
         wdt.feed()
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(10)
 
 

tests/patterns/chase.py

@@ -2,7 +2,7 @@
 import utime
 from machine import WDT
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 
 
 def run_for(p, wdt, ms):
@@ -10,7 +10,7 @@ def run_for(p, wdt, ms):
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < ms:
         wdt.feed()
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(10)
 
 
@@ -123,7 +123,7 @@ def main():
     print("  Advancing pattern with 10 beats (select + tick)...")
     for i in range(10):
         p.select("chase_manual")  # Simulate beat - restarts generator
-        p.tick()  # Advance one step
+        run_tick(p)  # Advance one step
         utime.sleep_ms(500)  # Pause to see the pattern
         wdt.feed()
         print(f"  Beat {i+1}: step={p.step}")
@@ -141,7 +141,7 @@ def main():
     p.step = 0
     initial_step = p.step
     p.select("chase_manual2")
-    p.tick()
+    run_tick(p)
     final_step = p.step
     print(f"  Step updated from {initial_step} to {final_step} (expected: 1)")
     if final_step == 1:

tests/patterns/circle.py

@@ -2,7 +2,7 @@
 import utime
 from machine import WDT
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 
 
 def run_for(p, wdt, ms):
@@ -10,7 +10,7 @@ def run_for(p, wdt, ms):
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < ms:
         wdt.feed()
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(10)
 
 

tests/patterns/off.py

@@ -2,7 +2,7 @@
 import utime
 from machine import WDT
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 
 
 def main():
@@ -20,7 +20,7 @@ def main():
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < 200:
         wdt.feed()
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(10)
 
 

tests/patterns/on.py

@@ -2,7 +2,7 @@
 import utime
 from machine import WDT
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 
 
 def main():
@@ -29,7 +29,7 @@ def main():
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < 800:
         wdt.feed()
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(10)
 
     # OFF phase
@@ -37,7 +37,7 @@ def main():
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < 100:
         wdt.feed()
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(10)
 
 

tests/patterns/pulse.py

@@ -2,7 +2,7 @@
 import utime
 from machine import WDT
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 
 
 def run_for(p, wdt, ms):
@@ -10,7 +10,7 @@ def run_for(p, wdt, ms):
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < ms:
         wdt.feed()
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(10)
 
 

tests/patterns/rainbow.py

@@ -2,7 +2,7 @@
 import utime
 from machine import WDT
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 
 
 def run_for(p, wdt, ms):
@@ -10,7 +10,7 @@ def run_for(p, wdt, ms):
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < ms:
         wdt.feed()
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(10)
 
 
@@ -81,7 +81,7 @@ def main():
     for i in range(10):
         p.select("rainbow5")
         # One tick advances the generator one frame when auto=False
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(100)
         wdt.feed()
 
@@ -94,7 +94,7 @@ def main():
     })
     initial_step = p.step
     p.select("rainbow6")
-    p.tick()
+    run_tick(p)
     final_step = p.step
     print(f"Step updated from {initial_step} to {final_step} (expected increment: 1)")
 
@@ -130,7 +130,7 @@ def main():
     p.step = 0
     initial_step = p.step
     p.select("rainbow9")
-    p.tick()
+    run_tick(p)
     final_step = p.step
     expected_step = (initial_step + 5) % 256
     print(f"Step updated from {initial_step} to {final_step} (expected: {expected_step})")

tests/patterns/transition.py

@@ -2,7 +2,7 @@
 import utime
 from machine import WDT
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 
 
 def run_for(p, wdt, ms):
@@ -10,7 +10,7 @@ def run_for(p, wdt, ms):
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < ms:
         wdt.feed()
-        p.tick()
+        run_tick(p)
         utime.sleep_ms(10)
 
 

tests/peers.py

@@ -0,0 +1,25 @@
+from espnow import ESPNow
+import network
+
+sta = network.WLAN(network.STA_IF)
+sta.active(True)
+
+espnow = ESPNow()
+espnow.active(True)
+
+# add_peer() expects a 6-byte MAC (bytes/bytearray), not integers.
+# Unicast placeholders (not broadcast/multicast) so get_peers() lists them.
+# PEERS = aa:aa:aa:aa:aa:START … aa:aa:aa:aa:aa:END (inclusive last octet).
+_PREFIX = b"\xaa\xaa\xaa\xaa\xaa"
+_START_LAST_OCTET = 1
+_END_LAST_OCTET = 40
+PEERS = tuple(_PREFIX + bytes((i,)) for i in range(_START_LAST_OCTET, _END_LAST_OCTET + 1))
+for peer in PEERS:
+    espnow.add_peer(peer)
+
+print("peers:", PEERS)
+
+for peer in PEERS:
+    espnow.send(peer, b"Hello, world!")
+
+print(espnow.get_peers())

tests/test_ap_pm0.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""MicroPython AP example with power management disabled (pm=0).
+
+Run on device:
+  mpremote connect /dev/ttyACM0 run tests/test_ap_pm0.py
+"""
+
+import network
+import time
+
+AP_SSID = "led-ap"
+AP_PASSWORD = "ledpass123"
+AP_CHANNEL = 6
+
+
+def main():
+    ap = network.WLAN(network.AP_IF)
+    ap.active(True)
+
+    # Explicitly disable Wi-Fi power save for AP mode.
+    try:
+        ap.config(pm=0)
+    except (AttributeError, ValueError, TypeError):
+        try:
+            ap.config(pm=network.WLAN.PM_NONE)
+        except (AttributeError, ValueError, TypeError):
+            pass
+
+    ap.config(essid=AP_SSID, password=AP_PASSWORD, channel=AP_CHANNEL, authmode=3)
+
+    print("[ap-pm0] AP active:", ap.active())
+    print("[ap-pm0] SSID:", AP_SSID)
+    print("[ap-pm0] IFCONFIG:", ap.ifconfig())
+    print("[ap-pm0] Waiting for clients. Ctrl+C to stop.")
+
+    while True:
+        time.sleep(2)
+
+
+if __name__ == "__main__":
+    main()

tests/test_espnow_receive.py

@@ -4,7 +4,7 @@ import json
 import os
 import utime
 from settings import Settings
-from presets import Presets
+from presets import Presets, run_tick
 from utils import convert_and_reorder_colors
 
 
@@ -54,7 +54,7 @@ def run_main_loop_iterations(espnow, patterns, settings, wdt, max_iterations=10)
     
     while iterations < max_iterations:
         wdt.feed()
-        patterns.tick()
+        run_tick(patterns)
         
         if espnow.any():
             host, msg = espnow.recv()
@@ -363,7 +363,7 @@ def test_switch_presets():
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < 2000:
         wdt.feed()
-        patterns.tick()
+        run_tick(patterns)
         utime.sleep_ms(10)
     
     # Switch to second preset and run for 2 seconds
@@ -381,7 +381,7 @@ def test_switch_presets():
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < 2000:
         wdt.feed()
-        patterns.tick()
+        run_tick(patterns)
         utime.sleep_ms(10)
     
     # Switch to third preset and run for 2 seconds
@@ -399,7 +399,7 @@ def test_switch_presets():
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < 2000:
         wdt.feed()
-        patterns.tick()
+        run_tick(patterns)
         utime.sleep_ms(10)
     
     # Switch back to first preset and run for 2 seconds
@@ -417,7 +417,7 @@ def test_switch_presets():
     start = utime.ticks_ms()
     while utime.ticks_diff(utime.ticks_ms(), start) < 2000:
         wdt.feed()
-        patterns.tick()
+        run_tick(patterns)
         utime.sleep_ms(10)
     
     print("  ✓ Preset switching works correctly")
@@ -577,7 +577,7 @@ def test_select_with_step():
     mock_espnow.send_message(b"\xbb\xbb\xbb\xbb\xbb\xbb", msg2)
     run_main_loop_iterations(mock_espnow, patterns, settings, wdt, max_iterations=2)
     # Ensure tick() is called after select() to advance the step
-    patterns.tick()
+    run_tick(patterns)
     
     assert patterns.selected == "step_preset", "Should select step_preset"
     # Step is set to 10, then tick() advances it, so it should be 11
@@ -596,7 +596,7 @@ def test_select_with_step():
     initial_step = patterns.step  # Should be 11
     run_main_loop_iterations(mock_espnow, patterns, settings, wdt, max_iterations=2)
     # Ensure tick() is called after select() to advance the step
-    patterns.tick()
+    run_tick(patterns)
     # Since it's the same preset, step should not be reset, but tick() will advance it
     # So step should be initial_step + 1 (one tick call)
     assert patterns.step == initial_step + 1, f"Step should advance from {initial_step} to {initial_step + 1} (not reset), got {patterns.step}"
@@ -614,7 +614,7 @@ def test_select_with_step():
     mock_espnow.send_message(b"\xdd\xdd\xdd\xdd\xdd\xdd", msg4)
     run_main_loop_iterations(mock_espnow, patterns, settings, wdt, max_iterations=2)
     # Ensure tick() is called after select() to advance the step
-    patterns.tick()
+    run_tick(patterns)
     
     assert patterns.selected == "other_preset", "Should select other_preset"
     # Step is set to 5, then tick() advances it, so it should be 6

tests/test_mdns.py

@@ -15,7 +15,7 @@ Deploy src to the device (including utils.py with mdns_hostname), then from the
 
   mpremote connect PORT run tests/test_mdns.py
 
-If ImportError: copy utils.py from src/ to the device, or rely on the built-in fallback below.
+Copy ``utils.py`` from ``src/`` onto the device if imports fail.
 
 Or with cwd led-driver:
 
@@ -30,26 +30,7 @@ import utime
 from machine import WDT
 
 from settings import Settings
-
+from utils import mdns_hostname
-try:
-    from utils import mdns_hostname
-except ImportError:
-
-    def mdns_hostname(settings):
-        """Same as utils.mdns_hostname (fallback if device utils.py is older than host repo)."""
-        raw = settings.get("name") or "led"
-        suffix = []
-        for c in str(raw).lower():
-            o = ord(c)
-            if (48 <= o <= 57) or (97 <= o <= 122):
-                suffix.append(c)
-        s = "".join(suffix)
-        if not s:
-            s = "device"
-        h = "led" + s
-        if len(h) > 32:
-            h = h[:32]
-        return h
 
 CONNECT_TIMEOUT_S = 45
 # ESP32 MicroPython WDT timeout is capped (typically 10000 ms). Longer blocking work
@@ -213,16 +194,6 @@ def main():
             "Set SELF_LOCAL_GETADDRINFO = True to attempt (may hang)."
         )
 
-    # Optional: built-in mdns module (not present on all ESP32 builds)
-    _dbg(t0, "checking for optional 'mdns' module")
-    try:
-        import mdns  # noqa: F401
-
-        print("Note: 'mdns' module is present; check your port's docs for Server/API.")
-    except ImportError:
-        print("No top-level 'mdns' module; relying on stack mDNS from hostname.")
-    _dbg(t0, "mdns import check done")
-
     if HOLD_S != 0:
         forever = HOLD_S < 0
         _dbg(

tests/test_wifi.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+"""Wi-Fi connection smoke test for MicroPython on ESP32.
+
+Runs on-device via mpremote and uses /settings.json credentials.
+
+Usage:
+  mpremote connect /dev/ttyACM0 run tests/test_wifi.py
+"""
+
+import time
+import utime
+import network
+from machine import WDT
+
+from settings import Settings
+
+CONNECT_TIMEOUT_S = 30
+RETRY_DELAY_S = 2
+WDT_TIMEOUT_MS = 10000
+
+
+def _wifi_status_label(code):
+    names = {
+        getattr(network, "STAT_IDLE", 0): "idle",
+        getattr(network, "STAT_CONNECTING", 1): "connecting",
+        getattr(network, "STAT_WRONG_PASSWORD", -3): "wrong_password",
+        getattr(network, "STAT_NO_AP_FOUND", -2): "no_ap_found",
+        getattr(network, "STAT_CONNECT_FAIL", -1): "connect_fail",
+        getattr(network, "STAT_GOT_IP", 3): "got_ip",
+    }
+    return names.get(code, str(code))
+
+
+def connect_wifi_with_wdt(sta, ssid, password, wdt):
+    attempt = 0
+    while not sta.isconnected():
+        attempt += 1
+        print("[wifi-test] attempt", attempt, "ssid=", repr(ssid))
+        try:
+            sta.disconnect()
+        except Exception:
+            pass
+        sta.connect(ssid, password)
+
+        start = utime.time()
+        last_status = None
+        while not sta.isconnected():
+            status = sta.status()
+            if status != last_status:
+                print("[wifi-test] status:", status, _wifi_status_label(status))
+                last_status = status
+            if status in (
+                getattr(network, "STAT_WRONG_PASSWORD", -3),
+                getattr(network, "STAT_NO_AP_FOUND", -2),
+                getattr(network, "STAT_CONNECT_FAIL", -1),
+            ):
+                break
+            if utime.time() - start >= CONNECT_TIMEOUT_S:
+                print("[wifi-test] timeout after", CONNECT_TIMEOUT_S, "seconds")
+                break
+            time.sleep(1)
+            wdt.feed()
+
+        if sta.isconnected():
+            return True
+
+        print("[wifi-test] retry in", RETRY_DELAY_S, "seconds")
+        for _ in range(RETRY_DELAY_S):
+            time.sleep(1)
+            wdt.feed()
+    return True
+
+
+def main():
+    settings = Settings()
+    ssid = settings.get("ssid") or ""
+    password = settings.get("password") or ""
+
+    if not ssid:
+        print("[wifi-test] skipped: settings.ssid is empty")
+        raise SystemExit(0)
+
+    wdt = WDT(timeout=WDT_TIMEOUT_MS)
+    wdt.feed()
+
+    sta = network.WLAN(network.STA_IF)
+    sta.active(True)
+    try:
+        sta.config(pm=network.WLAN.PM_NONE)
+    except (AttributeError, ValueError, TypeError):
+        pass
+
+    ok = connect_wifi_with_wdt(sta, ssid, password, wdt)
+    if not ok or not sta.isconnected():
+        print("[wifi-test] FAILED: not connected")
+        raise SystemExit(1)
+
+    print("[wifi-test] OK:", sta.ifconfig())
+
+
+if __name__ == "__main__":
+    main()