2023 day 20.

src/holt59/aoc/2023/day19.py

@@ -28,12 +28,12 @@ def accept(workflows: dict[str, Workflow], part: Part) -> bool:
 
     while decision is None:
         for check, target in workflows[workflow]:
-            ok = check is None
+            passed = check is None
             if check is not None:
                 category, sense, value = check
-                ok = OPERATORS[sense](part[category], value)
+                passed = OPERATORS[sense](part[category], value)
 
-            if ok:
+            if passed:
                 if target in workflows:
                     workflow = target
                 else:
@@ -44,35 +44,42 @@ def accept(workflows: dict[str, Workflow], part: Part) -> bool:
 
 
 def propagate(workflows: dict[str, Workflow], start: PartWithBounds) -> int:
+    def _fmt(meta: PartWithBounds) -> str:
+        return "{" + ", ".join(f"{k}={v}" for k, v in meta.items()) + "}"
+
     def transfer_or_accept(
         target: str, meta: PartWithBounds, queue: list[tuple[PartWithBounds, str]]
     ) -> int:
+        count = 0
         if target in workflows:
             logging.info(f"    transfer to {target}")
             queue.append((meta, target))
-            return 0
         elif target == "A":
-            logging.info("  accepted")
-            return prod((high - low + 1) for low, high in meta.values())
+            count = prod((high - low + 1) for low, high in meta.values())
+            logging.info(f"    accepted ({count})")
         else:
             logging.info("    rejected")
-            return 0
+        return count
 
     accepted = 0
     queue: list[tuple[PartWithBounds, str]] = [(start, "in")]
 
+    n_iterations = 0
+
     while queue:
+        n_iterations += 1
         meta, workflow = queue.pop()
-        logging.info(f"{workflow}: {meta}")
+        logging.info(f"{workflow}: {_fmt(meta)}")
         for check, target in workflows[workflow]:
             if check is None:
+                logging.info("  end-of-workflow")
                 accepted += transfer_or_accept(target, meta, queue)
                 continue
 
             category, sense, value = check
             bounds, op = meta[category], OPERATORS[sense]
 
-            logging.info(f"  splitting {meta} into {category} {sense} {value}")
+            logging.info(f"  checking {_fmt(meta)} against {category} {sense} {value}")
 
             if not op(bounds[0], value) and not op(bounds[1], value):
                 logging.info("    reject, always false")
@@ -84,16 +91,16 @@ def propagate(workflows: dict[str, Workflow], start: PartWithBounds) -> int:
                 break
 
             meta2 = meta.copy()
+            low, high = meta[category]
             if sense == "<":
-                meta2[category] = (meta[category][0], value - 1)
-                meta[category] = (value, meta[category][1])
+                meta[category], meta2[category] = (value, high), (low, value - 1)
             else:
-                meta2[category] = (value + 1, meta[category][1])
-                meta[category] = (meta[category][0], value)
-            logging.info(f"  split {meta2} ({target}), {meta}")
+                meta[category], meta2[category] = (low, value), (value + 1, high)
+            logging.info(f"    split {_fmt(meta2)} ({target}), {_fmt(meta)}")
 
             accepted += transfer_or_accept(target, meta2, queue)
 
+    logging.info(f"run took {n_iterations} iterations")
     return accepted
 
 

src/holt59/aoc/2023/day20.py

@@ -1,13 +1,161 @@
+import logging
+import os
 import sys
 from collections import defaultdict
-from dataclasses import dataclass
+from math import lcm
+from typing import Literal, TypeAlias
+
+VERBOSE = os.getenv("AOC_VERBOSE") == "True"
+logging.basicConfig(level=logging.INFO if VERBOSE else logging.WARNING)
+
+
+ModuleType: TypeAlias = Literal["broadcaster", "conjunction", "flip-flop"]
+PulseType: TypeAlias = Literal["high", "low"]
+
+modules: dict[str, tuple[ModuleType, list[str]]] = {}
 
 lines = sys.stdin.read().splitlines()
 
+for line in lines:
+    name, outputs_s = line.split(" -> ")
+    outputs = outputs_s.split(", ")
+    if name == "broadcaster":
+        modules["broadcaster"] = ("broadcaster", outputs)
+    else:
+        modules[name[1:]] = (
+            "conjunction" if name.startswith("&") else "flip-flop",
+            outputs,
+        )
+
+
+def process(
+    start: tuple[str, str, PulseType],
+    flip_flop_states: dict[str, Literal["on", "off"]],
+    conjunction_states: dict[str, dict[str, PulseType]],
+) -> tuple[dict[PulseType, int], dict[str, dict[PulseType, int]]]:
+    pulses: list[tuple[str, str, PulseType]] = [start]
+    counts: dict[PulseType, int] = {"low": 0, "high": 0}
+    inputs: dict[str, dict[PulseType, int]] = defaultdict(lambda: {"low": 0, "high": 0})
+
+    logging.info("starting process... ")
+
+    while pulses:
+        input, name, pulse = pulses.pop(0)
+        logging.info(f"{input} -{pulse}-> {name}")
+        counts[pulse] += 1
+
+        inputs[name][pulse] += 1
+
+        if name not in modules:
+            continue
+
+        type, outputs = modules[name]
+
+        if type == "broadcaster":
+            ...
+
+        elif type == "flip-flop":
+            if pulse == "high":
+                continue
+
+            if flip_flop_states[name] == "off":
+                flip_flop_states[name] = "on"
+                pulse = "high"
+            else:
+                flip_flop_states[name] = "off"
+                pulse = "low"
+
+        else:
+            conjunction_states[name][input] = pulse
+
+            if all(state == "high" for state in conjunction_states[name].values()):
+                pulse = "low"
+            else:
+                pulse = "high"
+
+        pulses.extend((name, output, pulse) for output in outputs)
+
+    return counts, inputs
+
+
+with open("./day20.dot", "w") as fp:
+    fp.write("digraph G {\n")
+    fp.write("rx [shape=circle, color=red, style=filled];\n")
+    for name, (type, outputs) in modules.items():
+        if type == "conjunction":
+            shape = "diamond"
+        elif type == "flip-flop":
+            shape = "box"
+        else:
+            shape = "circle"
+        fp.write(f"{name} [shape={shape}];\n")
+    for name, (type, outputs) in modules.items():
+        for output in outputs:
+            fp.write(f"{name} -> {output};\n")
+    fp.write("}\n")
+
 # part 1
-answer_1 = ...
+flip_flop_states: dict[str, Literal["on", "off"]] = {
+    name: "off" for name, (type, _) in modules.items() if type == "flip-flop"
+}
+conjunction_states: dict[str, dict[str, PulseType]] = {
+    name: {input: "low" for input, (_, outputs) in modules.items() if name in outputs}
+    for name, (type, _) in modules.items()
+    if type == "conjunction"
+}
+counts: dict[PulseType, int] = {"low": 0, "high": 0}
+for _ in range(1000):
+    result, _ = process(
+        ("button", "broadcaster", "low"), flip_flop_states, conjunction_states
+    )
+    for pulse in ("low", "high"):
+        counts[pulse] += result[pulse]
+answer_1 = counts["low"] * counts["high"]
 print(f"answer 1 is {answer_1}")
 
 # part 2
-answer_2 = ...
+
+# reset states
+for name in flip_flop_states:
+    flip_flop_states[name] = "off"
+
+for name in conjunction_states:
+    for input in conjunction_states[name]:
+        conjunction_states[name][input] = "low"
+
+# find the conjunction connected to rx
+to_rx = [name for name, (_, outputs) in modules.items() if "rx" in outputs]
+assert len(to_rx) == 1, "cannot handle multiple module inputs for rx"
+assert (
+    modules[to_rx[0]][0] == "conjunction"
+), "can only handle conjunction as input to rx"
+
+to_rx_inputs = [name for name, (_, outputs) in modules.items() if to_rx[0] in outputs]
+assert all(
+    modules[i][0] == "conjunction" and len(modules[i][1]) == 1 for i in to_rx_inputs
+), "can only handle inversion as second-order inputs to rx"
+
+
+count = 1
+cycles: dict[str, int] = {}
+second: dict[str, int] = {}
+while len(second) != len(to_rx_inputs):
+    _, inputs = process(
+        ("button", "broadcaster", "low"), flip_flop_states, conjunction_states
+    )
+
+    for node in to_rx_inputs:
+        if inputs[node]["low"] == 1:
+            if node not in cycles:
+                cycles[node] = count
+            elif node not in second:
+                second[node] = count
+
+    count += 1
+
+assert all(
+    second[k] == cycles[k] * 2 for k in to_rx_inputs
+), "cannot only handle cycles starting at the beginning"
+
+answer_2 = lcm(*cycles.values())
 print(f"answer 2 is {answer_2}")

src/holt59/aoc/inputs/holt59/2023/day20.txt

@@ -0,0 +1,58 @@
+%cf -> tz
+%kr -> xn, gq
+%cp -> sq, bd
+broadcaster -> vn, sj, tg, kn
+%hc -> pm
+%fd -> xn, mj
+%qz -> xf
+%vf -> mc, pm
+%zm -> rz, pm
+%cn -> bd, qz
+%jj -> bp
+%ks -> ff
+%nb -> xn, ks
+%bm -> pm, vf
+&xn -> kc, jb, cb, tg, ks, tx
+%lm -> rk
+%dn -> bd, cn
+%ft -> dn
+%pn -> pm, ll
+%rk -> bp, fs
+%tz -> bp, gp
+%mc -> jx
+%fs -> kx
+%jf -> bd, fm
+%rz -> hc, pm
+%tg -> cb, xn
+&hf -> rx
+%vp -> pn
+&pm -> ll, mc, sj, vd, vp
+%rn -> kc, xn
+%vn -> bd, cp
+&nd -> hf
+%fm -> bd, gc
+%ff -> xn, fd
+&bp -> cf, fh, pc, kn, fs, gn, lm
+&pc -> hf
+%mj -> xn
+%qg -> bd
+%fh -> lm
+%kc -> nb
+%xf -> bd, jf
+%gc -> qg, bd
+&bd -> vn, sq, qz, ft, nd
+%jb -> kr
+%gp -> bp, rp
+%gq -> xn, rn
+%sj -> pm, bm
+%rp -> bp, jj
+%sq -> ft
+%cb -> jb
+&vd -> hf
+%gn -> cf
+%kx -> gn, bp
+%ll -> zm
+&tx -> hf
+%jx -> md, pm
+%md -> pm, vp
+%kn -> fh, bp

src/holt59/aoc/inputs/tests/2023/day20.txt

@@ -0,0 +1,5 @@
+broadcaster -> a, b, c
+%a -> b
+%b -> c
+%c -> inv
+&inv -> a

src/holt59/aoc/inputs/tests/2023/day20_2.txt

@@ -0,0 +1,5 @@
+broadcaster -> a
+%a -> inv, con
+&inv -> b
+%b -> con
+&con -> output