simplified chains.py
This commit is contained in:
3
.idea/.gitignore
generated
vendored
3
.idea/.gitignore
generated
vendored
@@ -1,3 +0,0 @@
|
||||
# Default ignored files
|
||||
/shelf/
|
||||
/workspace.xml
|
||||
52
.idea/inspectionProfiles/Project_Default.xml
generated
52
.idea/inspectionProfiles/Project_Default.xml
generated
@@ -1,52 +0,0 @@
|
||||
<component name="InspectionProjectProfileManager">
|
||||
<profile version="1.0">
|
||||
<option name="myName" value="Project Default" />
|
||||
<inspection_tool class="HtmlUnknownTag" enabled="true" level="WARNING" enabled_by_default="true">
|
||||
<option name="myValues">
|
||||
<value>
|
||||
<list size="12">
|
||||
<item index="0" class="java.lang.String" itemvalue="nobr" />
|
||||
<item index="1" class="java.lang.String" itemvalue="noembed" />
|
||||
<item index="2" class="java.lang.String" itemvalue="comment" />
|
||||
<item index="3" class="java.lang.String" itemvalue="noscript" />
|
||||
<item index="4" class="java.lang.String" itemvalue="embed" />
|
||||
<item index="5" class="java.lang.String" itemvalue="script" />
|
||||
<item index="6" class="java.lang.String" itemvalue="my-paragraph" />
|
||||
<item index="7" class="java.lang.String" itemvalue="mc-diorama" />
|
||||
<item index="8" class="java.lang.String" itemvalue="mc-world" />
|
||||
<item index="9" class="java.lang.String" itemvalue="mc-frame" />
|
||||
<item index="10" class="java.lang.String" itemvalue="mc-timeline" />
|
||||
<item index="11" class="java.lang.String" itemvalue="mc-camera" />
|
||||
</list>
|
||||
</value>
|
||||
</option>
|
||||
<option name="myCustomValuesEnabled" value="true" />
|
||||
</inspection_tool>
|
||||
<inspection_tool class="PyPackageRequirementsInspection" enabled="true" level="WARNING" enabled_by_default="true">
|
||||
<option name="ignoredPackages">
|
||||
<list>
|
||||
<option value="numpy" />
|
||||
<option value="opencv-contrib-python-headless" />
|
||||
<option value="joblib" />
|
||||
<option value="loguru" />
|
||||
<option value="matplotlib" />
|
||||
<option value="pymatreader" />
|
||||
<option value="tqdm" />
|
||||
<option value="ipykernel" />
|
||||
<option value="itk-montage" />
|
||||
<option value="simpleitk" />
|
||||
<option value="multiview-stitcher" />
|
||||
<option value="itk-elastix" />
|
||||
<option value="paraview" />
|
||||
</list>
|
||||
</option>
|
||||
</inspection_tool>
|
||||
<inspection_tool class="PyStubPackagesAdvertiser" enabled="true" level="WARNING" enabled_by_default="true">
|
||||
<option name="ignoredPackages">
|
||||
<list>
|
||||
<option value="pandas" />
|
||||
</list>
|
||||
</option>
|
||||
</inspection_tool>
|
||||
</profile>
|
||||
</component>
|
||||
6
.idea/inspectionProfiles/profiles_settings.xml
generated
6
.idea/inspectionProfiles/profiles_settings.xml
generated
@@ -1,6 +0,0 @@
|
||||
<component name="InspectionProjectProfileManager">
|
||||
<settings>
|
||||
<option name="USE_PROJECT_PROFILE" value="false" />
|
||||
<version value="1.0" />
|
||||
</settings>
|
||||
</component>
|
||||
8
.idea/modules.xml
generated
8
.idea/modules.xml
generated
@@ -1,8 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="ProjectModuleManager">
|
||||
<modules>
|
||||
<module fileurl="file://$PROJECT_DIR$/.idea/wireless-formalism.iml" filepath="$PROJECT_DIR$/.idea/wireless-formalism.iml" />
|
||||
</modules>
|
||||
</component>
|
||||
</project>
|
||||
6
.idea/vcs.xml
generated
6
.idea/vcs.xml
generated
@@ -1,6 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="VcsDirectoryMappings">
|
||||
<mapping directory="" vcs="Git" />
|
||||
</component>
|
||||
</project>
|
||||
10
.idea/wireless-formalism.iml
generated
10
.idea/wireless-formalism.iml
generated
@@ -1,10 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<module external.system.id="pyproject.toml" type="PYTHON_MODULE" version="4">
|
||||
<component name="NewModuleRootManager">
|
||||
<content url="file://$MODULE_DIR$">
|
||||
<excludeFolder url="file://$MODULE_DIR$/.venv" />
|
||||
</content>
|
||||
<orderEntry type="jdk" jdkName="~/src/wireless-formalism/.venv" jdkType="Python SDK" />
|
||||
<orderEntry type="sourceFolder" forTests="false" />
|
||||
</component>
|
||||
</module>
|
||||
82
chains.py
Normal file
82
chains.py
Normal file
@@ -0,0 +1,82 @@
|
||||
"""Enumerate flattened redstone delay-chains for a given total delay."""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from typing import NamedTuple
|
||||
|
||||
|
||||
class Comp(NamedTuple):
|
||||
delay: int
|
||||
priority: int
|
||||
blocks: tuple
|
||||
|
||||
|
||||
# Append custom subcomponents here, e.g. Comp(6, -1, ("-", "R2", "C")).
|
||||
COMPONENTS = [
|
||||
Comp(8, -3, ("R4",)),
|
||||
Comp(6, -3, ("R3",)),
|
||||
Comp(4, -3, ("R2",)),
|
||||
Comp(2, -3, ("R1",)),
|
||||
Comp(8, -1, ("-", "R4")),
|
||||
Comp(6, -1, ("-", "R3")),
|
||||
Comp(4, -1, ("-", "R2")),
|
||||
Comp(2, -1, ("C",)),
|
||||
Comp(2, -1, ("-", "R1")),
|
||||
Comp(2, 0, ("-", "C")),
|
||||
]
|
||||
|
||||
assert all(c.blocks for c in COMPONENTS), "components must have >= 1 block"
|
||||
_ORDERED = sorted(COMPONENTS, key=lambda c: (c.priority, -c.delay))
|
||||
|
||||
|
||||
def _canon(comps):
|
||||
seen = {}
|
||||
for c in comps:
|
||||
seen.setdefault((c.priority, c.delay), c)
|
||||
return list(seen.values())
|
||||
|
||||
|
||||
HEAD = _canon([c for c in _ORDERED if c.blocks[0] == "-"])
|
||||
BODY = _canon(_ORDERED)
|
||||
|
||||
|
||||
def chains(total: int):
|
||||
for c in HEAD:
|
||||
if c.delay > total:
|
||||
continue
|
||||
if total == c.delay:
|
||||
yield c.blocks
|
||||
else:
|
||||
for rest in _body(total - c.delay):
|
||||
yield c.blocks + rest
|
||||
|
||||
|
||||
def _body(remaining: int):
|
||||
for c in BODY:
|
||||
if c.delay > remaining:
|
||||
continue
|
||||
if remaining == c.delay:
|
||||
yield c.blocks
|
||||
else:
|
||||
for rest in _body(remaining - c.delay):
|
||||
yield c.blocks + rest
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description=__doc__)
|
||||
ap.add_argument("total", type=int, help="total delay to enumerate")
|
||||
ap.add_argument("count", nargs="*", type=int, help="total element count")
|
||||
args = ap.parse_args()
|
||||
|
||||
tileset = chains(args.total)
|
||||
if args.count:
|
||||
tileset = (chain for chain in tileset if len(chain) in args.count)
|
||||
tileset = list(tileset)
|
||||
|
||||
for chain in tileset:
|
||||
print(*chain)
|
||||
print(len(tileset), "chains", file=sys.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,533 +0,0 @@
|
||||
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)
|
||||
366
finite_state.py
366
finite_state.py
@@ -1,366 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
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 | float, ...]:
|
||||
if self.delay == 0:
|
||||
return ()
|
||||
return (self.priority,) + (-inf,) * (self.delay - 1)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Transition:
|
||||
from_state: str
|
||||
to_state: str
|
||||
entries: Tuple[Entry, ...]
|
||||
|
||||
def linearize(self) -> Tuple[int, ...]:
|
||||
return tuple(p for e in self.entries for p in e.linear_parts())
|
||||
|
||||
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[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)
|
||||
|
||||
def _check_divergence(
|
||||
self, prefix: Tuple[int, ...], target: Tuple[int, ...]
|
||||
) -> int:
|
||||
"""
|
||||
Returns -1 if prefix is lexicographically < target,
|
||||
1 if prefix is lexicographically > target,
|
||||
0 if prefix is a strict prefix of target (no divergence yet).
|
||||
"""
|
||||
for p_val, t_val in zip(prefix, target):
|
||||
if p_val < t_val:
|
||||
return -1
|
||||
if p_val > t_val:
|
||||
return 1
|
||||
return 0
|
||||
|
||||
def find_next(
|
||||
self, current_lin: Tuple[int, ...], target_delay: int
|
||||
) -> Optional[List[Entry]]:
|
||||
best_next_chain = None
|
||||
best_next_lin = None
|
||||
|
||||
def dfs(
|
||||
chain_so_far: Tuple[Transition, ...], delay_so_far: int, current_state: str
|
||||
):
|
||||
nonlocal best_next_chain, best_next_lin
|
||||
|
||||
# Base Case: Exact delay reached AND machine is in an accepting state
|
||||
if delay_so_far == target_delay:
|
||||
if current_state in self.accept_states:
|
||||
lin = tuple(p for t in chain_so_far for p in t.linearize())
|
||||
if lin > current_lin:
|
||||
if best_next_lin is None or lin < best_next_lin:
|
||||
best_next_lin = lin
|
||||
best_next_chain = chain_so_far
|
||||
return
|
||||
|
||||
if delay_so_far > target_delay:
|
||||
return
|
||||
|
||||
p_lin = tuple(p for t in chain_so_far for p in t.linearize())
|
||||
|
||||
# Prefix Pruning
|
||||
if self._check_divergence(p_lin, current_lin) == -1:
|
||||
return
|
||||
if best_next_lin and self._check_divergence(p_lin, best_next_lin) == 1:
|
||||
return
|
||||
|
||||
for t in self.adj_list.get(current_state, []):
|
||||
dfs(chain_so_far + (t,), delay_so_far + t.total_delay(), t.to_state)
|
||||
|
||||
dfs((), 0, self.start_state)
|
||||
|
||||
if best_next_chain:
|
||||
return [entry for t in best_next_chain for entry in t.entries]
|
||||
return None
|
||||
|
||||
def parse_names(self, names: List[str]) -> List[Entry]:
|
||||
"""
|
||||
Takes a sequence of component names and infers their priorities and delays
|
||||
by walking the state machine.
|
||||
"""
|
||||
current_state = self.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 self.adj_list.get(current_state, []):
|
||||
# Extract the names of the components in this transition
|
||||
t_names = [e.name for e in t.entries]
|
||||
|
||||
# Check if this transition matches the next components in our input
|
||||
if names[i : i + len(t_names)] == 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 self.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
|
||||
|
||||
|
||||
transitions = []
|
||||
|
||||
# We expand our signals to track if the unique -2 priority was utilized
|
||||
signals = ["NORMAL", "INV_PENDING", "INV_UTILIZED"]
|
||||
contexts = ["DIODE", "WIRE", "OTHER"]
|
||||
|
||||
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}_WIRE",
|
||||
entries=(Entry(0, 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"),),
|
||||
)
|
||||
)
|
||||
|
||||
# D. Repeaters
|
||||
for d in (2, 4, 6, 8):
|
||||
if context == "DIODE":
|
||||
# Facing a diode yields -3 and leaves our utilization state unchanged
|
||||
rep_pri = -3
|
||||
next_signal = signal
|
||||
allow_rep = True
|
||||
else:
|
||||
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
|
||||
|
||||
if allow_rep:
|
||||
transitions.append(
|
||||
Transition(
|
||||
from_state=current_state,
|
||||
to_state=f"{next_signal}_DIODE",
|
||||
entries=(Entry(rep_pri, d, f"re{d}"),),
|
||||
)
|
||||
)
|
||||
|
||||
# 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"),
|
||||
)
|
||||
transitions.append(
|
||||
Transition(
|
||||
from_state=current_state,
|
||||
to_state="NORMAL_OTHER",
|
||||
entries=fluid_macro,
|
||||
)
|
||||
)
|
||||
|
||||
# ==========================================
|
||||
# 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="START", accept_states=accept_states
|
||||
)
|
||||
|
||||
# ==========================================
|
||||
# 2. USAGE EXAMPLES
|
||||
# ==========================================
|
||||
|
||||
|
||||
def display_chain(entries):
|
||||
# Print the chain, highlighting the name and generated priority
|
||||
return " <- ".join([f"{e.name}({e.priority})" for e in entries])
|
||||
|
||||
|
||||
def linearize_entries(entries):
|
||||
return tuple(p for e in entries for p in e.linear_parts())
|
||||
|
||||
|
||||
def generate_all_fast(model: TilesetFSM, target_delay: int) -> List[List[Entry]]:
|
||||
"""
|
||||
Uses Dynamic Programming to generate all valid chains of a given delay in O(N log N) time,
|
||||
completely eliminating dead-end traversal.
|
||||
"""
|
||||
# 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)}
|
||||
|
||||
# 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
|
||||
|
||||
# 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 = []
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
with tqdm.tqdm() as bar:
|
||||
build_chain(model.start_state, target_delay, ())
|
||||
|
||||
# 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())
|
||||
|
||||
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
|
||||
|
||||
# 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()),
|
||||
)
|
||||
|
||||
return sorted_chains
|
||||
|
||||
|
||||
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(",")
|
||||
for el in linearize_entries(chain):
|
||||
f.write(f"{el:+}" if isfinite(el) else " ")
|
||||
f.write(" ")
|
||||
f.write("\n")
|
||||
184
grammar.py
184
grammar.py
@@ -1,184 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import Tuple, Optional, List
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Entry:
|
||||
priority: int
|
||||
delay: int
|
||||
name: str = "-"
|
||||
|
||||
def linear_parts(self) -> Tuple[int, ...]:
|
||||
return (self.priority,) * self.delay
|
||||
|
||||
|
||||
# A ComponentBlock is an atomic sequence of Entries that can be physically built.
|
||||
ComponentBlock = Tuple[Entry, ...]
|
||||
|
||||
|
||||
def block_linearize(block: ComponentBlock) -> Tuple[int, ...]:
|
||||
"""Linearizes a single macro-block."""
|
||||
return tuple(p for e in block for p in e.linear_parts())
|
||||
|
||||
|
||||
def chain_linearize(chain: Tuple[ComponentBlock, ...]) -> Tuple[int, ...]:
|
||||
"""Linearizes a full sequence of blocks."""
|
||||
return tuple(p for b in chain for p in block_linearize(b))
|
||||
|
||||
|
||||
def flatten_chain(chain: Tuple[ComponentBlock, ...]) -> List[Entry]:
|
||||
"""Converts the internal tuple-based chain back to your flat List[Entry] format."""
|
||||
return [entry for block in chain for entry in block]
|
||||
|
||||
|
||||
class TilesetModel:
|
||||
def __init__(self, allowed_blocks: List[ComponentBlock]):
|
||||
self.blocks = allowed_blocks
|
||||
|
||||
def _check_divergence(
|
||||
self, prefix: Tuple[int, ...], target: Tuple[int, ...]
|
||||
) -> int:
|
||||
"""
|
||||
Returns -1 if prefix is lexicographically < target,
|
||||
1 if prefix is lexicographically > target,
|
||||
0 if prefix is a strict prefix of target (no divergence yet).
|
||||
"""
|
||||
for p_val, t_val in zip(prefix, target):
|
||||
if p_val < t_val:
|
||||
return -1
|
||||
if p_val > t_val:
|
||||
return 1
|
||||
return 0
|
||||
|
||||
def find_next(
|
||||
self, current_chain: Tuple[ComponentBlock, ...]
|
||||
) -> Optional[Tuple[ComponentBlock, ...]]:
|
||||
"""Finds the lexicographically next chain that perfectly matches the total delay."""
|
||||
target_delay = sum(e.delay for b in current_chain for e in b)
|
||||
current_lin = chain_linearize(current_chain)
|
||||
|
||||
best_next_chain = None
|
||||
best_next_lin = None
|
||||
|
||||
def dfs(chain_so_far: Tuple[ComponentBlock, ...], delay_so_far: int):
|
||||
nonlocal best_next_chain, best_next_lin
|
||||
|
||||
# Base Case: Exact delay reached
|
||||
if delay_so_far == target_delay:
|
||||
lin = chain_linearize(chain_so_far)
|
||||
if lin > current_lin:
|
||||
if best_next_lin is None or lin < best_next_lin:
|
||||
best_next_lin = lin
|
||||
best_next_chain = chain_so_far
|
||||
return
|
||||
|
||||
# Base Case: Overshot delay limits
|
||||
if delay_so_far > target_delay:
|
||||
return
|
||||
|
||||
p_lin = chain_linearize(chain_so_far)
|
||||
|
||||
# PRUNING 1: If prefix diverges and is strictly LESS than current_lin,
|
||||
# any suffix appended to it will also be strictly less. Prune the branch.
|
||||
if self._check_divergence(p_lin, current_lin) == -1:
|
||||
return
|
||||
|
||||
# PRUNING 2: If prefix diverges and is strictly GREATER than the best_next_lin
|
||||
# we've already found, any suffix will also be greater. It can't beat our current best.
|
||||
if best_next_lin is not None:
|
||||
if self._check_divergence(p_lin, best_next_lin) == 1:
|
||||
return
|
||||
|
||||
# Branching
|
||||
for b in self.blocks:
|
||||
dfs(chain_so_far + (b,), delay_so_far + sum(e.delay for e in b))
|
||||
|
||||
dfs((), 0)
|
||||
return best_next_chain
|
||||
|
||||
def find_previous(
|
||||
self, current_chain: Tuple[ComponentBlock, ...]
|
||||
) -> Optional[Tuple[ComponentBlock, ...]]:
|
||||
"""Finds the lexicographically previous chain matching total delay."""
|
||||
target_delay = sum(e.delay for b in current_chain for e in b)
|
||||
current_lin = chain_linearize(current_chain)
|
||||
|
||||
best_prev_chain = None
|
||||
best_prev_lin = None
|
||||
|
||||
def dfs(chain_so_far: Tuple[ComponentBlock, ...], delay_so_far: int):
|
||||
nonlocal best_prev_chain, best_prev_lin
|
||||
|
||||
if delay_so_far == target_delay:
|
||||
lin = chain_linearize(chain_so_far)
|
||||
if lin < current_lin:
|
||||
if best_prev_lin is None or lin > best_prev_lin:
|
||||
best_prev_lin = lin
|
||||
best_prev_chain = chain_so_far
|
||||
return
|
||||
if delay_so_far > target_delay:
|
||||
return
|
||||
|
||||
p_lin = chain_linearize(chain_so_far)
|
||||
|
||||
# Prune if prefix diverges and is strictly GREATER than current_lin
|
||||
if self._check_divergence(p_lin, current_lin) == 1:
|
||||
return
|
||||
|
||||
# Prune if prefix diverges and is strictly LESS than best_prev_lin
|
||||
if best_prev_lin is not None:
|
||||
if self._check_divergence(p_lin, best_prev_lin) == -1:
|
||||
return
|
||||
|
||||
for b in self.blocks:
|
||||
dfs(chain_so_far + (b,), delay_so_far + sum(e.delay for e in b))
|
||||
|
||||
dfs((), 0)
|
||||
return best_prev_chain
|
||||
|
||||
|
||||
# --- Define the "Alphabet" of valid Macro-Blocks ---
|
||||
|
||||
# Standard Components
|
||||
repeaters = [(Entry(p, d, f"rep({p},{d})"),) for p in (-1, -3) for d in (2, 4, 6, 8)]
|
||||
comparators = [(Entry(p, 2, f"cmp({p})"),) for p in (0, -1)]
|
||||
others = [(Entry(0, 2, "other"),)]
|
||||
|
||||
# Subsequences: Signal Inverted Repeaters
|
||||
# (Note: Updates flow right-to-left. Torch2 updates latest so it goes on the left)
|
||||
inv_repeaters = [
|
||||
(Entry(0, 2, "torch2"), Entry(p, d, f"inv_rep({p},{d})"), Entry(0, 2, "torch1"))
|
||||
for p in (-2, -3)
|
||||
for d in (2, 4, 6, 8)
|
||||
]
|
||||
|
||||
# Subsequences: Fluids & Observers
|
||||
# (Observer updates latest so it goes on the left)
|
||||
fluids = [
|
||||
(Entry(0, 2, "obs"), Entry(1, d, f"fluid({d})"), Entry(0, 4, "disp"))
|
||||
for d in (5, 10, 30)
|
||||
]
|
||||
|
||||
# Compile the generative grammar
|
||||
ALL_MACRO_BLOCKS = repeaters + comparators + others + inv_repeaters + fluids
|
||||
|
||||
# --- Usage Example ---
|
||||
|
||||
model = TilesetModel(ALL_MACRO_BLOCKS)
|
||||
|
||||
# Let's say we have a specific chain of Total Delay = 14
|
||||
current = (
|
||||
(Entry(0, 2, "cmp(0)"),),
|
||||
(Entry(-3, 8, "rep(-3,8)"),),
|
||||
(Entry(-1, 4, "rep(-1,4)"),),
|
||||
)
|
||||
|
||||
print(f"Current Chain: {flatten_chain(current)}")
|
||||
|
||||
next_chain = model.find_next(current)
|
||||
if next_chain:
|
||||
print(f"Next Chain: {flatten_chain(next_chain)}")
|
||||
|
||||
prev_chain = model.find_previous(current)
|
||||
if prev_chain:
|
||||
print(f"Prev Chain: {flatten_chain(prev_chain)}")
|
||||
190
permute.py
190
permute.py
@@ -1,190 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
from pprint import pprint
|
||||
from random import shuffle
|
||||
from typing import Iterable
|
||||
|
||||
# Here we model tilesets as "chains"; sequences of named components and their priority-delay pairs.
|
||||
# Components are processed via a priority queue; so on each tick, components which are scheduled to update will be processed in priority order.
|
||||
# When components update, they trigger the following component to update after some delay. Then, in the future, that component is updated according to its priority and the process continues down the chain.
|
||||
# If two components update in the same tick with the same priority, then they do so in the order in which they were scheduled.
|
||||
|
||||
# Note that components which update later in time have stronger significance on the order of execution; so for our notation we place these on the left-hand-side (first in lists) to match left-to-right lexicographic notation.
|
||||
|
||||
|
||||
@dataclass
|
||||
class Entry:
|
||||
priority: int
|
||||
delay: int
|
||||
name: str = "-"
|
||||
|
||||
|
||||
# We can model this by linearizing these tuples into tuples, then lexicographically sort these linearized keys.
|
||||
# We directly sort on priority: more-negative priorities update before more-positive priorities, so these are correctly sorted lexicographically. However we only want this behavior in the case that the components update at the same time, so we can 'pad out' the effect of delay by repeating the priority.
|
||||
|
||||
|
||||
def linear_parts(entry: Entry):
|
||||
assert entry.delay > 0
|
||||
# return (-np.inf,) * (entry.delay - 1) + (entry.priority,)
|
||||
return (entry.priority,) * entry.delay
|
||||
|
||||
|
||||
def linearize(entries: Iterable[Entry]):
|
||||
return [part for entry in entries for part in linear_parts(entry)]
|
||||
|
||||
|
||||
def sort(chains: list[Entry]):
|
||||
return sorted(chains, key=linearize)
|
||||
|
||||
|
||||
# one framework to produce a tileset is to encode binary values by comparator and repeater. all cases have equal delay, so it just falls to the priorities. comparators have priority 0 at the end and -1 in the middle, while repeaters have priority -1 at the end and -2 in the middle.
|
||||
|
||||
chains = [
|
||||
[
|
||||
Entry(-0, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
],
|
||||
[
|
||||
Entry(-0, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-2, 2, "rep"),
|
||||
],
|
||||
[
|
||||
Entry(-0, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-2, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
],
|
||||
[
|
||||
Entry(-0, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-2, 2, "rep"),
|
||||
Entry(-2, 2, "rep"),
|
||||
],
|
||||
[
|
||||
Entry(-0, 2, "cmp"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
],
|
||||
[
|
||||
Entry(-0, 2, "cmp"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-3, 2, "rep"),
|
||||
],
|
||||
[
|
||||
Entry(-0, 2, "cmp"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
],
|
||||
[
|
||||
Entry(-0, 2, "cmp"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
],
|
||||
[
|
||||
Entry(-1, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
],
|
||||
[
|
||||
Entry(-1, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-3, 2, "rep"),
|
||||
],
|
||||
[
|
||||
Entry(-1, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
],
|
||||
[
|
||||
Entry(-1, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
],
|
||||
[
|
||||
Entry(-1, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
],
|
||||
[
|
||||
Entry(-1, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
Entry(-3, 2, "rep"),
|
||||
],
|
||||
[
|
||||
Entry(-1, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-1, 2, "cmp"),
|
||||
],
|
||||
[
|
||||
Entry(-1, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
Entry(-3, 2, "rep"),
|
||||
],
|
||||
]
|
||||
shuffle(chains)
|
||||
|
||||
# another framework is to use permutations of repeaters on varying delay. now all cases have equal priority, so it just falls to the delays.
|
||||
|
||||
chains = [
|
||||
[Entry(-1, 8, "r4"), Entry(-3, 6, "r3"), Entry(-3, 4, "r2"), Entry(-3, 2, "r1")],
|
||||
[Entry(-1, 8, "r4"), Entry(-3, 6, "r3"), Entry(-3, 2, "r1"), Entry(-3, 4, "r2")],
|
||||
[Entry(-1, 8, "r4"), Entry(-3, 4, "r2"), Entry(-3, 6, "r3"), Entry(-3, 2, "r1")],
|
||||
[Entry(-1, 8, "r4"), Entry(-3, 4, "r2"), Entry(-3, 2, "r1"), Entry(-3, 6, "r3")],
|
||||
[Entry(-1, 8, "r4"), Entry(-3, 2, "r1"), Entry(-3, 6, "r3"), Entry(-3, 4, "r2")],
|
||||
[Entry(-1, 8, "r4"), Entry(-3, 2, "r1"), Entry(-3, 4, "r2"), Entry(-3, 6, "r3")],
|
||||
[Entry(-1, 6, "r3"), Entry(-3, 8, "r4"), Entry(-3, 4, "r2"), Entry(-3, 2, "r1")],
|
||||
[Entry(-1, 6, "r3"), Entry(-3, 8, "r4"), Entry(-3, 2, "r1"), Entry(-3, 4, "r2")],
|
||||
[Entry(-1, 6, "r3"), Entry(-3, 4, "r2"), Entry(-3, 8, "r4"), Entry(-3, 2, "r1")],
|
||||
[Entry(-1, 6, "r3"), Entry(-3, 4, "r2"), Entry(-3, 2, "r1"), Entry(-3, 8, "r4")],
|
||||
[Entry(-1, 6, "r3"), Entry(-3, 2, "r1"), Entry(-3, 8, "r4"), Entry(-3, 4, "r2")],
|
||||
[Entry(-1, 6, "r3"), Entry(-3, 2, "r1"), Entry(-3, 4, "r2"), Entry(-3, 8, "r4")],
|
||||
[Entry(-1, 4, "r2"), Entry(-3, 8, "r4"), Entry(-3, 6, "r3"), Entry(-3, 2, "r1")],
|
||||
[Entry(-1, 4, "r2"), Entry(-3, 8, "r4"), Entry(-3, 2, "r1"), Entry(-3, 6, "r3")],
|
||||
[Entry(-1, 4, "r2"), Entry(-3, 6, "r3"), Entry(-3, 8, "r4"), Entry(-3, 2, "r1")],
|
||||
[Entry(-1, 4, "r2"), Entry(-3, 6, "r3"), Entry(-3, 2, "r1"), Entry(-3, 8, "r4")],
|
||||
[Entry(-1, 4, "r2"), Entry(-3, 2, "r1"), Entry(-3, 8, "r4"), Entry(-3, 6, "r3")],
|
||||
[Entry(-1, 4, "r2"), Entry(-3, 2, "r1"), Entry(-3, 6, "r3"), Entry(-3, 8, "r4")],
|
||||
[Entry(-1, 2, "r1"), Entry(-3, 8, "r4"), Entry(-3, 6, "r3"), Entry(-3, 4, "r2")],
|
||||
[Entry(-1, 2, "r1"), Entry(-3, 8, "r4"), Entry(-3, 4, "r2"), Entry(-3, 6, "r3")],
|
||||
[Entry(-1, 2, "r1"), Entry(-3, 6, "r3"), Entry(-3, 8, "r4"), Entry(-3, 4, "r2")],
|
||||
[Entry(-1, 2, "r1"), Entry(-3, 6, "r3"), Entry(-3, 4, "r2"), Entry(-3, 8, "r4")],
|
||||
[Entry(-1, 2, "r1"), Entry(-3, 4, "r2"), Entry(-3, 8, "r4"), Entry(-3, 6, "r3")],
|
||||
[Entry(-1, 2, "r1"), Entry(-3, 4, "r2"), Entry(-3, 6, "r3"), Entry(-3, 8, "r4")],
|
||||
]
|
||||
shuffle(chains)
|
||||
|
||||
# However neither of these are the general case.
|
||||
#
|
||||
# Basically, we want a framework that can take a particular chain of (priority, delay) pairs and come up with the *next* (or previous) sequence given this `linearize` ordering using the (priority, delay) pairs which we can generate.
|
||||
#
|
||||
# the full list of generable pairs is described as follows.
|
||||
#
|
||||
# repeaters: [-1,-3] x [2,4,6,8]
|
||||
# comparators: [-0,-1] x [2]
|
||||
# all other components: [-0] x [2]
|
||||
#
|
||||
# it is in principle possible to generate sub-sequences of other tuples, however these require involved setup.
|
||||
#
|
||||
# by using redstone torches to invert a signal, it is possible to generate a subsequence involving priority -2. (While the signal is inverted, repeaters generate priorities [-2,-3] x [2,4,6,8])
|
||||
# [(-0,2), (-2,2), (-0,2)]
|
||||
#
|
||||
# by using fluids and observers, it is possible to generate signals of delay 5, 30 (in the overworld) or delay 10 only (in the nether). fluids tick after tile ticks, so it is effectively a priority of +1.
|
||||
# [(-0,2), (+1,5), (-0,4)] # the sequence here is a dispenser (-0,4) produces water (+1,5) which is observed by observer (-0,2).
|
||||
|
||||
# pprint([tuple(el.name for el in chain) for chain in chains])
|
||||
pprint([tuple(el.name for el in chain) for chain in sort(chains)])
|
||||
170
simple-enum.py
170
simple-enum.py
@@ -1,170 +0,0 @@
|
||||
from contextlib import redirect_stdout
|
||||
from functools import cache
|
||||
from pathlib import Path
|
||||
from pprint import pprint
|
||||
|
||||
|
||||
class Comp:
|
||||
def __init__(self, priority: int, delay: int, name: str, head: bool = True, blocks = ()):
|
||||
self.priority = priority
|
||||
self.delay = delay
|
||||
self.head = head
|
||||
self.name = name
|
||||
self.blocks = tuple(blocks)
|
||||
|
||||
def __lt__(self, other: Comp) -> bool:
|
||||
if self.priority == other.priority:
|
||||
return self.delay > other.delay
|
||||
else:
|
||||
return self.priority < other.priority
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.name
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.name} ({self.priority} / {self.delay})"
|
||||
|
||||
COMPONENTS = [
|
||||
Comp(priority=-3, delay=8, name="R4", head=False, blocks=[
|
||||
'repeater facing=south powered=false locked=false delay=4',
|
||||
]),
|
||||
Comp(priority=-3, delay=6, name="R3", head=False, blocks=[
|
||||
'repeater facing=south powered=false locked=false delay=3'
|
||||
]),
|
||||
Comp(priority=-3, delay=4, name="R2", head=False, blocks=[
|
||||
'repeater facing=south powered=false locked=false delay=2'
|
||||
]),
|
||||
Comp(priority=-3, delay=2, name="R1", head=False, blocks=[
|
||||
'repeater facing=south powered=false locked=false delay=1'
|
||||
]),
|
||||
Comp(priority=-1, delay=8, name="- R4", head=True, blocks=[
|
||||
'redstone_wire north=side south=side power=0',
|
||||
'repeater facing=south powered=false locked=false delay=4'
|
||||
]),
|
||||
Comp(priority=-1, delay=6, name="- R3", head=True, blocks=[
|
||||
'redstone_wire north=side south=side power=0',
|
||||
'repeater facing=south powered=false locked=false delay=3'
|
||||
]),
|
||||
Comp(priority=-1, delay=4, name="- R2", head=True, blocks=[
|
||||
'redstone_wire north=side south=side power=0',
|
||||
'repeater facing=south powered=false locked=false delay=2'
|
||||
]),
|
||||
Comp(priority=-1, delay=2, name="C", head=False, blocks=[
|
||||
'comparator facing=south powered=false mode=compare'
|
||||
]),
|
||||
Comp(priority=-1, delay=2, name="- R1", head=True, blocks=[
|
||||
'redstone_wire north=side south=side power=0',
|
||||
'repeater facing=south powered=false locked=false delay=1'
|
||||
]),
|
||||
Comp(priority=0, delay=2, name="- C", head=True, blocks=[
|
||||
'redstone_wire north=side south=side power=0',
|
||||
'comparator facing=south powered=false mode=compare'
|
||||
]),
|
||||
]
|
||||
|
||||
COMPONENTS.sort()
|
||||
|
||||
|
||||
from functools import cache
|
||||
|
||||
|
||||
@cache
|
||||
def tileset(total_delay, head=True):
|
||||
tree = {}
|
||||
seen = set()
|
||||
|
||||
for comp in COMPONENTS:
|
||||
if comp.delay <= total_delay:
|
||||
if head and not comp.head:
|
||||
continue
|
||||
|
||||
key = (comp.priority, comp.delay)
|
||||
if key in seen:
|
||||
continue
|
||||
|
||||
seen.add(key)
|
||||
tree[comp] = tileset(total_delay - comp.delay, head=False)
|
||||
|
||||
return tree
|
||||
|
||||
|
||||
def show(tree, flat=False):
|
||||
if flat:
|
||||
|
||||
def _print_flat(current_tree, current_path):
|
||||
# If the dict is empty, we've reached the end of a valid chain
|
||||
if not current_tree:
|
||||
print(" ".join(str(comp) for comp in current_path))
|
||||
return
|
||||
for comp, subtree in current_tree.items():
|
||||
_print_flat(subtree, current_path + [comp])
|
||||
|
||||
_print_flat(tree, [])
|
||||
|
||||
else:
|
||||
|
||||
def _print_tree(current_tree, prefix=""):
|
||||
items = list(current_tree.items())
|
||||
for i, (comp, subtree) in enumerate(items):
|
||||
is_last = i == len(items) - 1
|
||||
connector = "└ " if is_last else "├ "
|
||||
print(f"{prefix}{connector}{comp}")
|
||||
|
||||
# If this is the last item, children don't need a vertical line
|
||||
extension = " " if is_last else "│ "
|
||||
_print_tree(subtree, prefix + extension)
|
||||
|
||||
_print_tree(tree)
|
||||
|
||||
|
||||
# def diorama(tree, step=1.125):
|
||||
# x = 0
|
||||
#
|
||||
# def inner(prefix, node):
|
||||
# nonlocal x
|
||||
#
|
||||
# if not node:
|
||||
# z = 0
|
||||
# for pref in prefix:
|
||||
# for block in pref.blocks:
|
||||
# print(f'p {x:.3f} 0 {z} {block}')
|
||||
# print(f'p {x:.3f} -1 {z} smooth_stone_slab type=top')
|
||||
# z += 1
|
||||
# x -= step
|
||||
# else:
|
||||
# for comp, sub in node.items():
|
||||
# inner((*prefix, comp), sub)
|
||||
#
|
||||
# inner((), tree)
|
||||
|
||||
def dioramas(tree):
|
||||
count = 0
|
||||
|
||||
def inner(prefix, node):
|
||||
nonlocal count
|
||||
|
||||
if not node:
|
||||
count += 1
|
||||
print('```mc-diorama')
|
||||
z = 0
|
||||
for pref in prefix:
|
||||
for block in pref.blocks:
|
||||
print(f'p 0 0 {z} {block}')
|
||||
print(f'p 0 -1 {z} smooth_stone_slab type=top')
|
||||
z += 1
|
||||
print('```')
|
||||
else:
|
||||
for comp, sub in node.items():
|
||||
inner((*prefix, comp), sub)
|
||||
|
||||
inner((), tree)
|
||||
|
||||
return count
|
||||
|
||||
|
||||
# show(tileset(4, head=True))
|
||||
# print()
|
||||
|
||||
with Path('~/src/wireless/content/tilesets-lexicographic.typ').expanduser().open('w') as f, redirect_stdout(f):
|
||||
total = dioramas(tileset(10, head=True))
|
||||
print(f'{total = }')
|
||||
Reference in New Issue
Block a user