Split into pattern and low level methods

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/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)

src/boot.py

@@ -0,0 +1,9 @@
+import settings
+import wifi
+from settings import Settings
+
+s = Settings()
+
+name = s.get('name', 'led')
+password = s.get("ap_password", "")
+wifi.ap(name, password)

src/main.py

@@ -1,79 +1,54 @@
-
+import asyncio
+import aioespnow
 from settings import Settings
 from web import web
 from patterns import Patterns
 import gc
-import json
-import espnow
-import network
-import asyncio
-import json
+import utime
 import machine
+import time
+import wifi
+import json
+from p2p import p2p
 
-def main():
+async def main():
     settings = Settings()
+
+    patterns = Patterns(settings["led_pin"], settings["num_leds"], selected=settings["pattern"])
+    if settings["color_order"] == "rbg": color_order = (1, 5, 3)
+    else: color_order = (1, 3, 5)
+    patterns.set_color1(tuple(int(settings["color1"][i:i+2], 16) for i in color_order))
+    patterns.set_color2(tuple(int(settings["color2"][i:i+2], 16) for i in color_order))
+    patterns.set_brightness(int(settings["brightness"]))
+    patterns.set_delay(int(settings["delay"]))
+
+    async def tick():
+        while True:
+            patterns.tick()
+            await asyncio.sleep_ms(0)
+
+    async def system():
+        while True:
+            gc.collect()
+            for i in range(60):
+                wdt.feed()
+                await asyncio.sleep(1)
+
+    w = web(settings, patterns)
     print(settings)
-
-    if settings.get("color_order", "rgb") == "rbg":
-        color_order = (1, 5, 3)
-    else:
-        color_order = (1, 3, 5)
-    patterns = Patterns(settings["led_pin"], settings["num_leds"], brightness=255)
-
-    sta_if = network.WLAN(network.STA_IF)
-    sta_if.active(True)
-
-    e = espnow.ESPNow()
-    e.config(rxbuf=1024)
-    e.active(True)
-    # Increase buffer size for 8-bar payloads (default 526 bytes might be too small)  # Set to 1KB to handle larger multi-bar payloads
-
+    # start the server in a bacakground task
+    print("Starting")
+    server = asyncio.create_task(w.start_server(host="0.0.0.0", port=80))
     wdt = machine.WDT(timeout=10000)
     wdt.feed()
 
-    #print mac in hex
-    print("Mac address", sta_if.config("mac").hex())
-    print("Patterns", patterns.colors)
-    print("Patterns", patterns.selected)
+    asyncio.create_task(tick())
+    asyncio.create_task(p2p(settings, patterns))
+    asyncio.create_task(system())
 
-    patterns.select(patterns.selected)
-    while True:
-        # advance pattern based on its own returned schedule
-        # due = patterns.tick(due)
-        wdt.feed()
-        patterns.tick()        
-        # Drain all pending packets and only process the latest
-        last_msg = None
-        while True:
-            
-            host, msg = e.recv(0)
-            if not msg:
-                break
-            last_msg = msg
 
-        if last_msg:
-            try:
-                data = json.loads(last_msg)
-                print(data)
-                
-                # Always update parameters from message
-                patterns.brightness = data.get("brightness", patterns.brightness)
-                patterns.delay = data.get("delay", patterns.delay)
-                patterns.colors = data.get("colors", patterns.colors)
-                patterns.selected = data.get("pattern", patterns.selected)
-                patterns.n1 = data.get("n1", patterns.n1)
-                patterns.n2 = data.get("n2", patterns.n2)
-                patterns.n3 = data.get("n3", patterns.n3)
-                patterns.n4 = data.get("n4", patterns.n4)
-                patterns.step = data.get("step", patterns.step)
-                patterns.auto = data.get("auto", patterns.auto)
 
-                patterns.select(patterns.selected)
-                print("Selected pattern", patterns.selected)
+    # cleanup before ending the application
+    await server
 
-            except Exception as ex:
-                print(f"Failed to load espnow data {last_msg}: {ex}")
-                continue
-            finally:
-                gc.collect()
-main()
+asyncio.run(main())

src/p2p.py

@@ -0,0 +1,20 @@
+import asyncio
+import aioespnow
+import json
+
+async def p2p(settings, patterns):
+    e = aioespnow.AIOESPNow()  # Returns AIOESPNow enhanced with async support
+    e.active(True)
+    async for mac, msg in e:
+        try:
+            data = json.loads(msg)
+        except:
+            print(f"Failed to load espnow data {msg}")
+            continue
+        print(data)
+        if "names" not in data or settings.get("name") in data.get("names", []):
+            if "step" in settings and isinstance(settings["step"], int):
+                patterns.set_pattern_step(settings["step"])
+            else:
+                settings.set_settings(data.get("settings", {}), patterns, data.get("save", False))
+    print("should not print")

src/patterns.py

@@ -22,7 +22,8 @@ class Patterns:
             "flicker": self.flicker_step,
             "scanner": self.scanner_step, # New: Single direction scanner
             "bidirectional_scanner": self.bidirectional_scanner_step, # New: Bidirectional scanner
-            "external": None
+            "external": None,
+            "pulse": self.pulse
         }
         self.selected = selected
         # Ensure colors list always starts with at least two for robust transition handling
@@ -394,3 +395,23 @@ class Patterns:
                 self.pattern_step = 0 # Start moving forward from the first LED
 
             self.last_update = current_time
+
+    def pulse(self):
+        if self.pattern_step == 0:
+            self.fill(self.apply_brightness(self.colors[0]))
+            self.pattern_step = 1
+            self.last_update = utime.ticks_ms()
+        if utime.ticks_diff(utime.ticks_ms(), self.last_update) > self.delay:
+            self.fill((0, 0, 0))
+
+
+if __name__ == "__main__":
+    import time
+    from machine import WDT
+    wdt = WDT(timeout=2000)  # Enable watchdog with a 2 second timeout
+    p = Patterns(pin=10, num_leds=200, color1=(255,0,0), color2=(0,0,255), brightness=127, selected="bidirectional_scanner", delay=50)
+    p.select("pulse")
+    for i in range(1000):
+        p.tick()
+        wdt.feed()
+        time.sleep_ms(1)

src/patterns_base.py

@@ -0,0 +1,194 @@
+from machine import Pin
+from neopixel import NeoPixel
+import utime
+
+
+class PatternBase:
+    def __init__(self, pin, num_leds, color1=(0,0,0), color2=(0,0,0), brightness=127, selected="rainbow_cycle", delay=100):
+        self.n = NeoPixel(Pin(pin, Pin.OUT), num_leds)
+        self.num_leds = num_leds
+        self.pattern_step = 0
+        self.last_update = utime.ticks_ms()
+        self.delay = delay
+        self.brightness = brightness
+        self.patterns = {}
+        self.selected = selected
+        # Ensure colors list always starts with at least two for robust transition handling
+        self.colors = [color1, color2] if color1 != color2 else [color1, (255, 255, 255)] # Fallback if initial colors are same
+        if not self.colors: # Ensure at least one color exists
+            self.colors = [(0, 0, 0)]
+
+        self.transition_duration = delay * 50 # Default transition duration
+        self.hold_duration = delay * 10 # Default hold duration at each color
+        self.transition_step = 0 # Current step in the transition
+        self.current_color_idx = 0 # Index of the color currently being held/transitioned from
+        self.current_color = self.colors[self.current_color_idx] # The actual blended color
+
+        self.hold_start_time = utime.ticks_ms() # Time when the current color hold started
+
+        # New attributes for scanner patterns
+        self.scanner_direction = 1 # 1 for forward, -1 for backward
+        self.scanner_tail_length = 3 # Number of trailing pixels
+
+    def sync(self):
+        self.pattern_step=0
+        self.last_update = utime.ticks_ms() - self.delay
+        if self.selected == "color_transition":
+            self.transition_step = 0
+            self.current_color_idx = 0
+            self.current_color = self.colors[self.current_color_idx]
+            self.hold_start_time = utime.ticks_ms() # Reset hold time
+        # Reset scanner specific variables
+        self.scanner_direction = 1
+        self.tick()
+
+    def set_pattern_step(self, step):
+        self.pattern_step = step
+
+    def tick(self):
+        if self.patterns[self.selected]:
+            self.patterns[self.selected]()
+
+    def update_num_leds(self, pin, num_leds):
+        self.n = NeoPixel(Pin(pin, Pin.OUT), num_leds)
+        self.num_leds = num_leds
+        self.pattern_step = 0
+
+    def set_delay(self, delay):
+        self.delay = delay
+        # Update transition duration and hold duration when delay changes
+        self.transition_duration = self.delay * 50
+        self.hold_duration = self.delay * 10
+
+
+    def set_brightness(self, brightness):
+        self.brightness = brightness
+
+    def set_color1(self, color):
+        if len(self.colors) > 0:
+            self.colors[0] = color
+            if self.selected == "color_transition":
+                # If the first color is changed, potentially reset transition
+                # to start from this new color if we were about to transition from it
+                if self.current_color_idx == 0:
+                    self.transition_step = 0
+                    self.current_color = self.colors[0]
+                    self.hold_start_time = utime.ticks_ms()
+        else:
+            self.colors.append(color)
+
+
+    def set_color2(self, color):
+        if len(self.colors) > 1:
+            self.colors[1] = color
+        elif len(self.colors) == 1:
+            self.colors.append(color)
+        else: # List is empty
+            self.colors.append((0,0,0)) # Dummy color
+            self.colors.append(color)
+
+
+    def set_colors(self, colors):
+        if colors and len(colors) >= 2:
+            self.colors = colors
+            if self.selected == "color_transition":
+                self.sync() # Reset transition if new color list is provided
+        elif colors and len(colors) == 1:
+            self.colors = [colors[0], (255,255,255)] # Add a default second color
+            if self.selected == "color_transition":
+                print("Warning: 'color_transition' requires at least two colors. Adding a default second color.")
+                self.sync()
+        else:
+            print("Error: set_colors requires a list of at least one color.")
+            self.colors = [(0,0,0), (255,255,255)] # Fallback
+            if self.selected == "color_transition":
+                self.sync()
+
+    def set_color(self, num, color):
+        # Changed: More robust index check
+        if 0 <= num < len(self.colors):
+           self.colors[num] = color
+           # If the changed color is part of the current or next transition,
+           # restart the transition for smoother updates
+           if self.selected == "color_transition":
+               current_from_idx = self.current_color_idx
+               current_to_idx = (self.current_color_idx + 1) % len(self.colors)
+               if num == current_from_idx or num == current_to_idx:
+                   # If we change a color involved in the current transition,
+                   # it's best to restart the transition state for smoothness.
+                   self.transition_step = 0
+                   self.current_color_idx = current_from_idx # Stay at the current starting color
+                   self.current_color = self.colors[self.current_color_idx]
+                   self.hold_start_time = utime.ticks_ms() # Reset hold
+           return True
+        elif num == len(self.colors): # Allow setting a new color at the end
+            self.colors.append(color)
+            return True
+        return False
+
+    def add_color(self, color):
+        self.colors.append(color)
+        if self.selected == "color_transition" and len(self.colors) == 2:
+            # If we just added the second color needed for transition
+            self.sync()
+
+
+    def del_color(self, num):
+        # Changed: More robust index check and using del for lists
+        if 0 <= num < len(self.colors):
+           del self.colors[num]
+           # If the color being deleted was part of the current transition,
+           # re-evaluate the current_color_idx
+           if self.selected == "color_transition":
+               if len(self.colors) < 2: # Need at least two colors for transition
+                   print("Warning: Not enough colors for 'color_transition'. Switching to 'on'.")
+                   self.select("on") # Or some other default
+               else:
+                   # Adjust index if it's out of bounds after deletion or was the one transitioning from
+                   self.current_color_idx %= len(self.colors)
+                   self.transition_step = 0
+                   self.current_color = self.colors[self.current_color_idx]
+                   self.hold_start_time = utime.ticks_ms()
+           return True
+        return False
+
+    def apply_brightness(self, color, brightness_override=None):
+        effective_brightness = brightness_override if brightness_override is not None else self.brightness
+        return tuple(int(c * effective_brightness / 255) for c in color)
+
+    def select(self, pattern):
+        if pattern in self.patterns:
+            self.selected = pattern
+            self.sync() # Reset pattern state when selecting a new pattern
+            if pattern == "color_transition":
+                if len(self.colors) < 2:
+                    print("Warning: 'color_transition' requires at least two colors. Switching to 'on'.")
+                    self.selected = "on" # Fallback if not enough colors
+                    self.sync() # Re-sync for the new pattern
+                else:
+                    self.transition_step = 0
+                    self.current_color_idx = 0 # Start from the first color in the list
+                    self.current_color = self.colors[self.current_color_idx]
+                    self.hold_start_time = utime.ticks_ms() # Reset hold timer
+                    self.transition_duration = self.delay * 50 # Initialize transition duration
+                    self.hold_duration = self.delay * 10 # Initialize hold duration
+            return True
+        return False
+
+    def set(self, i, color):
+        self.n[i] = color
+
+    def write(self):
+        self.n.write()
+
+    def fill(self, color=None):
+        fill_color = color if color is not None else self.colors[0]
+        for i in range(self.num_leds):
+            self.n[i] = fill_color
+        self.n.write()
+
+    def off(self):
+        self.fill((0, 0, 0))
+
+    def on(self):
+        self.fill(self.apply_brightness(self.colors[0]))

src/static/main.css

@@ -0,0 +1,109 @@
+body {
+    font-family: Arial, sans-serif;
+    max-width: 600px;
+    margin: 0 auto;
+    padding: 20px;
+    line-height: 1.6;
+}
+h1 {
+    text-align: center;
+}
+
+form {
+    margin-bottom: 20px;
+}
+label {
+    display: block;
+    margin-bottom: 5px;
+}
+input[type="text"],
+input[type="submit"],
+input[type="range"],
+input[type="color"] {
+    width: 100%;
+
+    margin-bottom: 10px;
+    box-sizing: border-box;
+}
+input[type="range"] {
+    -webkit-appearance: none;
+    appearance: none;
+    height: 25px;
+    background: #d3d3d3;
+    outline: none;
+    opacity: 0.7;
+    transition: opacity 0.2s;
+}
+input[type="range"]:hover {
+    opacity: 1;
+}
+input[type="range"]::-webkit-slider-thumb {
+    -webkit-appearance: none;
+    appearance: none;
+    width: 25px;
+    height: 25px;
+    background: #4caf50;
+    cursor: pointer;
+    border-radius: 50%;
+}
+input[type="range"]::-moz-range-thumb {
+    width: 25px;
+    height: 25px;
+    background: #4caf50;
+    cursor: pointer;
+    border-radius: 50%;
+}
+#pattern_buttons {
+    display: flex;
+    flex-wrap: wrap;
+    gap: 10px;
+    margin-bottom: 20px;
+}
+#pattern_buttons button {
+    flex: 1 0 calc(33.333% - 10px);
+    padding: 10px;
+    background-color: #4caf50;
+    color: white;
+    border: none;
+    cursor: pointer;
+    transition: background-color 0.3s;
+}
+#pattern_buttons button:hover {
+    background-color: #45a049;
+}
+@media (max-width: 480px) {
+    #pattern_buttons button {
+        flex: 1 0 calc(50% - 10px);
+    }
+}
+#connection-status {
+    width: 15px;
+    height: 15px;
+    border-radius: 50%;
+    display: inline-block; /* Or block, depending on where you put it */
+    margin-left: 10px; /* Adjust spacing as needed */
+    vertical-align: middle; /* Align with nearby text */
+    background-color: grey; /* Default: Unknown */
+}
+
+#connection-status.connecting {
+    background-color: yellow;
+}
+
+#connection-status.open {
+    background-color: green;
+}
+
+#connection-status.closing,
+#connection-status.closed {
+    background-color: red;
+}
+
+#color_order_form label,
+#color_order_form input[type="radio"] {
+    /* Ensures they behave as inline elements */
+    display: inline-block;
+    /* Adds some space between them for readability */
+    margin-right: 10px;
+    vertical-align: middle; /* Aligns them nicely if heights vary */
+}

src/static/main.js

@@ -0,0 +1,244 @@
+let delayTimeout;
+let brightnessTimeout;
+let colorTimeout;
+let color2Timeout;
+let ws; // Variable to hold the WebSocket connection
+let connectionStatusElement; // Variable to hold the connection status element
+
+// Function to update the connection status indicator
+function updateConnectionStatus(status) {
+  if (!connectionStatusElement) {
+    connectionStatusElement = document.getElementById("connection-status");
+  }
+  if (connectionStatusElement) {
+    connectionStatusElement.className = ""; // Clear existing classes
+    connectionStatusElement.classList.add(status);
+    // Optionally, you could also update text content based on status
+    // connectionStatusElement.textContent = status.charAt(0).toUpperCase() + status.slice(1);
+  }
+}
+
+// Function to establish WebSocket connection
+function connectWebSocket() {
+  // Determine the WebSocket URL based on the current location
+  const wsUrl = `ws://${window.location.host}/ws`;
+  ws = new WebSocket(wsUrl);
+
+  updateConnectionStatus("connecting"); // Indicate connecting state
+
+  ws.onopen = function (event) {
+    console.log("WebSocket connection opened:", event);
+    updateConnectionStatus("open"); // Indicate open state
+    // Optionally, you could send an initial message here
+  };
+
+  ws.onmessage = function (event) {
+    console.log("WebSocket message received:", event.data);
+  };
+
+  ws.onerror = function (event) {
+    console.error("WebSocket error:", event);
+    updateConnectionStatus("closed"); // Indicate error state (treat as closed)
+  };
+
+  ws.onclose = function (event) {
+    if (event.wasClean) {
+      console.log(
+        `WebSocket connection closed cleanly, code=${event.code}, reason=${event.reason}`,
+      );
+      updateConnectionStatus("closed"); // Indicate closed state
+    } else {
+      console.error("WebSocket connection died");
+      updateConnectionStatus("closed"); // Indicate closed state
+    }
+    // Attempt to reconnect after a delay
+    setTimeout(connectWebSocket, 1000);
+  };
+}
+
+// Function to send data over WebSocket
+function sendWebSocketData(data) {
+  if (ws && ws.readyState === WebSocket.OPEN) {
+    console.log("Sending data over WebSocket:", data);
+    ws.send(JSON.stringify(data));
+  } else {
+    console.error("WebSocket is not connected. Cannot send data:", data);
+    // You might want to queue messages or handle this in a different way
+  }
+}
+
+// Keep the post and get functions for now, they might still be useful
+async function post(path, data) {
+  console.log(`POST to ${path}`, data);
+  try {
+    const response = await fetch(path, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify(data),
+    });
+    if (!response.ok) {
+      throw new Error(`HTTP error! Status: ${response.status}`);
+    }
+  } catch (error) {
+    console.error("Error during POST request:", error);
+  }
+}
+
+async function get(path) {
+  try {
+    const response = await fetch(path);
+    if (!response.ok) {
+      throw new Error(`HTTP error! Status: ${response.status}`);
+    }
+    return await response.json();
+  } catch (error) {
+    console.error("Error during GET request:", error);
+  }
+}
+
+async function updateColor(event) {
+  event.preventDefault();
+  clearTimeout(colorTimeout);
+  colorTimeout = setTimeout(function () {
+    const color = document.getElementById("color").value;
+    sendWebSocketData({ color1: color });
+  }, 500);
+}
+
+async function updateColor2(event) {
+  event.preventDefault();
+  clearTimeout(color2Timeout);
+  color2Timeout = setTimeout(function () {
+    const color = document.getElementById("color2").value;
+    sendWebSocketData({ color2: color });
+  }, 500);
+}
+
+async function updatePattern(pattern) {
+  sendWebSocketData({ pattern: pattern });
+}
+
+async function updateBrightness(event) {
+  event.preventDefault();
+  clearTimeout(brightnessTimeout);
+  brightnessTimeout = setTimeout(function () {
+    const brightness = document.getElementById("brightness").value;
+    sendWebSocketData({ brightness: brightness });
+  }, 500);
+}
+
+async function updateDelay(event) {
+  event.preventDefault();
+  clearTimeout(delayTimeout);
+  delayTimeout = setTimeout(function () {
+    const delay = document.getElementById("delay").value;
+    sendWebSocketData({ delay: delay });
+  }, 500);
+}
+
+async function updateNumLeds(event) {
+  event.preventDefault();
+  const numLeds = document.getElementById("num_leds").value;
+  sendWebSocketData({ num_leds: parseInt(numLeds) });
+}
+
+async function updateName(event) {
+  event.preventDefault();
+  const name = document.getElementById("name").value;
+  sendWebSocketData({ name: name });
+}
+
+async function updateID(event) {
+  event.preventDefault();
+  const id = document.getElementById("id").value;
+  sendWebSocketData({ id: parseInt(id) });
+}
+
+async function updateLedPin(event) {
+  event.preventDefault();
+  const ledpin = document.getElementById("led_pin").value;
+  sendWebSocketData({ led_pin: parseInt(ledpin) });
+}
+
+function handleRadioChange(event) {
+  event.preventDefault();
+  console.log("Selected color order:", event.target.value);
+  // Add your specific logic here
+  if (event.target.value === "rgb") {
+    console.log("RGB order selected!");
+  } else if (event.target.value === "rbg") {
+    console.log("RBG order selected!");
+  }
+  sendWebSocketData({ color_order: event.target.value });
+}
+
+function createPatternButtons(patterns) {
+  const container = document.getElementById("pattern_buttons");
+  container.innerHTML = ""; // Clear previous buttons
+
+  patterns.forEach((pattern) => {
+    const button = document.createElement("button");
+    button.type = "button";
+    button.textContent = pattern;
+    button.value = pattern;
+    button.addEventListener("click", async function (event) {
+      event.preventDefault();
+      await updatePattern(pattern);
+    });
+    container.appendChild(button);
+  });
+}
+
+document.addEventListener("DOMContentLoaded", async function () {
+  // Get the connection status element once the DOM is ready
+  connectionStatusElement = document.getElementById("connection-status");
+
+  // Establish WebSocket connection on page load
+  connectWebSocket();
+
+  document.getElementById("color").addEventListener("input", updateColor);
+  document.getElementById("color2").addEventListener("input", updateColor2);
+  document.getElementById("delay").addEventListener("input", updateDelay);
+  document
+    .getElementById("brightness")
+    .addEventListener("input", updateBrightness);
+  document
+    .getElementById("num_leds_form")
+    .addEventListener("submit", updateNumLeds);
+  document.getElementById("name_form").addEventListener("submit", updateName);
+  document.getElementById("id_form").addEventListener("submit", updateID);
+  document
+    .getElementById("led_pin_form")
+    .addEventListener("submit", updateLedPin);
+  document.getElementById("delay").addEventListener("touchend", updateDelay);
+  document
+    .getElementById("brightness")
+    .addEventListener("touchend", updateBrightness);
+
+  document.getElementById("rgb").addEventListener("change", handleRadioChange);
+  document.getElementById("rbg").addEventListener("change", handleRadioChange);
+  document.querySelectorAll(".pattern_button").forEach((button) => {
+    console.log(button.value);
+    button.addEventListener("click", async (event) => {
+      event.preventDefault();
+      await updatePattern(button.value);
+    });
+  });
+});
+
+// Function to toggle the display of the settings menu
+function selectSettings() {
+  const settingsMenu = document.getElementById("settings_menu");
+  controls = document.getElementById("controls");
+  settingsMenu.style.display = "block";
+  controls.style.display = "none";
+}
+
+function selectControls() {
+  const settingsMenu = document.getElementById("settings_menu");
+  controls = document.getElementById("controls");
+  settingsMenu.style.display = "none";
+  controls.style.display = "block";
+}

src/templates/index.html

@@ -0,0 +1,124 @@
+{% args settings, patterns, mac %}
+<!doctype html>
+<html lang="en">
+    <head>
+        <meta charset="UTF-8" />
+        <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+        <title>{{settings['name']}}</title>
+        <script src="static/main.js"></script>
+        <link rel="stylesheet" href="static/main.css" />
+    </head>
+    <body>
+        <h1>{{settings['name']}}</h1>
+        <button onclick="selectControls()">Controls</button>
+        <button onclick="selectSettings()">Settings</button>
+
+        <!-- Main LED Controls -->
+        <div id="controls">
+            <div id="pattern_buttons">
+                {% for p in patterns %}
+                <button class="pattern_button" value="{{p}}">{{p}}</button>
+                {% endfor %}
+
+                <!-- Pattern buttons will be inserted here -->
+            </div>
+            <form id="delay_form" method="post" action="/delay">
+                <label for="delay">Delay:</label>
+                <input
+                    type="range"
+                    id="delay"
+                    name="delay"
+                    min="1"
+                    max="1000"
+                    value="{{settings['delay']}}"
+                    step="10"
+                />
+            </form>
+            <form id="brightness_form" method="post" action="/brightness">
+                <label for="brightness">Brightness:</label>
+                <input
+                    type="range"
+                    id="brightness"
+                    name="brightness"
+                    min="0"
+                    max="100"
+                    value="{{settings['brightness']}}"
+                    step="1"
+                />
+            </form>
+            <form id="color_form" method="post" action="/color">
+                <input
+                    type="color"
+                    id="color"
+                    name="color"
+                    value="{{settings['color1']}}"
+                />
+            </form>
+            <form id="color2_form" method="post" action="/color2">
+                <input
+                    type="color"
+                    id="color2"
+                    name="color2"
+                    value="{{settings['color2']}}"
+                />
+            </form>
+        </div>
+
+        <!-- Settings Menu for num_leds, Wi-Fi SSID, and Password -->
+
+        <div id="settings_menu" style="display: none">
+            <h2>Settings</h2>
+
+            <form id="name_form" method="post" action="/name">
+                <label for="name">Name:</label>
+                <input
+                    type="text"
+                    id="name"
+                    name="num_leds"
+                    value="{{settings['name']}}"
+                />
+                <input type="submit" value="Update Name" />
+            </form>
+            <form id="id_form" method="post" action="/id">
+                <label for="id">ID:</label>
+                <input
+                    type="text"
+                    id="id"
+                    name="id"
+                    value="{{settings['id']}}"
+                />
+                <input type="submit" value="Update ID" />
+            </form>
+            <!-- Separate form for submitting num_leds -->
+            <form id="num_leds_form" method="post" action="/num_leds">
+                <label for="num_leds">Number of LEDs:</label>
+                <input
+                    type="text"
+                    id="num_leds"
+                    name="num_leds"
+                    value="{{settings['num_leds']}}"
+                />
+                <input type="submit" value="Update Number of LEDs" />
+            </form>
+            <form id="led_pin_form" method="post" action="/led_pin">
+                <label for="num_leds">Led pin:</label>
+                <input
+                    type="text"
+                    id="led_pin"
+                    name="led_pin"
+                    value="{{settings['led_pin']}}"
+                />
+                <input type="submit" value="Update Led Pin" />
+            </form>
+            <form id="color_order_form">
+                <label for="rgb">RGB:</label>
+                <input type="radio" id="rgb" name="color_order" value="rgb" {{'checked' if settings["color_order"]=="rgb" else ''}} />
+                <label for="rbg">RBG</label>
+                <input type="radio" id="rbg" name="color_order" value="rbg" {{'checked' if settings["color_order"]=="rbg" else ''}}/>
+            </form>
+
+            <p>Mac address: {{mac}}</p>
+        </div>
+        <div id="connection-status"></div>
+    </body>
+</html>

src/web.py

@@ -0,0 +1,43 @@
+from microdot import Microdot, send_file, Response
+from microdot.utemplate import Template
+from microdot.websocket import with_websocket
+import machine
+import wifi
+import json
+
+def web(settings, patterns):
+    app = Microdot()
+    Response.default_content_type = 'text/html'
+
+    @app.route('/')
+    async def index_hnadler(request):
+        mac = wifi.get_mac().hex()
+        return Template('/index.html').render(settings=settings, patterns=patterns.patterns.keys(), mac=mac)
+
+    @app.route("/static/<path:path>")
+    def static_handler(request, path):
+        if '..' in path:
+            # Directory traversal is not allowed
+            return 'Not found', 404
+        return send_file('static/' + path)
+
+    @app.post("/settings")
+    def settings_handler(request):
+        # Keep the POST handler for compatibility or alternative usage if needed
+        # For WebSocket updates, the /ws handler is now primary
+        return settings.set_settings(request.body.decode('utf-8'), patterns)
+
+    @app.route("/ws")
+    @with_websocket
+    async def ws(request, ws):
+        while True:
+            data = await ws.receive()
+            if data:
+
+                # Process the received data
+                _, status_code = settings.set_settings(json.loads(data), patterns, True)
+                #await ws.send(status_code)
+            else:
+                break
+
+    return app

src/wifi.py

@@ -0,0 +1,39 @@
+import network
+from time import sleep
+
+def connect(ssid, password, ip, gateway):
+
+    try:
+        sta_if = network.WLAN(network.STA_IF)
+        if not sta_if.isconnected():
+            if ssid == "" or password == "":
+                print("Missing ssid or password")
+                return None
+            if ip != "" and gateway != "":
+                sta_if.ifconfig((ip, '255.255.255.0', gateway, '1.1.1.1'))
+            print('connecting to network...')
+            sta_if.active(True)
+            sta_if.connect(ssid, password)
+            sleep(0.1)
+            if sta_if.isconnected():
+                return sta_if.ifconfig()
+            return None
+        return sta_if.ifconfig()
+    except Exception as e:
+        print(f"Failed to connect to wifi {e}")
+        return None
+
+
+def ap(ssid, password):
+    ap_if = network.WLAN(network.AP_IF)
+    ap_mac = ap_if.config('mac')
+    print(ssid)
+    ap_if.active(True)
+    ap_if.config(essid=ssid, password=password)
+    ap_if.active(False)
+    ap_if.active(True)
+    print(ap_if.ifconfig())
+
+def get_mac():
+    ap_if = network.WLAN(network.AP_IF)
+    return ap_if.config('mac')