#!/usr/bin/env python3 """ compress-fseq.py — Gothic Folly power compressor (issue #113) Takes a finished xLights .fseq, expands the rose in-memory, and writes a corrected, expanded .fseq in which — when driven at a target port cap — no PSU exceeds its rating (Rule 2) and the whole system stays under the generator budget (Rule 3), while preserving the show's look as much as possible. Runs once in show-prep, right where expand-fseq sits. Policy (locked with John 2026-07-27) — four Bright Sets, each scaled in unison: A = spires, quads, canopy, spirelets, wash B = minor arches C = major arches D = rose petals Rule 1 each set reduced by one gain, applied to every element (keeps symmetry). Rule 2 (per-PSU, first) each set's gain = worst PSU in it → rating/load, so no PSU in the set exceeds its amp rating at the target cap. Rule 3 (generator, after) if total draw still exceeds the generator budget, spread a further reduction across sets weighted A>B>C>D, NO floors — a set may hit 0, then the remainder rolls to the next set (graduated → cascade for severe frames). Protection order: rose > major > minor > A. Smoothing instant attack (protect now), slow release (no pumping on recovery). Usage: python3 pixel-map/compress-fseq.py SOURCE.fseq [OUT.fseq] [--cap 1.0] [--zstd] --cap C port cap you intend to run at (default 1.0 = compressor is the sole limiter; set lower if the F48V5 port cap still provides a floor) --headroom H target H×rating per PSU (default 1.0 = the rating; <1 leaves fuse margin) --gen W generator budget in DC watts (default = placeholder below) --dry-run analyze + report only, don't write Shares the PSU map / power model with psu_model.py. """ import sys, os, importlib.util HERE = os.path.dirname(os.path.abspath(__file__)) def _load(name, fn): spec = importlib.util.spec_from_file_location(name, os.path.join(HERE, fn)) m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m); return m PM = _load('psu_model', 'psu_model.py') EF = PM.EF # ── tunables (all overridable on the CLI) ───────────────────────────────────── GEN_BUDGET_W = 7200.0 # DECIDED 2026-08-06 (John): 8,000 W AC hard budget # x ~0.90 AC->DC efficiency = 7,200 W DC at the LEDs. # Port backstops are 80% (rose 70%) — run with # --cap 0.8; the compressor, not the port cap, is # the generator guarantee. See power/index.md. WEIGHTS = {'A': 8.0, 'B': 4.0, 'C': 2.0, 'D': 1.0} # Rule 3 reduction pressure (A most, D least) RELEASE_FRAMES = 12 # ~0.3 s at 40 fps — how fast a set recovers after a dip SETS = 'ABCD' def rule2_gains(psu_amps, headroom): """Per-set gain so no PSU in the set exceeds headroom×rating. psu_amps already ×cap.""" g = {s: 1.0 for s in SETS} for pid, a in psu_amps.items(): if a <= 0: continue s, _, rating, _ = PM.PSU_META[pid] g[s] = min(g[s], (rating * headroom) / a) return g def rule3_gains(set_watts, budget): """Extra per-set gain (≤1) spread A>B>C>D so total ≤ budget. No floors.""" total = sum(set_watts.values()) if total <= budget or total <= 0: return {s: 1.0 for s in SETS} excess = total - budget # find pressure k with Σ w[s]·min(1, k·weight[s]) = excess (monotonic in k) lo, hi = 0.0, 1e9 for _ in range(60): k = (lo + hi) / 2 removed = sum(set_watts[s] * min(1.0, k * WEIGHTS[s]) for s in SETS) if removed < excess: lo = k else: hi = k return {s: 1.0 - min(1.0, hi * WEIGHTS[s]) for s in SETS} def compress(frames, cap, headroom, gen_budget): psu_map = PM.build_psu_map() # per-set channel ranges (for applying gains) and per-PSU ranges (for measuring load) set_ranges = {s: [] for s in SETS} for pid, ranges in psu_map.items(): s = PM.PSU_META[pid][0] for st, n, _ in ranges: set_ranges[s].append((st, n)) maxch = max((st + n for r in psu_map.values() for st, n, _ in r), default=0) # The written header's channelCount MUST equal the actual per-frame size, not # just the PSU-mapped range. expand_frames() produces full 517-px/petal frames # (EXP_MAX_CHANNEL = 70401); the PSU map only reaches 70392 (3 px/petal short). # Using the PSU max here declared 70392 while frames were 70401 -> every player # mis-sliced frames, drifting 9 ch/frame into the rose (within-cell "chase"). maxch = max(maxch, max((len(fr) for fr in frames), default=0)) gprev = {s: 1.0 for s in SETS} rel = 1.0 / RELEASE_FRAMES stats = {'compressed_frames': 0, 'min_gain': {s: 1.0 for s in SETS}, 'gen_peak_w': 0.0, 'psu_peak_pct': {}} for fi, fr in enumerate(frames): if len(fr) < maxch: fr += bytearray(maxch - len(fr)) # measure per-PSU watts (design) and amps at the target cap psu_w = {} for pid, ranges in psu_map.items(): psu_w[pid] = sum(sum(fr[st:st+n]) * wpp / 765.0 for st, n, wpp in ranges) psu_a_cap = {pid: psu_w[pid] / PM.PSU_META[pid][1] * cap for pid in psu_w} for pid in psu_w: stats['psu_peak_pct'][pid] = max(stats['psu_peak_pct'].get(pid, 0.0), psu_a_cap[pid] / PM.PSU_META[pid][2]) # Rule 2 g2 = rule2_gains(psu_a_cap, headroom) # Rule 3 — on watts after Rule 2, at the target cap setw = {s: 0.0 for s in SETS} for pid, w in psu_w.items(): setw[PM.PSU_META[pid][0]] += w * cap * g2[PM.PSU_META[pid][0]] stats['gen_peak_w'] = max(stats['gen_peak_w'], sum(setw.values())) g3 = rule3_gains(setw, gen_budget) # combine + smooth (instant attack, slow release), then apply any_cut = False for s in SETS: gt = g2[s] * g3[s] g = gt if gt < gprev[s] else min(gt, gprev[s] + rel) gprev[s] = g stats['min_gain'][s] = min(stats['min_gain'][s], g) if g < 0.999: any_cut = True lut = bytes(int(v * g) for v in range(256)) for st, n in set_ranges[s]: fr[st:st+n] = fr[st:st+n].translate(lut) if any_cut: stats['compressed_frames'] += 1 return frames, maxch, stats def report(stats, nframes, cap, headroom, gen_budget, step): print(f"\n COMPRESSION SUMMARY (target cap {cap*100:.0f}%, headroom {headroom*100:.0f}% of rating)") cf = stats['compressed_frames'] print(f" frames touched: {cf}/{nframes} ({cf/max(nframes,1)*100:.1f}%)") print(f" per-set deepest cut: " + " ".join( f"{s}={( 1-stats['min_gain'][s])*100:4.1f}%" for s in SETS)) print(f" generator peak (design×cap): {stats['gen_peak_w']:.0f} W vs budget {gen_budget:.0f} W" f" ({stats['gen_peak_w']/gen_budget*100:.0f}%){' ⚠ over — Rule 3 engaged' if stats['gen_peak_w']>gen_budget else ''}") hot = sorted(stats['psu_peak_pct'].items(), key=lambda kv: -kv[1])[:5] print(f" hottest PSUs pre-compression (at cap): " + " ".join(f"{p}={v*100:.0f}%" for p, v in hot)) print(f" budget {gen_budget:.0f}W DC = the 8 kW AC decision (2026-08-06). " f"Confirm altitude derate with the electrician\n" f" (XP12000EH propane 9,025 W at sea level; BRC ~3,900 ft).\n") def main(): args = sys.argv[1:] if not args or args[0] in ('-h', '--help'): print(__doc__); return cap, headroom, gen, zstd, dry = 1.0, 1.0, GEN_BUDGET_W, False, False pos = [] i = 0 while i < len(args): a = args[i] if a == '--cap': i += 1; cap = float(args[i]) elif a == '--headroom': i += 1; headroom = float(args[i]) elif a == '--gen': i += 1; gen = float(args[i]) elif a == '--zstd': zstd = True elif a == '--dry-run': dry = True else: pos.append(a) i += 1 src = pos[0] out = pos[1] if len(pos) > 1 else os.path.splitext(src)[0] + '-compressed.fseq' print(f"Reading {src} ...") meta, frames = EF.read_fseq(src) print(f" {meta['num_frames']} frames @ {1000//meta['step_time']}fps") print("Expanding rose in-memory ...") frames, _ = EF.expand_frames(frames, EF.load_petal_segments()) print(f"Compressing (Rule 2 per-PSU + Rule 3 generator) ...") frames, maxch, stats = compress(frames, cap, headroom, gen) report(stats, meta['num_frames'], cap, headroom, gen, meta['step_time']) if dry: print(" --dry-run: no file written."); return if zstd: EF.write_fseq_zstd(out, frames, maxch, meta['step_time']) else: EF.write_fseq(out, frames, maxch, meta['step_time']) print(f"Wrote {out} ({os.path.getsize(out)/1e6:.1f} MB{', zstd' if zstd else ''})") print(" -> this is the EXPANDED, compressed file; play it on FPP (rose ports per expand-fseq --map).") if __name__ == '__main__': main()