me/wireless-docs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tilesets as finite state machines

Browse Source
This commit is contained in:
David Allemang
2026-09-04 17:02:15 -04:00
parent 3149659af9
commit dd460f3973
4 changed files with 613 additions and 333 deletions
Download Patch File Download Diff File Expand all files Collapse all files

434
finite-state-refine.py Normal file
View File

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

481
finite_state.py
View File

@@ -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 <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')

4
pyproject.toml
View File

@@ -2,4 +2,6 @@
name = "wireless-formalism"
version = "0.1.0"
requires-python = ">=3.14"
dependencies = []
dependencies = [
"tqdm>=4.70.0",
]

27
uv.lock generated
View File

@@ -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" }]