From dd460f3973eff6d17ef513d3c677d516d12d8e78 Mon Sep 17 00:00:00 2001 From: David Allemang Date: Fri, 4 Sep 2026 17:02:15 -0400 Subject: [PATCH] tilesets as finite state machines --- finite-state-refine.py | 434 +++++++++++++++++++++++++++++++++++++ finite_state.py | 481 +++++++++++++---------------------------- pyproject.toml | 4 +- uv.lock | 27 +++ 4 files changed, 613 insertions(+), 333 deletions(-) create mode 100644 finite-state-refine.py diff --git a/finite-state-refine.py b/finite-state-refine.py new file mode 100644 index 00000000..d7b0a3db --- /dev/null +++ b/finite-state-refine.py @@ -0,0 +1,434 @@ +from enum import IntEnum +from dataclasses import dataclass +from typing import Tuple, List, Set, Dict, Union +from functools import cmp_to_key + +import numba +import tqdm + +# ========================================== +# 1. CORE TYPES & ENUMS +# ========================================== + +class Edge(IntEnum): + RISING = 0 + FALLING_PENDING = 1 + FALLING_UTILIZED = 2 + +class Context(IntEnum): + OTHER = 0 + WIRE = 1 + DIODE = 2 + +START_STATE = "START" + +@dataclass(frozen=True) +class Entry: + priority: int + delay: int + name: str = '-' + +@dataclass(frozen=True) +class Transition: + from_state: Union[Tuple[Edge, Context], str] + to_state: Union[Tuple[Edge, Context], str] + entries: Tuple[Entry, ...] + + def total_delay(self) -> int: + return sum(e.delay for e in self.entries) + +class TilesetFSM: + def __init__(self, transitions: List[Transition], start_state: str, accept_states: Set[Union[Tuple[Edge, Context], str]]): + self.transitions = transitions + self.start_state = start_state + self.accept_states = accept_states + + # Pre-group transitions by their source state for O(1) branching lookup + self.adj_list = {state: [] for state in set(t.from_state for t in transitions)} + for t in transitions: + self.adj_list[t.from_state].append(t) + + +# ========================================== +# 2. EVENT-DRIVEN SORTING (Numba-Ready) +# ========================================== + +def get_timing_events(chain: Tuple[Entry, ...]) -> Tuple[Tuple[int, int], ...]: + """Converts a chain into a dense tuple of (tick, priority) events.""" + events = [] + tick = 0 + for e in chain: + if e.delay > 0: + events.append((tick, e.priority)) + tick += e.delay + return tuple(events) + +def cmp_chains(chain_a: Tuple[Entry, ...], chain_b: Tuple[Entry, ...]) -> int: + """ + O(K) lexicographical comparison replicating the behavior of `-inf` padding. + Strictly uses integers, making it a prime candidate for Numba acceleration. + """ + events_a = get_timing_events(chain_a) + events_b = get_timing_events(chain_b) + + idx_a, idx_b = 0, 0 + while idx_a < len(events_a) and idx_b < len(events_b): + tick_a, pri_a = events_a[idx_a] + tick_b, pri_b = events_b[idx_b] + + if tick_a == tick_b: + if pri_a < pri_b: return -1 + if pri_a > pri_b: return 1 + idx_a += 1 + idx_b += 1 + elif tick_a < tick_b: + # A has an event while B implies -inf. -inf is smaller, so B is smaller. + return 1 + else: + return -1 + + if idx_a < len(events_a): return 1 + if idx_b < len(events_b): return -1 + + # Tie-breaker on component count (fewer components is preferred/smaller) + if len(chain_a) < len(chain_b): return -1 + if len(chain_a) > len(chain_b): return 1 + return 0 + + +# ========================================== +# 3. GENERATE THE CARTESIAN STATE MACHINE +# ========================================== + +transitions = [] + +for edge in Edge: + for ctx in Context: + state = (edge, ctx) + + # A. Wire + if ctx == Context.DIODE: + transitions.append(Transition(state, (edge, Context.WIRE), (Entry(0, 0, '---'),))) + + # B. Torch + if ctx == Context.DIODE: + if edge == Edge.RISING: + transitions.append(Transition(state, (Edge.FALLING_PENDING, Context.OTHER), (Entry(0, 2, 'torch '),))) + elif edge == Edge.FALLING_UTILIZED: + transitions.append(Transition(state, (Edge.RISING, Context.OTHER), (Entry(0, 2, 'torch '),))) + + # C. Comparator + cmp_pri = -1 if ctx == Context.DIODE else 0 + transitions.append(Transition(state, (edge, Context.DIODE), (Entry(cmp_pri, 2, 'cmp'),))) + + # D. Repeaters + for d in (2, 4, 6, 8): + if ctx == Context.DIODE: + transitions.append(Transition(state, (edge, Context.DIODE), (Entry(-3, d, f're{d}'),))) + else: + if edge == Edge.RISING: + if ctx != Context.WIRE: + transitions.append(Transition(state, (edge, Context.DIODE), (Entry(-1, d, f're{d}'),))) + else: + transitions.append(Transition(state, (Edge.FALLING_UTILIZED, Context.DIODE), (Entry(-2, d, f're{d}'),))) + + # E. Fluids + if edge == Edge.RISING: + for d in (5, 10, 30): + fluid_macro = (Entry(0, 2, 'obs'), Entry(1, d, f'fluid{d}'), Entry(0, 4, 'disp')) + transitions.append(Transition(state, (Edge.RISING, Context.OTHER), fluid_macro)) + +# START state boundary +start_transitions = [] +for t in transitions: + if t.from_state == (Edge.RISING, Context.OTHER): + if not (len(t.entries) == 1 and t.entries[0].name == '---'): + start_transitions.append(Transition(START_STATE, t.to_state, t.entries)) + +start_transitions.append(Transition(START_STATE, (Edge.FALLING_PENDING, Context.OTHER), (Entry(0, 2, 'torch '),))) +transitions.extend(start_transitions) + +all_states = set(t.from_state for t in transitions) | set(t.to_state for t in transitions) + +# Accept states are any normal state tuple (skip START_STATE string) +accept_states = {s for s in all_states if isinstance(s, tuple) and s[0] == Edge.RISING} + +model = TilesetFSM( + transitions=transitions, + start_state=START_STATE, + accept_states=accept_states +) + + +# ========================================== +# 4. FAST DP GENERATOR +# ========================================== + +def generate_all_fast(model: TilesetFSM, target_delay: int) -> List[List[Entry]]: + all_states = set(model.adj_list.keys()) | model.accept_states + can_reach = {d: {s: False for s in all_states} for d in range(target_delay + 1)} + + for s in model.accept_states: + can_reach[0][s] = True + + for d in range(1, target_delay + 1): + for state in all_states: + for t in model.adj_list.get(state, []): + t_delay = t.total_delay() + if t_delay <= d and can_reach[d - t_delay][t.to_state]: + can_reach[d][state] = True + + unique_chains = {} + stack = [(model.start_state, target_delay, ())] + + with tqdm.tqdm(desc="Traversing DFS") as bar: + while stack: + curr_state, rem_delay, chain = stack.pop() + bar.update(1) + + if rem_delay == 0 and curr_state in model.accept_states: + # Use the dense events tuple as an instant hashable uniqueness key + events_key = get_timing_events(chain) + if events_key not in unique_chains: + unique_chains[events_key] = chain + else: + if len(chain) < len(unique_chains[events_key]): + unique_chains[events_key] = chain + continue + + for t in model.adj_list.get(curr_state, []): + t_delay = t.total_delay() + if t_delay <= rem_delay and can_reach[rem_delay - t_delay][t.to_state]: + stack.append((t.to_state, rem_delay - t_delay, chain + t.entries)) + + # Sort strictly using the dense event comparator + sorted_chains = sorted(unique_chains.values(), key=cmp_to_key(cmp_chains)) + return [list(c) for c in sorted_chains] + + +def cmp_prefix(events_prefix: Tuple[Tuple[int, int], ...], events_target: Tuple[Tuple[int, int], ...]) -> int: + """ + Evaluates prefix divergence. Returns 0 if they match exactly so far. + If it returns non-zero, the divergence is permanent for ANY valid completion. + """ + idx_p, idx_t = 0, 0 + while idx_p < len(events_prefix) and idx_t < len(events_target): + tick_p, pri_p = events_prefix[idx_p] + tick_t, pri_t = events_target[idx_t] + + if tick_p == tick_t: + if pri_p < pri_t: return -1 + if pri_p > pri_t: return 1 + idx_p += 1 + idx_t += 1 + elif tick_p < tick_t: + return 1 + else: + return -1 + return 0 + + +def neighborhood(model: TilesetFSM, target_chain: List[Entry], before: int, after: int) -> Tuple[List[List[Entry]], List[List[Entry]]]: + """ + Finds the exact immediate neighborhood around a target chain without enumerating the language. + Uses DP-guided Branch and Bound to prune the astronomical search space. + """ + target_delay = sum(e.delay for e in target_chain) + events_target = get_timing_events(tuple(target_chain)) + + # 1. Build the Reachability Table (O(States * Target Delay) - Extremely Fast) + all_states = set(model.adj_list.keys()) | model.accept_states + can_reach = {d: {s: False for s in all_states} for d in range(target_delay + 1)} + for s in model.accept_states: can_reach[0][s] = True + for d in range(1, target_delay + 1): + for state in all_states: + for t in model.adj_list.get(state, []): + t_delay = t.total_delay() + if t_delay <= d and can_reach[d - t_delay][t.to_state]: + can_reach[d][state] = True + + # 2. Sort transitions to aggressively target the bounds + def cmp_transitions(t1: Transition, t2: Transition) -> int: + return cmp_chains(t1.entries, t2.entries) + + # Ascending sort: Explores lexicographically smaller transitions first + adj_list_asc = {k: sorted(v, key=cmp_to_key(cmp_transitions)) for k, v in model.adj_list.items()} + # Descending sort: Explores lexicographically larger transitions first + adj_list_desc = {k: sorted(v, key=cmp_to_key(cmp_transitions), reverse=True) for k, v in model.adj_list.items()} + + # --- SEARCH AFTER --- + found_after = {} + ub_events, ub_chain = None, None + stack = [(model.start_state, target_delay, (), ())] + + while stack and after > 0: + curr_state, rem_delay, chain, events_so_far = stack.pop() + + if rem_delay == 0 and curr_state in model.accept_states: + if cmp_chains(chain, tuple(target_chain)) > 0: + ev_key = get_timing_events(chain) + # Deduplicate and tie-break on length + if ev_key not in found_after or len(chain) < len(found_after[ev_key]): + found_after[ev_key] = chain + + if len(found_after) > after: + sorted_items = sorted(found_after.values(), key=cmp_to_key(cmp_chains)) + found_after = {get_timing_events(c): c for c in sorted_items[:after]} + ub_chain = sorted_items[after - 1] + ub_events = get_timing_events(ub_chain) + continue + + # Prune branches that are <= target or > our worst accepted bound + div_t = cmp_prefix(events_so_far, events_target) + if div_t < 0: continue + if ub_events and cmp_prefix(events_so_far, ub_events) > 0: continue + + # Push in reverse so the smallest transitions are popped/explored first + for t in reversed(adj_list_asc.get(curr_state, [])): + t_delay = t.total_delay() + if t_delay <= rem_delay and can_reach[rem_delay - t_delay][t.to_state]: + new_chain = chain + t.entries + stack.append((t.to_state, rem_delay - t_delay, new_chain, get_timing_events(new_chain))) + + # --- SEARCH BEFORE --- + found_before = {} + lb_events, lb_chain = None, None + stack = [(model.start_state, target_delay, (), ())] + + while stack and before > 0: + curr_state, rem_delay, chain, events_so_far = stack.pop() + + if rem_delay == 0 and curr_state in model.accept_states: + if cmp_chains(chain, tuple(target_chain)) < 0: + ev_key = get_timing_events(chain) + if ev_key not in found_before or len(chain) < len(found_before[ev_key]): + found_before[ev_key] = chain + + if len(found_before) > before: + sorted_items = sorted(found_before.values(), key=cmp_to_key(cmp_chains)) + found_before = {get_timing_events(c): c for c in sorted_items[-before:]} + lb_chain = sorted_items[-before] + lb_events = get_timing_events(lb_chain) + continue + + # Prune branches that are >= target or < our worst accepted bound + div_t = cmp_prefix(events_so_far, events_target) + if div_t > 0: continue + if lb_events and cmp_prefix(events_so_far, lb_events) < 0: continue + + # Push in reverse so the largest transitions are popped/explored first + for t in reversed(adj_list_desc.get(curr_state, [])): + t_delay = t.total_delay() + if t_delay <= rem_delay and can_reach[rem_delay - t_delay][t.to_state]: + new_chain = chain + t.entries + stack.append((t.to_state, rem_delay - t_delay, new_chain, get_timing_events(new_chain))) + + sorted_before = sorted(found_before.values(), key=cmp_to_key(cmp_chains)) + sorted_after = sorted(found_after.values(), key=cmp_to_key(cmp_chains)) + + return [list(c) for c in sorted_before], [list(c) for c in sorted_after] + +def format_csv_timing(chain: List[Entry], target_delay: int) -> str: + """Helper to reconstruct your original CSV string format without allocating infs.""" + events = dict(get_timing_events(tuple(chain))) + out = [] + for tick in range(target_delay): + if tick in events: + out.append(f"{events[tick]:+}") + else: + out.append(" ") + return " ".join(out) + + +# if __name__ == "__main__": +# target_delay = 20 +# +# with open('chains.csv', 'w') as f: +# for chain in generate_all_fast(model, target_delay): +# for el in chain: +# f.write(el.name) +# f.write(' ') +# f.write(',') +# f.write(format_csv_timing(chain, target_delay)) +# f.write('\n') + + +def parse_names(model: TilesetFSM, names: List[str]) -> List[Entry]: + """ + Parses a sequence of string names into a validated List[Entry] with inferred priorities. + """ + current_state = model.start_state + i = 0 + inferred_chain = [] + + while i < len(names): + match_found = False + + # Look at all valid transitions from our current state + for t in model.adj_list.get(current_state, []): + # Clean up FSM internal names (e.g., 'torch ' -> 'torch') + t_names = [e.name.strip() for e in t.entries] + + # Normalize user input for comparison + chunk = names[i : i + len(t_names)] + chunk_norm = [] + for n in chunk: + n = n.strip() + if n == 'wire': n = '---' + if n.startswith('rep'): n = n.replace('rep', 're') + chunk_norm.append(n) + + # Check if this transition matches the input + if chunk_norm == t_names: + inferred_chain.extend(t.entries) + current_state = t.to_state + i += len(t_names) + match_found = True + break + + if not match_found: + raise ValueError( + f"Syntax Error: Cannot place '{names[i]}' while in state {current_state} " + f"at index {i}." + ) + + if current_state not in model.accept_states: + raise ValueError( + f"Unexpected EOF: Sequence ended in non-accepting state {current_state}. " + "Did you forget to un-invert a torch?" + ) + + return inferred_chain + +# ========================================== +# USAGE EXAMPLE +# ========================================== + +if __name__ == "__main__": + # Your target sequence (using friendly names!) + input_str = 'cmp rep2 torch rep2 torch' + names_list = input_str.split() + + print(f"Parsing: {names_list}...") + target_chain = parse_names(model, names_list) + + print(f"Finding neighborhood...") + before_chains, after_chains = neighborhood(model, target_chain, before=5, after=5) + + def print_chain(chain: List[Entry], prefix: str = ""): + # Reconstruct the string without trailing spaces from the names + clean_names = " ".join(e.name.strip() for e in chain) + # Calculate the linearized delay score for visual context + events = get_timing_events(tuple(chain)) + print(f"{prefix}{clean_names:<35} | Events: {events}") + + print("\n--- BEFORE ---") + for c in before_chains: + print_chain(c) + + print("\n--- TARGET ---") + print_chain(target_chain, prefix=">> ") + + print("\n--- AFTER ---") + for c in after_chains: + print_chain(c) \ No newline at end of file diff --git a/finite_state.py b/finite_state.py index 6158c732..0e1727d6 100644 --- a/finite_state.py +++ b/finite_state.py @@ -1,19 +1,20 @@ -import math -from contextlib import redirect_stdout from dataclasses import dataclass -from itertools import islice +from math import isfinite, inf from typing import Tuple, Optional, List, Set +import tqdm + + @dataclass(frozen=True) class Entry: priority: int delay: int name: str = '-' - def linear_parts(self) -> Tuple[int, ...]: + def linear_parts(self) -> Tuple[int | float, ...]: if self.delay == 0: return () - return (self.priority,) + (-9999,) * (self.delay - 1) + return (self.priority,) + (-inf,) * (self.delay - 1) @dataclass(frozen=True) class Transition: @@ -118,61 +119,83 @@ class TilesetFSM: if current_state not in self.accept_states: raise ValueError( f"Unexpected EOF: Sequence ended in non-accepting state '{current_state}'. " - "Did you forget to un-invert a torch?" + "Did you forget to un-invert a torch ?" ) return inferred_chain -# ========================================== -# 1. GENERATE THE CARTESIAN STATE MACHINE -# ========================================== - transitions = [] -for signal in ["NORMAL", "INVERTED"]: - for faces in ["OTHER", "DIODE"]: - current_state = f"{signal}_{faces}" +# We expand our signals to track if the unique -2 priority was utilized +signals = ["NORMAL", "INV_PENDING", "INV_UTILIZED"] +contexts = ["DIODE", "WIRE", "OTHER"] - # A. Wire (Zero-delay state reset. Breaks diode chains) - # FIX: Only allow wire if we are actually facing a diode. - if faces == "DIODE": +for signal in signals: + for context in contexts: + current_state = f"{signal}_{context}" + + # A. Wire + if context == "DIODE": transitions.append(Transition( from_state=current_state, - to_state=f"{signal}_OTHER", - entries=(Entry(0, 0, 'wire'),) + to_state=f"{signal}_WIRE", + entries=(Entry(0, 0, '---'),) )) - # B. Comparator - cmp_pri = -1 if faces == "DIODE" else 0 + # B. Torch + if context == "DIODE": + if signal == "NORMAL": + # Enter inverted mode as PENDING (haven't used -2 yet) + transitions.append(Transition( + from_state=current_state, + to_state="INV_PENDING_OTHER", + entries=(Entry(0, 2, 'torch '),) + )) + elif signal == "INV_UTILIZED": + # Un-invert is ONLY allowed if we successfully utilized the -2 priority + transitions.append(Transition( + from_state=current_state, + to_state="NORMAL_OTHER", + entries=(Entry(0, 2, 'torch '),) + )) + # Notice there is no torch transition for INV_PENDING! + + # C. Comparator + cmp_pri = -1 if context == "DIODE" else 0 transitions.append(Transition( from_state=current_state, to_state=f"{signal}_DIODE", entries=(Entry(cmp_pri, 2, 'cmp'),) )) - # C. Repeaters + # D. Repeaters for d in (2, 4, 6, 8): - if faces == "DIODE": + if context == "DIODE": + # Facing a diode yields -3 and leaves our utilization state unchanged rep_pri = -3 + next_signal = signal + allow_rep = True else: - rep_pri = -1 if signal == "NORMAL" else -2 + if signal == "NORMAL": + rep_pri = -1 + next_signal = signal + allow_rep = (context != "WIRE") # Block redundant rep after + else: + # We are in an inverted mode and facing OTHER/WIRE. + # This yields the special -2 priority! + rep_pri = -2 + next_signal = "INV_UTILIZED" # Mark the -2 as successfully utilized + allow_rep = True - transitions.append(Transition( - from_state=current_state, - to_state=f"{signal}_DIODE", - entries=(Entry(rep_pri, d, f'rep{d}'),) - )) + if allow_rep: + transitions.append(Transition( + from_state=current_state, + to_state=f"{next_signal}_DIODE", + entries=(Entry(rep_pri, d, f're{d}'),) + )) - # D. Torch (Inverts signal, is not a diode) - next_sig = "INVERTED" if signal == "NORMAL" else "NORMAL" - transitions.append(Transition( - from_state=current_state, - to_state=f"{next_sig}_OTHER", - entries=(Entry(0, 2, 'torch'),) - )) - - # E. Fluids (Only allowed on rising edges/NORMAL) + # E. Fluids if signal == "NORMAL": for d in (5, 10, 30): fluid_macro = (Entry(0, 2, 'obs'), Entry(1, d, f'fluid{d}'), Entry(0, 4, 'disp')) @@ -182,15 +205,36 @@ for signal in ["NORMAL", "INVERTED"]: entries=fluid_macro )) -# Initialize assuming the end of the chain faces nothing (OTHER). -# We require the final accepting state to be NORMAL (un-inverted) to be valid. +# ========================================== +# 2. CREATE THE BOUNDARY 'START' STATE +# ========================================== + +start_transitions = [] +for t in transitions: + if t.from_state == "NORMAL_OTHER": + # Allow everything EXCEPT the zero-delay at the very start + if not (len(t.entries) == 1 and t.entries[0].name == '---'): + start_transitions.append(Transition( + from_state="START", + to_state=t.to_state, + entries=t.entries + )) +start_transitions.append(Transition( + from_state="START", + to_state="INV_PENDING_OTHER", + entries=(Entry(0, 2, 'torch '),) +)) +transitions.extend(start_transitions) + +all_states = set(t.from_state for t in transitions) | set(t.to_state for t in transitions) +accept_states = {s for s in all_states if s.startswith("NORMAL")} + model = TilesetFSM( transitions=transitions, - start_state='NORMAL_OTHER', - accept_states={'NORMAL_OTHER', 'NORMAL_DIODE'} + start_state='START', + accept_states=accept_states ) - # ========================================== # 2. USAGE EXAMPLES # ========================================== @@ -202,308 +246,81 @@ def display_chain(entries): def linearize_entries(entries): return tuple(p for e in entries for p in e.linear_parts()) -# if __name__ == "__main__": -# # Example 1: The Zero-Delay Wire breaking the context -# chain_names_1 = ["cmp", "wire", "cmp"] -# entries_1 = model.parse_names(chain_names_1) -# -# print(f"Input: {chain_names_1}") -# print(f"Inferred: {display_chain(entries_1)}") -# # Output: cmp(0) -> wire(0) -> cmp(0) -# print() -# -# # Example 2: Diode-facing-diode context -# chain_names_2 = ["cmp", "cmp"] -# entries_2 = model.parse_names(chain_names_2) -# -# print(f"Input: {chain_names_2}") -# print(f"Inferred: {display_chain(entries_2)}") -# # Output: cmp(0) -> cmp(-1) -# print() -# -# # Example 3: Falling-edge inversion behavior -# chain_names_3 = ["torch", "rep4", "wire", "rep4", "torch"] -# entries_3 = model.parse_names(chain_names_3) -# -# print(f"Input: {chain_names_3}") -# print(f"Inferred: {display_chain(entries_3)}") -# # Output: torch(0) -> rep4(-3) -> wire(0) -> rep4(-2) -> torch(0) -# # (Notice the first rep4 faces the torch (a diode), getting -3, -# # while the second faces the wire (OTHER) during an inverted state, getting -2) -# print() -# -# # Example 4: Compiling a sequence to use as a search target -# target_names = ["cmp", "cmp", "cmp", "cmp"] -# target_entries = model.parse_names(target_names) -# target_lin = linearize_entries(target_entries) -# -# print(f"Finding NEXT chain for target: {target_names}") -# print(f"Inferred: {display_chain(target_entries)}") -# next_chain = model.find_next(target_lin, target_delay=8) -# if next_chain: -# print(f"Next: {display_chain(next_chain)}") -# else: -# print(f'Next: None') -# # Output: Next: rep2(-1) -> rep2(-3) -> rep2(-3) -> rep2(-3) - -def enumerate_chains(model: TilesetFSM, target_delay: int): +def generate_all_fast(model: TilesetFSM, target_delay: int) -> List[List[Entry]]: """ - Yields every valid tile sequence of the target delay in lexicographical order. + Uses Dynamic Programming to generate all valid chains of a given delay in O(N log N) time, + completely eliminating dead-end traversal. """ - # Bootstrap with an infinitely low sequence to find the very first chain - current_lin = (-999,) * target_delay + # 1. DP Table Setup + # can_reach[d][state] = True if we can reach an accept state from 'state' in exactly 'd' ticks + all_states = set(model.adj_list.keys()) | model.accept_states + can_reach = {d: {s: False for s in all_states} for d in range(target_delay + 1)} - while True: - # Find the next valid sequence - next_chain = model.find_next(current_lin, target_delay) + # Base cases: Delay 0 is only valid if we are in an accept state + for s in model.accept_states: + can_reach[0][s] = True - # If no next sequence exists, we've enumerated the entire language - if next_chain is None: - break + # 2. Build the DP table bottom-up + for d in range(1, target_delay + 1): + for state in all_states: + for t in model.adj_list.get(state, []): + t_delay = t.total_delay() + # If this transition fits within our remaining delay... + if t_delay <= d: + # ...and the state it leads to can successfully finish the chain + if can_reach[d - t_delay][t.to_state]: + can_reach[d][state] = True + # 3. Guided DFS + raw_chains = [] - yield next_chain + def build_chain(current_state: str, remaining_delay: int, current_chain: Tuple[Entry, ...]): + bar.update(1) + if remaining_delay == 0 and current_state in model.accept_states: + raw_chains.append(list(current_chain)) + return - # Update our pointer for the next iteration - current_lin = linearize_entries(next_chain) + for t in model.adj_list.get(current_state, []): + t_delay = t.total_delay() + if t_delay <= remaining_delay: + if can_reach[remaining_delay - t_delay][t.to_state]: + build_chain(t.to_state, remaining_delay - t_delay, current_chain + t.entries) -# ========================================== -# USAGE EXAMPLE -# ========================================== + with tqdm.tqdm() as bar: + build_chain(model.start_state, target_delay, ()) -# if __name__ == "__main__": -# target_delay = 6 -# print(f"Enumerating all valid chains of Delay {target_delay}...\n") -# -# chain_count = 0 -# for chain in enumerate_chains(model, target_delay): -# chain_count += 1 -# -# # We can extract the names to see the raw component sequence -# names = [e.name for e in chain] -# print(f"{chain_count: 3d}. {names}") -# -# print(f"Total sequences found for delay {target_delay}: {chain_count}") + # 4. Mathematically Guarantee Distinct Sequences + with tqdm.tqdm() as bar: + unique_chains = {} + for chain in raw_chains: + # Calculate the linearized tuple to use as our uniqueness key + lin_key = tuple(p for e in chain for p in e.linear_parts()) -import heapq -from typing import List + if lin_key not in unique_chains: + unique_chains[lin_key] = chain + bar.update(1) + else: + # TIE-BREAKER: If two chains produce the exact same physical timing, + # keep the one with the fewest components (eliminates redundant wires/torches) + if len(chain) < len(unique_chains[lin_key]): + unique_chains[lin_key] = chain -def generate_render_document(chain_entries: List['Entry']) -> str: - """ - Generates a setblock-style animation document for a given redstone chain. - """ - # Reverse chain so x=0 is the first component to activate (upstream) - components = chain_entries[::-1] - N = len(components) + # 5. Sort the purely unique chains + sorted_chains = sorted(unique_chains.values(), + key=lambda chain: tuple(p for e in chain for p in e.linear_parts())) - # 1. Forward sweep to calculate resting states (t=0) - resting_power = False - resting_states = [] - for c in components: - if c.name == 'torch': - # Torches invert the resting power for downstream components - resting_power = not resting_power - resting_states.append(resting_power) - elif c.name == 'obs': - # Observers block resting power - resting_power = False - resting_states.append(False) - else: - # rep, cmp, and wire pass resting power - resting_states.append(resting_power) - - lines = [] - - actual_visual_state = {} - posmap = {} - x = 0 - for i, (c, resting) in enumerate(zip(components, resting_states)): - is_powered = "true" if resting else "false" - - if c.name == 'wire': - lines.append(f"p {x} 0 0 white_concrete") - lines.append(f"p {x} -1 0 smooth_stone_slab type=top") - elif c.name == 'torch': - lines.append(f"p {x} 0 0 white_concrete") - lines.append(f"p {x} -1 0 smooth_stone_slab type=top") - x += 1 - lines.append(f"p {x} 0 0 redstone_wall_torch facing=east lit={is_powered}") - lines.append(f"p {x} -1 0 smooth_stone_slab type=top") - actual_visual_state[(i, 'lit')] = is_powered - posmap[i] = x - elif c.name.startswith('rep'): - delay_ticks = int(c.name[3:]) // 2 - lines.append(f"p {x} 0 0 repeater facing=west powered={is_powered} locked=false delay={delay_ticks}") - lines.append(f"p {x} -1 0 smooth_stone_slab type=top") - actual_visual_state[(i, 'powered')] = is_powered - posmap[i] = x - elif c.name == 'cmp': - lines.append(f"p {x} 0 0 comparator facing=west powered={is_powered} mode=compare") - lines.append(f"p {x} -1 0 smooth_stone_slab type=top") - actual_visual_state[(i, 'powered')] = is_powered - posmap[i] = x - elif c.name == 'obs': - lines.append(f"p {x} 0 0 observer facing=west powered=false") - lines.append(f"p {x} -1 0 smooth_stone_slab type=top") - actual_visual_state[(i, 'powered')] = "false" - posmap[i] = x - - x += 1 - - # 4. Simulate a 2gt pulse passing through the system - # pq holds (time, component_index, input_level) - pq = [] - heapq.heappush(pq, (0, 0, True)) - heapq.heappush(pq, (2, 0, False)) - - visual_changes = {} # time -> dict of (index, prop) -> value - - while pq: - t, i, level = heapq.heappop(pq) - - if i >= N: - continue - - c = components[i] - - if c.name == 'wire': - # Instantly passes the signal to the next component - heapq.heappush(pq, (t, i+1, level)) - - elif c.name == 'torch': - out_level = not level - out_t = t + 2 - heapq.heappush(pq, (out_t, i+1, out_level)) - - val_str = "true" if out_level else "false" - if out_t not in visual_changes: visual_changes[out_t] = {} - visual_changes[out_t][(i, 'lit')] = val_str - - elif c.name.startswith('rep'): - delay = int(c.name[3:]) - out_t = t + delay - heapq.heappush(pq, (out_t, i+1, level)) - - val_str = "true" if level else "false" - if out_t not in visual_changes: visual_changes[out_t] = {} - visual_changes[out_t][(i, 'powered')] = val_str - - elif c.name == 'cmp': - out_t = t + 2 - heapq.heappush(pq, (out_t, i+1, level)) - - val_str = "true" if level else "false" - if out_t not in visual_changes: visual_changes[out_t] = {} - visual_changes[out_t][(i, 'powered')] = val_str - - elif c.name == 'obs': - # An observer fires a 2gt pulse whenever its input changes - out_t_on = t + 2 - out_t_off = t + 4 - - heapq.heappush(pq, (out_t_on, i+1, True)) - heapq.heappush(pq, (out_t_off, i+1, False)) - - if out_t_on not in visual_changes: visual_changes[out_t_on] = {} - if out_t_off not in visual_changes: visual_changes[out_t_off] = {} - - # Python's heapq pops `False` before `True` if times match. - # This perfectly replicates observer pulse-extension: if an OFF and ON - # hit on the exact same tick, the ON overwrites the OFF. - visual_changes[out_t_on][(i, 'powered')] = "true" - visual_changes[out_t_off][(i, 'powered')] = "false" - - # 5. Format the dynamic animation frames - sorted_times = sorted(visual_changes.keys()) - - for t in sorted_times: - frame_lines = [] - changes_at_t = visual_changes[t] - - for (i, prop), val in changes_at_t.items(): - x = posmap[i] - # Only emit if it actually changes the visual state - if actual_visual_state.get((i, prop)) != val: - actual_visual_state[(i, prop)] = val - frame_lines.append(f"p {x} 0 0 {prop}={val}") - - if frame_lines: - lines.append(f"t {t}") - lines.extend(frame_lines) - - # 6. Add final padding frame - last_t = sorted_times[-1] if sorted_times else 0 - lines.append(f"t {last_t + 4}") - lines.append("p 0 0 0") - - return "\n".join(lines) - - - -def makedoc(chain: list[Entry], z=0): - x = 0 - - for el in chain: - match el.name: - case 'cmp': - yield f'p {x} 0 {z} comparator facing=west powered=false mode=compare' - yield f'p {x} -1 {z} smooth_stone_slab type=top' - case 'rep2': - yield f'p {x} 0 {z} repeater facing=west powered=false locked=false delay=1' - yield f'p {x} -1 {z} smooth_stone_slab type=top' - case 'rep4': - yield f'p {x} 0 {z} repeater facing=west powered=false locked=false delay=2' - yield f'p {x} -1 {z} smooth_stone_slab type=top' - case 'rep6': - yield f'p {x} 0 {z} repeater facing=west powered=false locked=false delay=3' - yield f'p {x} -1 {z} smooth_stone_slab type=top' - case 'rep8': - yield f'p {x} 0 {z} repeater facing=west powered=false locked=false delay=4' - yield f'p {x} -1 {z} smooth_stone_slab type=top' - case 'wire': - yield f'p {x} 0 {z} white_concrete' - yield f'p {x} -1 {z} smooth_stone_slab type=top' - case 'torch': - yield f'p {x} 0 {z} redstone_wall_torch facing=east lit=false' - yield f'p {x} -1 {z} smooth_stone_slab type=top' - x -= 1 - yield f'p {x} 0 {z} white_concrete' - yield f'p {x} -1 {z} smooth_stone_slab type=top' - x -= 1 - - - # return ' '.join(el.name for el in chain) - - # for el in chain: - # match el.name: - # print(el.name) + return sorted_chains if __name__ == "__main__": - target_delay = 6 + target_delay = 20 - with open('/home/local/KHQ/david.allemang/src/wireless/content/tilesets.typ', 'w') as f: - with redirect_stdout(f): - print(''' - #import "/lib.typ": diorama, example, note, todo - - == Tilesets - - #show raw.where(lang: "mc-diorama"): it => diorama( - zoom: true, - theta: 180, - phi: 60, - radius: 30, - height: "120em", - it.text, - ) - ''') - - z = 0 - - print('```mc-diorama') - for chain in enumerate_chains(model, target_delay): - for line in makedoc(chain, z): - print(line) - - z -= 1.25 - print('```') + with open('chains.csv', 'w') as f: + for chain in generate_all_fast(model, target_delay): + for el in chain: + f.write(el.name) + f.write(' ') + f.write(',') + for el in linearize_entries(chain): + f.write(f'{el:+}' if isfinite(el) else ' ') + f.write(' ') + f.write('\n') diff --git a/pyproject.toml b/pyproject.toml index 6f7db073..cdbc45ae 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,4 +2,6 @@ name = "wireless-formalism" version = "0.1.0" requires-python = ">=3.14" -dependencies = [] +dependencies = [ + "tqdm>=4.70.0", +] diff --git a/uv.lock b/uv.lock index a9fb26f6..145c88cf 100644 --- a/uv.lock +++ b/uv.lock @@ -2,7 +2,34 @@ version = 1 revision = 3 requires-python = ">=3.14" +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, +] + +[[package]] +name = "tqdm" +version = "4.70.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/21/3b/6c24bec5be5e743ffd99576daa5cc077722fc7d5bbc00bd133fa0c698dc6/tqdm-4.70.0.tar.gz", hash = "sha256:55b0b0dbd97462d06ebee91e4dac24ed4d4702be82b24f07e6c1d27e08cea220", size = 795438, upload-time = "2026-07-27T11:33:15.271Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f9/1c/01bfd571a64e7f270e6bab5e33777debe0edc56759233ce84f27dec92d14/tqdm-4.70.0-py3-none-any.whl", hash = "sha256:7f585706bfddbdebf89daac705b2dfcc16890130727d3197ca62c732b4310953", size = 80184, upload-time = "2026-07-27T11:33:13.167Z" }, +] + [[package]] name = "wireless-formalism" version = "0.1.0" source = { virtual = "." } +dependencies = [ + { name = "tqdm" }, +] + +[package.metadata] +requires-dist = [{ name = "tqdm", specifier = ">=4.70.0" }]