Patternflow

"""Console pages: edit them as HTML, ship them as PROGMEM — gzipped.

The device serves each console page from a C string literal in a header —
the right thing to ship, the wrong thing to work on. A page inside
`R"HTML(...)HTML"` cannot be opened in a browser, cannot be saved and
reloaded, and cannot be handed to someone who designs in HTML. Iterating on
one meant build, flash, look, repeat.

So the HTML lives in console/*.html, and this splices it back:

    python firmware/toolchain/console_pages.py extract   # .h  -> .html
    python firmware/toolchain/console_pages.py build     # .html -> .h (+gzip)
    python firmware/toolchain/console_pages.py check     # are they in sync?

Pair it with console_serve.py and the loop becomes save-and-refresh.

`build` also lays a gzip twin right under every literal it manages:

    // <gz:NAME> GENERATED by firmware/toolchain/console_pages.py build — do not edit.
    // gzip -9 of NAME (26551 bytes → 8634 bytes), served with Content-Encoding: gzip.
    const uint8_t NAME_GZ[] PROGMEM = {
      0x1f,0x8b,...
    };
    constexpr size_t NAME_GZ_LEN = sizeof(NAME_GZ);
    // </gz:NAME>

The twin is what the HTTP handlers send. The reason is the link, not flash:
a panel whose Wi-Fi hop was measured moving 2–5 KB/s (round trips of ~0.4 s
with retransmissions, and a transmit queue shared with everything else the
panel sends) took 5–7 s over a 26 KB page and never finished a 63 KB one
inside the sender's old 5 s budget — the browser got a page cut off
mid-script. Gzipped they are a third the size, so a third the time on
exactly the links where time is the whole cost. Compressing on the device is out
(deflate's working set is a few hundred KB; the runtime heap is a few tens),
so the desk does it once, here, and the header carries the result.

The two JS bundles the pages load (ASSETS below) get the same block. They
have no .html twin — the literal IS their source — so for them `build` only
regenerates the gzip and `check` only verifies it; `extract` skips them.

Two things about the gzip block that look fussy and are not:

  - The payload is the literal with CRLF folded to LF. GCC folds line
    endings inside raw strings, so that is the byte sequence the device
    serves raw (the sizes in the comment match sizeof(NAME)-1), and it is
    also what git stores: core.autocrlf on a Windows desk gives a CRLF
    working tree over an LF index. Compressing the on-disk bytes would make
    CI's check fail for every header built on Windows.
  - `check` verifies by DECOMPRESSING the block and comparing to the
    literal, never by recompressing and comparing bytes: zlib versions
    differ between desks and CI and emit different, equally valid streams
    for the same input. mtime is pinned to 0 so a machine's output is at
    least stable with itself.

This replaces ONLY the bytes between `R"HTML(` and `)HTML";` and the
<gz:…> block right after them. Everything else in the header — the doc
comment explaining what the page is for, the includes, `static` or not —
is left exactly as it was. That matters: those comments carry design intent
and attribution that a whole-file generator would quietly delete, and the
storage class is not this tool's business (the twin copies whatever the
literal has).

`extract` is lossless by construction (build right after it is a no-op),
and `build` is idempotent.
"""

import gzip
import os
import re
import sys
import zlib

HERE = os.path.dirname(os.path.abspath(__file__))
SKETCH = os.path.normpath(os.path.join(HERE, "..", "patternflow"))
HTML_DIR = os.path.join(SKETCH, "console")

# page name -> header path, relative to the sketch. Order is the console's
# own nav order, so `check` output reads like the site map.
PAGES = [
    ("home", "src/home_index.h"),
    ("patterns", "src/patterns_index.h"),
    ("status", "src/status_index.h"),
    ("wifi", "src/wifi_index.h"),
    ("knobs", "src/knobs_index.h"),
    ("update", "src/web_update_index.h"),
    ("show", "features/show/show_index.h"),
    ("weather", "features/weather/weather_index.h"),
    ("mqtt", "features/mqtt/mqtt_index.h"),
    ("audio-in", "features/audio_in/audio_in_index.h"),
    ("midi", "features/midi/midi_index.h"),
    ("clock", "features/clock/clock_index.h"),
]

# The JS the pages load beside themselves: URL name, header, and the raw
# string delimiter that header uses (each picked one that cannot occur in
# its own JS). No .html twin — the literal is the source, so only the gzip
# block below it is this tool's to write.
ASSETS = [
    ("pf-console.js", "src/theme_index.h", "JS"),
    ("fflate.js", "src/fflate_js.h", "FFLATE"),
]

PAGE_DELIM = "HTML"

# Dropped into the header's existing comment block, once, so someone opening
# the .h is told where the real source is before they start editing PROGMEM.
NOTE = (
    "// The page body below is generated from console/{name}.html — edit that,\n"
    "// then run: python firmware/toolchain/console_pages.py build\n"
)
NOTE_MARK = "generated from console/"

# The literal's declaration, ending in R"DELIM(. The gzip twin copies the
# storage class from here and takes its name from the identifier.
DECL = r'(static const|const)\s+char\s+(\w+)\s*\[\]\s*PROGMEM\s*=\s*R"%s\($'


def header_path(rel):
    return os.path.join(SKETCH, rel.replace("/", os.sep))


def html_path(name):
    return os.path.join(HTML_DIR, name + ".html")


def read(path):
    with open(path, encoding="utf-8", newline="") as f:
        return f.read()


def write(path, text):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8", newline="") as f:
        f.write(text)


def opener(delim):
    return 'R"%s(' % delim


def closer(delim):
    return ')%s";' % delim


def newline_of(text):
    return "\r\n" if "\r\n" in text else "\n"


def targets():
    """(name, header rel, delimiter, html path or None) for everything built."""
    for name, rel in PAGES:
        yield name, rel, PAGE_DELIM, html_path(name)
    for name, rel, delim in ASSETS:
        yield name, rel, delim, None


def split(text, name, delim=PAGE_DELIM):
    """prefix, body, suffix — the literal's contents and everything around it."""
    OPEN, CLOSE = opener(delim), closer(delim)
    a = text.find(OPEN)
    if a < 0:
        raise SystemExit("%s: no %s in the header" % (name, OPEN))
    a += len(OPEN)
    b = text.find(CLOSE, a)
    if b < 0:
        raise SystemExit("%s: no %s closing the literal" % (name, CLOSE))
    if text.find(OPEN, b) >= 0:
        raise SystemExit("%s: more than one %s literal — splice by hand" % (name, delim))
    return text[:a], text[a:b], text[b:]


def declaration(prefix, name, delim):
    """(storage class, identifier) of the literal the prefix ends in."""
    m = re.search(DECL % delim, prefix, re.M)
    if not m:
        raise SystemExit(
            '%s: cannot read the declaration — expected `const char NAME[] '
            'PROGMEM = R"%s(` (static or not) on one line' % (name, delim))
    return m.group(1), m.group(2)


def with_note(prefix, name):
    """Add the where-to-edit note to the header's comment block, once."""
    if NOTE_MARK in prefix:
        return prefix
    note = NOTE.format(name=name)
    # Land it at the end of the leading comment block, above #pragma once,
    # so it reads as the last thing the file says about itself.
    m = re.search(r"^#pragma once", prefix, re.M)
    if not m:
        return prefix
    nl = "\r\n" if "\r\n" in prefix else "\n"
    if nl == "\r\n":
        note = note.replace("\n", "\r\n")
    return prefix[: m.start()] + note + prefix[m.start() :]


# ── the gzip twin ──────────────────────────────────────────────────────────

def payload(body):
    """The bytes the device serves for this literal — CRLF folded to LF, as
    GCC does inside a raw string. See the docstring for why not the on-disk
    bytes."""
    return body.replace("\r\n", "\n").encode("utf-8")


def gz_open(ident):
    return "// <gz:%s>" % ident


def gz_close(ident):
    return "// </gz:%s>" % ident


def gz_block(ident, storage, data, nl):
    """(block text, compressed size) in the header's own newline style."""
    gz = gzip.compress(data, compresslevel=9, mtime=0)
    lines = [
        "%s GENERATED by firmware/toolchain/console_pages.py build — do not edit."
        % gz_open(ident),
        "// gzip -9 of %s (%d bytes → %d bytes), served with Content-Encoding: gzip."
        % (ident, len(data), len(gz)),
        "%s uint8_t %s_GZ[] PROGMEM = {" % (storage, ident),
    ]
    for i in range(0, len(gz), 16):
        lines.append("  " + ",".join("0x%02x" % b for b in gz[i:i + 16]) + ",")
    lines += [
        "};",
        "constexpr size_t %s_GZ_LEN = sizeof(%s_GZ);" % (ident, ident),
        gz_close(ident),
    ]
    return nl.join(lines) + nl, len(gz)


def split_suffix(suffix, ident, name):
    """close_line, block, rest — the literal's closing line, the gzip block
    already in the header (None if absent), and whatever else is there.

    The block is found by its markers wherever it sits, so a header whose
    block was moved or hand-edited still comes back with exactly one, in its
    place, on the next build."""
    nl_at = suffix.find("\n")
    if nl_at < 0:
        close_line, rest = suffix, ""  # the file ends on the closing line
    else:
        close_line, rest = suffix[: nl_at + 1], suffix[nl_at + 1 :]
    a = rest.find(gz_open(ident))
    if a < 0:
        return close_line, None, rest
    b = rest.find(gz_close(ident), a)
    if b < 0:
        raise SystemExit("%s: %s has no closing %s — repair by hand"
                         % (name, gz_open(ident), gz_close(ident)))
    e = rest.find("\n", b)
    e = len(rest) if e < 0 else e + 1
    return close_line, rest[a:e], rest[:a] + rest[e:]


def assemble(prefix, body, suffix, ident, storage, nl, name):
    """(header text, compressed size): prefix + body + closing line + a fresh
    gzip block + whatever else followed."""
    close_line, _, rest = split_suffix(suffix, ident, name)
    if not close_line.endswith("\n"):
        close_line += nl
    block, gz_len = gz_block(ident, storage, payload(body), nl)
    return prefix + body + close_line + block + rest, gz_len


def gz_problem(block, ident, storage, body):
    """None if the block stands for this literal, else what is wrong with it.

    Decompress and compare — never recompress and compare bytes: zlib
    versions differ between desks and CI and produce different, equally
    valid streams for the same input."""
    if block is None:
        return "no gz block"
    if ("%s uint8_t %s_GZ[] PROGMEM = {" % (storage, ident)) not in block:
        return "gz block declares another storage class or name"
    m = re.search(r"\{(.*?)\};", block, re.S)
    if not m:
        return "gz block has no byte array"
    data = bytes(int(h, 16) for h in re.findall(r"0x([0-9a-fA-F]{2})", m.group(1)))
    try:
        if gzip.decompress(data) != payload(body):
            return "gz block is not this literal"
    except (OSError, EOFError, zlib.error) as e:
        return "gz block does not decompress (%s)" % e
    return None


# ── commands ───────────────────────────────────────────────────────────────

def cmd_extract():
    for name, rel in PAGES:
        _, body, _ = split(read(header_path(rel)), name)
        write(html_path(name), body)
        print("  %-9s <- %s" % (name + ".html", rel))
    print("\n%d pages extracted. `build` now is a no-op apart from the" % len(PAGES))
    print("edit-me note and the gzip twins; anything else in the diff is a bug")
    print("in this script.")


def cmd_build():
    changed = 0
    for name, rel, delim, src in targets():
        dest = header_path(rel)
        old = read(dest)
        prefix, body, suffix = split(old, name, delim)
        storage, ident = declaration(prefix, name, delim)
        if src is not None:
            if not os.path.exists(src):
                raise SystemExit(name + ": " + src + " missing (run extract first)")
            prefix, body = with_note(prefix, name), read(src)
        new, gz_len = assemble(prefix, body, suffix, ident, storage, newline_of(old), name)
        status = "up to date"
        if new != old:
            write(dest, new)
            changed += 1
            status = "wrote " + rel
        print("  %-13s %6d -> %6d bytes gz  %s" % (name, len(payload(body)), gz_len, status))
    print(("%d header(s) updated" % changed) if changed else "already up to date")


def cmd_check():
    stale = []
    for name, rel, delim, src in targets():
        prefix, body, suffix = split(read(header_path(rel)), name, delim)
        storage, ident = declaration(prefix, name, delim)
        if src is not None:
            if not os.path.exists(src):
                stale.append(name + " (no .html)")
                continue
            if body != read(src) or with_note(prefix, name) != prefix:
                stale.append(name)
                continue
        _, block, _ = split_suffix(suffix, ident, name)
        problem = gz_problem(block, ident, storage, body)
        if problem:
            stale.append("%s (%s)" % (name, problem))
    if stale:
        print("out of sync: " + ", ".join(stale))
        print("run: python firmware/toolchain/console_pages.py build")
        return 1
    print("all %d console headers match their HTML; all %d gzip twins "
          "decompress to their literals" % (len(PAGES), len(PAGES) + len(ASSETS)))
    return 0


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else ""
    if cmd == "extract":
        cmd_extract()
    elif cmd == "build":
        cmd_build()
    elif cmd == "check":
        sys.exit(cmd_check())
    else:
        print(__doc__)
        sys.exit(2)