fibonacci_ring_observatory_v1.zip (9.3 KB)
fibonacci_ring_observatory_smoke_logs.zip (5.2 KB)
#!/usr/bin/env python3
"""Fibonacci Ring Cosmological Observatory
A self-contained Python 3.13 reference implementation of the supplied
Fibonacci Ring pseudocode plus an independent, read-only observatory and
shadow-only mutation laboratory.
No octave/depth counter is stored in the primary ring. The address K is
recovered from the current Fibonacci pair (A, B).
Standard library only. Exact topology is represented as a finite abstract
simplicial complex. Homology is computed over F_2.
IMPORTANT RESOURCE NOTE:
Barycentric subdivision of a tetrahedral complex multiplies the number of
3-simplices by 24 per refinement. Use small --steps initially. The default
runs through the first subdivision (K=4), not thousands of subdivisions.
"""
from __future__ import annotations
import argparse
import copy
import hashlib
import itertools
import json
import math
import os
import platform
import sys
import time
from dataclasses import dataclass, field
from decimal import Decimal, getcontext
from pathlib import Path
from typing import Any, Iterable
VERSION = "1.0.0"
getcontext().prec = 80
def decimal_phi() -> Decimal:
"""Compute phi from its fixed point x = 1 + 1/x; no stored phi constant."""
x = Decimal("1.5")
one = Decimal(1)
for _ in range(400):
nxt = one + one / x
if nxt == x:
break
x = nxt
return x
def fib_index(a: int, b: int) -> int:
"""Recover k from (F_k,F_{k+1}) by reversing the recurrence."""
if a < 0 or b < 0:
raise ValueError("Fibonacci pair must be nonnegative")
if (a, b) == (0, 1):
return 0
x, y = a, b
reverse_steps = 0
while (x, y) != (0, 1):
if y < x:
raise ValueError(f"Not a canonical Fibonacci pair: {(a, b)}")
x, y = y - x, x
reverse_steps += 1
if reverse_steps > 100000:
raise ValueError("Fibonacci pair index recovery exceeded safety bound")
return reverse_steps
def fib_pair_at(k: int) -> tuple[int, int]:
if k < 0:
raise ValueError("k must be nonnegative")
a, b = 0, 1
for _ in range(k):
a, b = b, a + b
return a, b
def canonical_simplex(s: Iterable[int]) -> tuple[int, ...]:
return tuple(sorted(s))
@dataclass
class SimplicialComplex:
"""Finite abstract simplicial complex; simplices include all faces."""
simplices: set[tuple[int, ...]] = field(default_factory=set)
next_vertex: int = 0
@classmethod
def complete_simplex(cls, vertex_count: int) -> "SimplicialComplex":
if vertex_count < 1:
raise ValueError("A simplex requires at least one vertex")
out = cls()
vertices = tuple(range(vertex_count))
for r in range(1, vertex_count + 1):
for face in itertools.combinations(vertices, r):
out.simplices.add(face)
out.next_vertex = vertex_count
return out
@property
def vertices(self) -> set[int]:
return {s[0] for s in self.simplices if len(s) == 1}
def dimension(self) -> int:
return max((len(s) - 1 for s in self.simplices), default=-1)
def vertex_count(self) -> int:
return len(self.vertices)
def edge_count(self) -> int:
return sum(1 for s in self.simplices if len(s) == 2)
def euler(self) -> int:
return sum((1 if (len(s) - 1) % 2 == 0 else -1) for s in self.simplices)
def adjacency(self) -> set[tuple[int, int]]:
edges = {s for s in self.simplices if len(s) == 2}
return {(e[0], e[1]) for e in edges}
def adjacency_signature(self) -> str:
return digest_json(sorted([list(e) for e in self.adjacency()]))
def signature(self) -> str:
"""Deterministic labelled signature, suitable for exact replay checks."""
payload = [list(s) for s in sorted(self.simplices, key=lambda z: (len(z), z))]
return digest_json(payload)
def validate(self) -> bool:
if not self.simplices:
return False
if any(not s for s in self.simplices):
return False
if any(tuple(sorted(set(s))) != s for s in self.simplices):
return False
for simplex in self.simplices:
for r in range(1, len(simplex)):
for face in itertools.combinations(simplex, r):
if face not in self.simplices:
return False
return True
def barycentric_subdivision(self, max_simplices: int = 2_000_000) -> "SimplicialComplex":
"""Subdivide: new vertices are old nonempty faces; simplices are chains."""
old_faces = sorted(self.simplices, key=lambda s: (len(s), s))
face_id = {face: i for i, face in enumerate(old_faces)}
if len(old_faces) > max_simplices:
raise MemoryError(f"Subdivision needs {len(old_faces)} new vertices; cap={max_simplices}")
result: set[tuple[int, ...]] = set()
# Every chain has a unique largest face. Generate chains inside each old simplex.
for top in old_faces:
subfaces = [f for f in old_faces if len(f) <= len(top) and set(f).issubset(top)]
subfaces.sort(key=lambda f: (len(f), f))
# Recursively enumerate all nonempty strictly increasing face chains.
def extend(chain: list[tuple[int, ...]], candidates: list[tuple[int, ...]]) -> None:
if chain:
ids = tuple(face_id[f] for f in chain)
result.add(ids)
if len(result) > max_simplices:
raise MemoryError(f"Subdivision exceeded simplex cap={max_simplices}")
last_size = len(chain[-1]) if chain else 0
for face in candidates:
if len(face) > last_size and (not chain or set(chain[-1]).issubset(face)):
extend(chain + [face], candidates)
extend([], subfaces)
return SimplicialComplex(result, next_vertex=len(old_faces))
def betti_numbers_f2(self) -> tuple[int, ...]:
"""Compute Betti numbers using boundary-matrix ranks over F_2."""
if not self.simplices:
return tuple()
maxdim = self.dimension()
by_dim = [[s for s in self.simplices if len(s) == d + 1] for d in range(maxdim + 1)]
ranks = [0] * (maxdim + 2) # rank(boundary_d), indexed by d
for d in range(1, maxdim + 1):
lower = {s: i for i, s in enumerate(by_dim[d - 1])}
basis: dict[int, int] = {}
for simplex in by_dim[d]:
col = 0
for face in itertools.combinations(simplex, d):
col ^= (1 << lower[face])
while col:
pivot = col.bit_length() - 1
if pivot in basis:
col ^= basis[pivot]
else:
basis[pivot] = col
break
ranks[d] = len(basis)
betti = []
for d in range(maxdim + 1):
betti.append(len(by_dim[d]) - ranks[d] - ranks[d + 1])
return tuple(betti)
def digest_json(value: Any) -> str:
raw = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()
return hashlib.sha256(raw).hexdigest()
def complex_signature(complex_: SimplicialComplex | None) -> str | None:
return None if complex_ is None else complex_.signature()
def topology_metrics(c: SimplicialComplex | None) -> dict[str, Any] | None:
if c is None:
return None
return {
"dimension": c.dimension(),
"euler": c.euler(),
"betti_f2": list(c.betti_numbers_f2()),
"vertices": c.vertex_count(),
"edges": c.edge_count(),
"adjacency_signature": c.adjacency_signature(),
"signature": c.signature(),
}
@dataclass
class RingObservation:
ring_state: tuple[int, int]
address: int
phi_power: str
echo: str
orientation: str
strands: dict[str, list[float]]
input_signature: str | None
output_signature: str
operation: str
verification: dict[str, bool]
input_metrics: dict[str, Any] | None
output_metrics: dict[str, Any]
transition_hash: str = ""
def payload(self) -> dict[str, Any]:
return {
"ring_state": list(self.ring_state), "address": self.address,
"phi_power": self.phi_power, "echo": self.echo,
"orientation": self.orientation, "strands": self.strands,
"input_signature": self.input_signature,
"output_signature": self.output_signature,
"operation": self.operation, "verification": self.verification,
"input_metrics": self.input_metrics, "output_metrics": self.output_metrics,
}
class FibonacciRing:
"""Primary system: persistent state is exactly Ω, A, B, TOPOLOGY, TRACE."""
def __init__(self, radius: float = 1.0, axial_scale: float = 1.0,
subdivision_cap: int = 2_000_000):
self.omega = decimal_phi()
self.a, self.b = 0, 1
self.topology: SimplicialComplex | None = None
self.trace: list[RingObservation] = []
self.g = 2.0 * math.pi * (2.0 - float(self.omega))
self.radius = radius
self.axial_scale = axial_scale
self.subdivision_cap = subdivision_cap
def resolve(self) -> RingObservation:
a, b = self.a, self.b
k = fib_index(a, b)
p = Decimal(a) * self.omega + Decimal(b - a)
e = Decimal(1) / p
angle = k * self.g
v = k + 1
plus = [self.radius * math.cos(angle), self.radius * math.sin(angle), self.axial_scale * k]
minus = [self.radius * math.cos(angle), -self.radius * math.sin(angle), self.axial_scale * k]
old = self.topology
input_metrics = topology_metrics(old)
if b < 5:
candidate = SimplicialComplex.complete_simplex(v)
operation = "COMPLETE_SIMPLEX"
checks = {"valid_simplicial_complex": candidate.validate()}
if k == 3:
checks["initial_dimension_3"] = candidate.dimension() == 3
else:
if old is None:
raise AssertionError("TOPOLOGY is NULL when subdivision is required")
candidate = old.barycentric_subdivision(self.subdivision_cap)
operation = "BARYCENTRIC_SUBDIVISION"
checks = {
"valid_simplicial_complex": candidate.validate(),
"dimension_preserved": candidate.dimension() == old.dimension(),
"euler_preserved": candidate.euler() == old.euler(),
"homology_preserved": candidate.betti_numbers_f2() == old.betti_numbers_f2(),
"vertex_count_increased": candidate.vertex_count() > old.vertex_count(),
"adjacency_changed": candidate.adjacency() != old.adjacency(),
}
if not all(checks.values()):
raise AssertionError(f"Required ring verification failed: {checks}")
obs = RingObservation(
ring_state=(a, b), address=k, phi_power=str(p), echo=str(e),
orientation=repr(angle), strands={"plus": plus, "minus": minus},
input_signature=complex_signature(old), output_signature=candidate.signature(),
operation=operation, verification=checks,
input_metrics=input_metrics, output_metrics=topology_metrics(candidate) or {},
)
self.trace.append(obs)
self.topology = candidate
return obs
def advance(self) -> None:
self.a, self.b = self.b, self.a + self.b
def step(self) -> RingObservation:
obs = self.resolve()
self.advance()
return obs
def snapshot(self) -> dict[str, Any]:
return {
"omega": str(self.omega), "a": self.a, "b": self.b,
"topology_signature": complex_signature(self.topology),
"trace_length": len(self.trace),
}
# ---------------------------- Independent observers -------------------------
def observer_progression(record: dict[str, Any], omega: Decimal) -> dict[str, Any]:
a, b = record["ring_state"]
k = fib_index(a, b)
p = Decimal(a) * omega + Decimal(b - a)
e = Decimal(1) / p
recorded_p = Decimal(record["phi_power"])
recorded_e = Decimal(record["echo"])
return {
"observer": "progression",
"pair_valid": fib_pair_at(k) == (a, b),
"address_matches": k == record["address"],
"phi_power_matches": p == recorded_p,
"phi_exponent_identity": abs(p - (omega ** k)) < Decimal("1e-65"),
"echo_reciprocal": abs(recorded_p * recorded_e - 1) < Decimal("1e-65"),
}
def observer_structure(c: SimplicialComplex | None, recorded: dict[str, Any] | None) -> dict[str, Any]:
valid = bool(c and c.validate())
# Do not run boundary-matrix calculations on a malformed, non-closed complex.
actual = topology_metrics(c) if valid else None
return {
"observer": "structure",
"topology_present": c is not None,
"valid_complex": valid,
"metrics_match": valid and actual == recorded,
"actual_metrics": actual,
}
def observer_history(trace_payloads: list[dict[str, Any]], initial: SimplicialComplex | None,
current: SimplicialComplex | None) -> dict[str, Any]:
"""Replay operations from trace; does not read the primary current topology."""
replay: SimplicialComplex | None = copy.deepcopy(initial)
checks = []
for entry in trace_payloads:
input_sig = complex_signature(replay)
if input_sig != entry["input_signature"]:
checks.append(False)
break
if entry["operation"] == "COMPLETE_SIMPLEX":
k = entry["address"]
replay = SimplicialComplex.complete_simplex(k + 1)
elif entry["operation"] == "BARYCENTRIC_SUBDIVISION":
if replay is None:
checks.append(False)
break
replay = replay.barycentric_subdivision()
else:
checks.append(False)
break
checks.append(complex_signature(replay) == entry["output_signature"])
return {
"observer": "history",
"all_transitions_replay": bool(checks) and all(checks),
"transition_checks": len(checks),
"reconstructed_matches_current": complex_signature(replay) == complex_signature(current),
"reconstructed_signature": complex_signature(replay),
}
def observer_geometry(record: dict[str, Any]) -> dict[str, Any]:
p = record["strands"]["plus"]
m = record["strands"]["minus"]
tol = 1e-10
angle = float(record["orientation"])
expected_x = math.cos(angle)
expected_y = math.sin(angle)
return {
"observer": "geometry",
"shared_axial_coordinate": abs(p[2] - m[2]) <= tol,
"opposite_transverse_orientation": abs(p[1] + m[1]) <= tol,
"shared_x_coordinate": abs(p[0] - m[0]) <= tol,
"angle_matches_strands": abs(p[0] - expected_x) <= tol and abs(p[1] - expected_y) <= tol and abs(m[1] + expected_y) <= tol,
}
# ------------------------------ Observatory ---------------------------------
class Observatory:
def __init__(self, ring: FibonacciRing, log_dir: Path):
self.ring = ring
self.log_dir = log_dir
self.log_dir.mkdir(parents=True, exist_ok=True)
self.baseline: dict[str, Any] | None = None
self.initial_topology: SimplicialComplex | None = None
self.ledger_path = self.log_dir / "observations.jsonl"
self.mutation_path = self.log_dir / "mutation_results.jsonl"
self.chain_hash = "0" * 64
def seal_baseline(self) -> None:
self.baseline = {
"ring_state": [self.ring.a, self.ring.b],
"omega": str(self.ring.omega),
"topology_signature": complex_signature(self.ring.topology),
"topology_metrics": topology_metrics(self.ring.topology),
"trace_length": len(self.ring.trace),
}
self.initial_topology = copy.deepcopy(self.ring.topology)
self._append_json(self.log_dir / "baseline.json", self.baseline, overwrite=True)
def _append_json(self, path: Path, obj: Any, overwrite: bool = False) -> None:
mode = "w" if overwrite else "a"
with path.open(mode, encoding="utf-8") as f:
f.write(json.dumps(obj, sort_keys=True, separators=(",", ":")) + "\n")
def observe_transition(self, obs: RingObservation) -> dict[str, Any]:
payload = obs.payload()
p_report = observer_progression(payload, self.ring.omega)
s_report = observer_structure(self.ring.topology, obs.output_metrics)
history_payloads = [x.payload() for x in self.ring.trace]
h_report = observer_history(history_payloads, None, self.ring.topology)
g_report = observer_geometry(payload)
reports = [p_report, s_report, h_report, g_report]
all_pass = all(v for report in reports for k, v in report.items()
if isinstance(v, bool))
self.chain_hash = hashlib.sha256((self.chain_hash + json.dumps(payload, sort_keys=True)).encode()).hexdigest()
entry = {
"timestamp_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"transition": payload,
"observers": reports,
"all_observers_pass": all_pass,
"ledger_chain_hash": self.chain_hash,
}
self._append_json(self.ledger_path, entry)
return entry
def verify_primary_unchanged(self, before: dict[str, Any]) -> bool:
return self.ring.snapshot() == before
def blind_mutation_campaign(self, trials_per_family: int = 1) -> list[dict[str, Any]]:
if self.ring.topology is None or not self.ring.trace:
return []
results: list[dict[str, Any]] = []
primary_before = self.ring.snapshot()
last_record = self.ring.trace[-1].payload()
families = ["ANGLE_DRIFT", "VERTEX_REMOVAL", "EDGE_REMOVAL", "TRACE_DELETION",
"SIGNATURE_FORGERY", "VERTEX_RELABELING_ISOMORPHISM"]
for family in families:
for trial in range(trials_per_family):
shadow_topology = copy.deepcopy(self.ring.topology)
shadow_record = copy.deepcopy(last_record)
shadow_trace = [x.payload() for x in self.ring.trace]
ground_truth: dict[str, Any] = {"family": family, "trial": trial}
if family == "ANGLE_DRIFT":
shadow_record["orientation"] = repr(float(shadow_record["orientation"]) + 0.01)
elif family == "VERTEX_REMOVAL":
vertices = sorted(shadow_topology.vertices)
if len(vertices) > 1:
victim = vertices[-1]
shadow_topology.simplices = {s for s in shadow_topology.simplices if victim not in s}
elif family == "EDGE_REMOVAL":
edges = sorted(s for s in shadow_topology.simplices if len(s) == 2)
if edges:
shadow_topology.simplices.discard(edges[-1])
elif family == "VERTEX_RELABELING_ISOMORPHISM":
# This is not a topological mutation: it is an isomorphic relabeling.
# It tests whether the observer confuses vertex identity with topology.
verts = sorted(shadow_topology.vertices)
mapping = {v: verts[(i + 1) % len(verts)] for i, v in enumerate(verts)}
shadow_topology.simplices = {tuple(sorted(mapping[v] for v in s)) for s in shadow_topology.simplices}
elif family == "TRACE_DELETION":
if shadow_trace:
shadow_trace.pop()
elif family == "SIGNATURE_FORGERY":
shadow_record["output_signature"] = "0" * 64
# Blind observers see shadow evidence only; ground truth is revealed afterward.
# Ensure mutations to the current record are reflected in the shadow trace.
if shadow_trace:
shadow_trace[-1] = copy.deepcopy(shadow_record)
p = observer_progression(shadow_record, self.ring.omega)
s = observer_structure(shadow_topology, shadow_record.get("output_metrics"))
g = observer_geometry(shadow_record)
# The history observer independently replays the shadow trace against the
# unmodified initial seed (NULL before first ring transition).
h = observer_history(shadow_trace, None, shadow_topology)
reports = [p, s, g, h]
detected = not all(v for report in reports for k, v in report.items() if isinstance(v, bool))
result = {
"family": family, "trial": trial,
"detected": detected,
"reports": reports,
"ground_truth": ground_truth,
"primary_unchanged": self.verify_primary_unchanged(primary_before),
}
results.append(result)
self._append_json(self.mutation_path, result)
if not self.verify_primary_unchanged(primary_before):
raise AssertionError("Mutation campaign altered the primary ring")
return results
def tomography(self, windows: tuple[int, ...] = (1, 8, 32, 128, 512)) -> dict[str, Any]:
n = len(self.ring.trace)
slices = {}
for width in windows:
if n == 0:
slices[str(width)] = {"segments": 0}
else:
starts = list(range(0, n, width))
slices[str(width)] = {
"segments": len(starts),
"segment_signatures": [
digest_json([x.payload() for x in self.ring.trace[s:s + width]])
for s in starts
],
}
return {"trace_length": n, "windows": slices}
def run(args: argparse.Namespace) -> int:
log_dir = Path(args.log_dir)
ring = FibonacciRing(radius=args.radius, axial_scale=args.axial_scale,
subdivision_cap=args.max_simplices)
observatory = Observatory(ring, log_dir)
observatory.seal_baseline()
print(f"Fibonacci Ring Observatory v{VERSION}")
print(f"Python: {platform.python_version()} | log: {log_dir.resolve()}")
print(f"phi fixed point: {ring.omega}")
print("Primary progression state: Ω, A, B | no octave/depth counter")
print("\nTRANSITIONS")
try:
for i in range(args.steps):
obs = ring.step()
report = observatory.observe_transition(obs)
print(
f"k={obs.address:>3} pair={obs.ring_state!s:<18} op={obs.operation:<24} "
f"dim={obs.output_metrics['dimension']} V={obs.output_metrics['vertices']} "
f"simplices={len(ring.topology.simplices):>8} observers={'PASS' if report['all_observers_pass'] else 'FAIL'}"
)
except (AssertionError, MemoryError, ValueError) as exc:
print(f"\nSTOPPED SAFELY: {type(exc).__name__}: {exc}", file=sys.stderr)
print("The last committed transition and its logs have been preserved.", file=sys.stderr)
return 2
print("\nTOMOGRAPHY")
tomo = observatory.tomography()
observatory._append_json(log_dir / "tomography.jsonl", tomo)
print(json.dumps(tomo, indent=2))
if args.mutations:
print("\nBLIND SHADOW MUTATION CAMPAIGN")
results = observatory.blind_mutation_campaign(args.trials_per_family)
for r in results:
print(f"{r['family']:<30} detected={str(r['detected']):<5} primary_unchanged={r['primary_unchanged']}")
final = {
"version": VERSION,
"final_ring": ring.snapshot(),
"final_topology_metrics": topology_metrics(ring.topology),
"trace_length": len(ring.trace),
"tomography": tomo,
"files": ["baseline.json", "observations.jsonl", "tomography.jsonl"] +
(["mutation_results.jsonl"] if args.mutations else []),
}
observatory._append_json(log_dir / "final_report.json", final, overwrite=True)
print("\nFINAL REPORT")
print(json.dumps(final, indent=2))
print("\nCompleted. Review baseline.json, observations.jsonl, and final_report.json.")
return 0
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--steps", type=int, default=5,
help="number of ring transitions (default: 5; includes first barycentric subdivision)")
parser.add_argument("--log-dir", default="fibonacci_ring_observatory_run")
parser.add_argument("--mutations", action="store_true", help="run blind mutations on shadow copies")
parser.add_argument("--trials-per-family", type=int, default=1)
parser.add_argument("--radius", type=float, default=1.0)
parser.add_argument("--axial-scale", type=float, default=1.0)
parser.add_argument("--max-simplices", type=int, default=2_000_000,
help="safety cap for generated simplices")
args = parser.parse_args()
if args.steps < 1 or args.trials_per_family < 1:
parser.error("--steps and --trials-per-family must be >= 1")
return run(args)
if __name__ == "__main__":
raise SystemExit(main())
fibonacci_ring_observatory_v1_1.zip (12.0 KB)
fibonacci_ring_finite_observatory_v2.zip (15.1 KB)
infinite complexity engine (different development path than the above .zips)
import hashlib
import json
import math
from itertools import combinations
from typing import Dict, FrozenSet, List, Optional, Set, Tuple
def fixed_point_phi(tol: float = 1e-15, max_iter: int = 1000) -> float:
x = 1.0
for _ in range(max_iter):
nxt = 1.0 + 1.0 / x
if abs(nxt - x) <= tol:
return nxt
x = nxt
return x
def fib_index(a: int, b: int) -> int:
if (a, b) == (0, 1):
return 0
fa, fb = 0, 1
k = 0
while (fa, fb) != (a, b):
if fb > b:
raise ValueError(f"Invalid Fibonacci pair: ({a}, {b})")
fa, fb = fb, fa + fb
k += 1
return k
class SimplicialComplex:
def __init__(self, facets: List[FrozenSet[int]]):
self.faces: Set[FrozenSet[int]] = set()
for facet in facets:
if not facet:
continue
for r in range(1, len(facet) + 1):
for sub in combinations(facet, r):
self.faces.add(frozenset(sub))
@classmethod
def complete_simplex(cls, vertex_count: int) -> "SimplicialComplex":
if vertex_count < 1:
raise ValueError("Simplex must have at least 1 vertex.")
return cls([frozenset(range(vertex_count))])
def is_valid(self) -> bool:
if not self.faces:
return False
for face in self.faces:
if len(face) > 1:
for v in face:
sub = face - {v}
if sub not in self.faces:
return False
return True
def dimension(self) -> int:
if not self.faces:
return -1
return max(len(f) - 1 for f in self.faces)
def faces_of_dim(self, d: int) -> List[FrozenSet[int]]:
return sorted(
[f for f in self.faces if len(f) - 1 == d],
key=lambda s: tuple(sorted(s)),
)
def vertex_count(self) -> int:
return len(self.faces_of_dim(0))
def euler_characteristic(self) -> int:
dim = self.dimension()
chi = 0
for d in range(dim + 1):
count = len(self.faces_of_dim(d))
chi += count if d % 2 == 0 else -count
return chi
def homology_z2(self) -> Tuple[int, ...]:
dim = self.dimension()
if dim < 0:
return ()
ordered_faces: Dict[int, List[FrozenSet[int]]] = {
d: self.faces_of_dim(d) for d in range(dim + 1)
}
face_index: Dict[int, Dict[FrozenSet[int], int]] = {
d: {f: i for i, f in enumerate(ordered_faces[d])}
for d in range(dim + 1)
}
def boundary_rank(d: int) -> int:
if d <= 0 or d > dim:
return 0
cols: List[int] = []
for simplex in ordered_faces[d]:
s_sorted = sorted(simplex)
col_mask = 0
for idx in range(len(s_sorted)):
sub = frozenset(s_sorted[:idx] + s_sorted[idx + 1 :])
row = face_index[d - 1][sub]
col_mask ^= 1 << row
if col_mask != 0:
cols.append(col_mask)
basis: Dict[int, int] = {}
for col in cols:
c = col
while c:
pivot = c.bit_length() - 1
if pivot in basis:
c ^= basis[pivot]
else:
basis[pivot] = c
break
return len(basis)
ranks = {d: boundary_rank(d) for d in range(dim + 2)}
betti: List[int] = []
for d in range(dim + 1):
c_d = len(ordered_faces[d])
z_d = c_d - ranks[d]
b_d = ranks[d + 1]
betti.append(z_d - b_d)
return tuple(betti)
def adjacency_signature(self) -> Tuple[Tuple[int, int], ...]:
edges = self.faces_of_dim(1)
normalized = []
for e in edges:
u, v = sorted(e)
normalized.append((u, v))
return tuple(sorted(normalized))
def signature(self) -> str:
serialized = [sorted(list(f)) for f in self.faces]
serialized.sort(key=lambda x: (len(x), x))
payload = json.dumps(serialized, separators=(",", ":")).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
def barycentric_subdivision(self) -> "SimplicialComplex":
sorted_faces = sorted(
self.faces, key=lambda f: (len(f), tuple(sorted(f)))
)
n = len(sorted_faces)
adj: List[List[int]] = [[] for _ in range(n)]
for i in range(n):
fi = sorted_faces[i]
for j in range(i + 1, n):
fj = sorted_faces[j]
if fi < fj:
adj[i].append(j)
new_facets: List[FrozenSet[int]] = []
def dfs(curr: int, chain: List[int]) -> None:
if not adj[curr]:
new_facets.append(frozenset(chain))
return
for nxt in adj[curr]:
chain.append(nxt)
dfs(nxt, chain)
chain.pop()
for start in range(n):
dfs(start, [start])
return SimplicialComplex(new_facets)
class FibonacciRingTestMachine:
def __init__(self, radius: float = 1.0, pitch: float = 1.0):
self.R: float = radius
self.L: float = pitch
self.omega: float = 0.0
self.a: int = 0
self.b: int = 1
self.G: float = 0.0
self.topology: Optional[SimplicialComplex] = None
self.trace: List[Dict] = []
self.fib_reset()
def fib_reset(self) -> None:
self.omega = fixed_point_phi()
self.a = 0
self.b = 1
self.topology = None
self.trace = []
self.G = 2.0 * math.pi * (2.0 - self.omega)
def fib_resolve(self) -> Dict:
p = self.a * self.omega + (self.b - self.a)
e = 1.0 / p
k = fib_index(self.a, self.b)
angle = k * self.G
v = k + 1
s_plus = (
self.R * math.cos(angle),
self.R * math.sin(angle),
self.L * k,
)
s_minus = (
self.R * math.cos(angle),
-self.R * math.sin(angle),
self.L * k,
)
verification_results: Dict[str, bool] = {}
if self.b < 5:
operation_used = f"COMPLETE_SIMPLEX({v})"
candidate = SimplicialComplex.complete_simplex(v)
valid = candidate.is_valid()
assert valid, "Candidate failed VALID_SIMPLICIAL_COMPLEX check."
verification_results["VALID_SIMPLICIAL_COMPLEX"] = valid
if k == 3:
dim_ok = candidate.dimension() == 3
assert dim_ok, "Dimension at K=3 must equal 3."
verification_results["DIMENSION_EQ_3"] = dim_ok
else:
assert self.topology is not None, "TOPOLOGY must not be NULL when B >= 5."
old_topology = self.topology
operation_used = "BARYCENTRIC_SUBDIVISION"
candidate = old_topology.barycentric_subdivision()
dim_eq = candidate.dimension() == old_topology.dimension()
euler_eq = (
candidate.euler_characteristic()
== old_topology.euler_characteristic()
)
hom_eq = candidate.homology_z2() == old_topology.homology_z2()
v_inc = candidate.vertex_count() > old_topology.vertex_count()
adj_diff = (
candidate.adjacency_signature()
!= old_topology.adjacency_signature()
)
assert dim_eq, "Dimension invariant violated."
assert euler_eq, "Euler characteristic invariant violated."
assert hom_eq, "Homology invariant violated."
assert v_inc, "Vertex count did not strictly increase."
assert adj_diff, "Adjacency structure did not change."
verification_results.update(
{
"DIMENSION_PRESERVED": dim_eq,
"EULER_PRESERVED": euler_eq,
"HOMOLOGY_PRESERVED": hom_eq,
"VERTEX_COUNT_INCREASED": v_inc,
"ADJACENCY_CHANGED": adj_diff,
}
)
input_sig = self.topology.signature() if self.topology is not None else "NULL"
output_sig = candidate.signature()
observation = {
"RING_STATE": (self.a, self.b),
"ADDRESS": k,
"PHI_POWER": p,
"ECHO": e,
"ORIENTATION": angle,
"STRANDS": (s_plus, s_minus),
"INPUT_SIGNATURE": input_sig,
"OUTPUT_SIGNATURE": output_sig,
"OPERATION": operation_used,
"METRICS": {
"DIMENSION": candidate.dimension(),
"VERTICES": candidate.vertex_count(),
"TOTAL_FACES": len(candidate.faces),
"EULER": candidate.euler_characteristic(),
"BETTI_Z2": candidate.homology_z2(),
},
"VERIFICATION": verification_results,
}
assert all(observation["VERIFICATION"].values()), "Required checks failed."
self.trace.append(observation)
self.topology = candidate
return observation
def fib_advance(self) -> None:
temp = self.a + self.b
self.a = self.b
self.b = temp
def run(self, steps: int = 7) -> List[Dict]:
for _ in range(steps):
self.fib_resolve()
self.fib_advance()
return self.trace
if __name__ == "__main__":
machine = FibonacciRingTestMachine(radius=1.0, pitch=1.0)
trace = machine.run(steps=7)
for obs in trace:
m = obs["METRICS"]
print(
f"K={obs['ADDRESS']} | Ring={obs['RING_STATE']} | "
f"Op={obs['OPERATION']:<23} | V={m['VERTICES']:<4} | "
f"Faces={m['TOTAL_FACES']:<5} | Dim={m['DIMENSION']} | "
f"Chi={m['EULER']} | Betti={m['BETTI_Z2']} | "
f"Sig={obs['OUTPUT_SIGNATURE'][:12]}..."
)
lossless flattening of steps of same as above
#!/usr/bin/env python3
"""
FIBONACCI RING — LOSSLESS FLATTENED TOPOLOGY
Version 1.0
Preserves:
OMEGA Fibonacci fixed point
A, B Fibonacci ring state
K Derived ring address
PHI_POWER A*OMEGA + (B-A)
ECHO Reciprocal scale
ORIENTATION Ring-derived angular progression
STRANDS Original two-strand geometry
Topology:
Initial complete simplices through K=3.
Barycentric subdivision from K=4 onward.
LOSSLESS STORAGE:
facets_Kxxxx.bin
Every maximal tetrahedron, as four uint64 vertex IDs.
vertices_Kxxxx.bin
Every new vertex, represented by its exact parent face.
Each record contains:
one byte: parent face cardinality
cardinality x uint64: parent vertex IDs
manifest_Kxxxx.json
Ring state, face counts, invariants, file hashes.
A simplicial complex is completely determined by its maximal
simplices and their vertex identities. The parent-face records
also preserve the subdivision ancestry.
No encryption.
No external dependencies.
No topology approximation.
No random identifiers.
No discarded faces.
SQLite is used as disk-backed deduplication storage.
"""
import argparse
import hashlib
import itertools
import json
import math
import os
import shutil
import sqlite3
import struct
import sys
import time
from pathlib import Path
VERSION = "1.0"
FACET_MAGIC = b"FIBFAC1\0"
VERTEX_MAGIC = b"FIBVTX1\0"
HEADER = struct.Struct("<8sIIQ")
TETRAHEDRON = struct.Struct("<QQQQ")
UINT64 = struct.Struct("<Q")
# Each facet is one tetrahedron.
# Vertex identifiers are 1-based uint64 integers.
# ============================================================
# FIBONACCI RING — ORIGINAL MECHANISM
# ============================================================
def fixed_point_phi(tol=1e-15, max_iter=1000):
x = 1.0
for _ in range(max_iter):
nxt = 1.0 + 1.0 / x
if abs(nxt - x) <= tol:
return nxt
x = nxt
raise RuntimeError("PHI fixed point did not converge")
def fib_pair(k):
a, b = 0, 1
for _ in range(k):
a, b = b, a + b
return a, b
def fib_index(a, b):
fa, fb = 0, 1
k = 0
while (fa, fb) != (a, b):
if fb > b:
raise ValueError("Invalid Fibonacci ring state")
fa, fb = fb, fa + fb
k += 1
return k
def ring_resolve(k, radius=1.0, pitch=1.0):
omega = fixed_point_phi()
a, b = fib_pair(k)
p = a * omega + (b - a)
echo = 1.0 / p
address = fib_index(a, b)
G = 2.0 * math.pi * (2.0 - omega)
angle = address * G
s_plus = (
radius * math.cos(angle),
radius * math.sin(angle),
pitch * address,
)
s_minus = (
radius * math.cos(angle),
-radius * math.sin(angle),
pitch * address,
)
return {
"RING_STATE": [a, b],
"ADDRESS": address,
"OMEGA": omega,
"PHI_POWER": p,
"ECHO": echo,
"ORIENTATION": angle,
"STRANDS": [s_plus, s_minus],
}
# ============================================================
# LOSSLESS FACE ENCODING
# ============================================================
def face_key(vertices):
"""
Canonical, lossless encoding of one nonempty face.
Format:
uint8 cardinality
cardinality x uint64 vertex IDs
"""
vertices = tuple(sorted(vertices))
if not 1 <= len(vertices) <= 4:
raise ValueError("Unexpected face cardinality")
return (
struct.pack("<B", len(vertices))
+ struct.pack("<" + "Q" * len(vertices), *vertices)
)
def tetrahedron_faces(tet):
"""
Return all 15 nonempty faces of one tetrahedron.
"""
for r in range(1, 5):
for subset in itertools.combinations(tet, r):
yield face_key(subset)
# ============================================================
# FLATTENED BINARY STORAGE
# ============================================================
def write_facet_file(path, step, facets):
"""
Write a complete flattened maximal-simplex representation.
"""
facets = list(facets)
with open(path, "wb") as f:
f.write(
HEADER.pack(
FACET_MAGIC,
1,
step,
len(facets),
)
)
for tet in facets:
f.write(TETRAHEDRON.pack(*sorted(tet)))
def read_facet_header(path):
with open(path, "rb") as f:
raw = f.read(HEADER.size)
if len(raw) != HEADER.size:
raise ValueError("Invalid facet header")
magic, version, step, count = HEADER.unpack(raw)
if magic != FACET_MAGIC or version != 1:
raise ValueError("Unknown facet format")
return step, count
def iter_facets(path):
step, count = read_facet_header(path)
with open(path, "rb") as f:
f.seek(HEADER.size)
for _ in range(count):
raw = f.read(TETRAHEDRON.size)
if len(raw) != TETRAHEDRON.size:
raise IOError("Truncated facet file")
yield TETRAHEDRON.unpack(raw)
def write_vertex_header(f, step, count):
f.write(
HEADER.pack(
VERTEX_MAGIC,
1,
step,
count,
)
)
def export_vertex_provenance(db, path, step):
"""
Export every generated vertex's parent face.
Records are ordered by vertex ID.
The record position therefore identifies the vertex ID.
Each record is variable length but self-describing.
"""
cursor = db.execute(
"SELECT key FROM face_map ORDER BY id"
)
count = db.execute(
"SELECT COUNT(*) FROM face_map"
).fetchone()[0]
with open(path, "wb") as f:
write_vertex_header(f, step, count)
written = 0
for (key,) in cursor:
f.write(key)
written += 1
if written != count:
raise IOError("Vertex provenance count mismatch")
return count
# ============================================================
# SIGNATURES AND MANIFESTS
# ============================================================
def sha256_file(path):
h = hashlib.sha256()
with open(path, "rb") as f:
while True:
block = f.read(1024 * 1024)
if not block:
break
h.update(block)
return h.hexdigest()
def save_json(path, data):
with open(path, "w", encoding="utf-8") as f:
json.dump(
data,
f,
indent=2,
sort_keys=True,
)
f.write("\n")
# ============================================================
# FACE-COUNT RECURRENCE
# ============================================================
def subdivided_face_counts(f):
"""
Barycentric subdivision of a 3-dimensional complex.
Input:
f = (vertices, edges, triangles, tetrahedra)
Output:
exact face counts of the subdivision.
"""
f0, f1, f2, f3 = f
return (
f0 + f1 + f2 + f3,
2 * f1 + 6 * f2 + 14 * f3,
6 * f2 + 36 * f3,
24 * f3,
)
def complete_simplex_counts(k):
"""
Face counts for a complete simplex with k+1 vertices.
"""
n = k + 1
result = []
for r in range(1, n + 1):
result.append(math.comb(n, r))
return result
# ============================================================
# DISK-BACKED BARYCENTRIC SUBDIVISION
# ============================================================
def make_database(path):
db = sqlite3.connect(path)
db.execute("PRAGMA journal_mode=OFF")
db.execute("PRAGMA synchronous=OFF")
db.execute("PRAGMA temp_store=FILE")
db.execute("PRAGMA cache_size=-65536")
db.execute("PRAGMA locking_mode=EXCLUSIVE")
db.execute("""
CREATE TABLE face_map (
id INTEGER PRIMARY KEY,
dim INTEGER NOT NULL,
key BLOB NOT NULL UNIQUE
)
""")
db.commit()
return db
def subdivision_step(
input_facets,
output_facets,
output_vertices,
work_db,
step,
old_counts,
min_free_gb=2.0,
):
"""
Perform one complete barycentric subdivision.
Pass 1:
Enumerate every old face.
Assign one unique new vertex ID to each old face.
Pass 2:
Enumerate every maximal chain of faces.
Each chain becomes a new tetrahedron.
No old face is discarded.
No new vertex is assigned to more than one old face.
"""
started = time.perf_counter()
expected_vertices = sum(old_counts)
expected_tetrahedra = 24 * old_counts[3]
output_dir = Path(output_facets).parent
free_bytes = shutil.disk_usage(output_dir).free
minimum_bytes = int(min_free_gb * 1024**3)
if free_bytes < minimum_bytes:
raise OSError(
f"Insufficient free disk space: "
f"{free_bytes / 1024**3:.2f} GB"
)
if Path(work_db).exists():
Path(work_db).unlink()
db = make_database(work_db)
insert_sql = """
INSERT OR IGNORE INTO face_map(dim, key)
VALUES (?, ?)
"""
# --------------------------------------------------------
# PASS 1 — ENUMERATE AND DEDUPLICATE OLD FACES
# --------------------------------------------------------
print(f"\nK={step}: PASS 1 — flattening old faces")
batch = []
old_tet_count = 0
try:
for tet in iter_facets(input_facets):
old_tet_count += 1
for key in tetrahedron_faces(tet):
batch.append((key[0], key))
if len(batch) >= 15000:
db.executemany(insert_sql, batch)
db.commit()
batch.clear()
if old_tet_count % 10000 == 0:
print(
f" Old tetrahedra: {old_tet_count:,}",
flush=True,
)
if batch:
db.executemany(insert_sql, batch)
db.commit()
batch.clear()
actual_vertices = db.execute(
"SELECT COUNT(*) FROM face_map"
).fetchone()[0]
if actual_vertices != expected_vertices:
raise AssertionError(
f"Face deduplication mismatch: "
f"{actual_vertices} != {expected_vertices}"
)
print(
f" Unique faces / new vertices: "
f"{actual_vertices:,}"
)
# ----------------------------------------------------
# PASS 2 — GENERATE ALL MAXIMAL CHAINS
# ----------------------------------------------------
print(f"K={step}: PASS 2 — generating new tetrahedra")
select_sql = """
SELECT key, id
FROM face_map
WHERE key IN (
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
)
"""
produced = 0
with open(output_facets, "wb") as out:
out.write(
HEADER.pack(
FACET_MAGIC,
1,
step,
0,
)
)
for tet in iter_facets(input_facets):
keys = list(tetrahedron_faces(tet))
lookup = {
key: vertex_id
for key, vertex_id in db.execute(
select_sql,
keys,
)
}
if len(lookup) != 15:
raise AssertionError(
"Incomplete face-to-vertex mapping"
)
# Every permutation determines one maximal
# chain:
#
# {v0}
# {v0,v1}
# {v0,v1,v2}
# {v0,v1,v2,v3}
#
# There are exactly 4! = 24 chains per
# old tetrahedron.
for perm in itertools.permutations(tet):
chain = (
lookup[face_key((perm[0],))],
lookup[face_key(perm[:2])],
lookup[face_key(perm[:3])],
lookup[face_key(perm)],
)
out.write(
TETRAHEDRON.pack(
*sorted(chain)
)
)
produced += 1
if produced % 240000 == 0:
print(
f" New tetrahedra: {produced:,}",
flush=True,
)
if produced != expected_tetrahedra:
raise AssertionError(
f"Tetrahedron count mismatch: "
f"{produced} != {expected_tetrahedra}"
)
# Rewrite header with final count.
out.seek(0)
out.write(
HEADER.pack(
FACET_MAGIC,
1,
step,
produced,
)
)
# ----------------------------------------------------
# PASS 3 — EXPORT LOSSLESS VERTEX PROVENANCE
# ----------------------------------------------------
print(f"K={step}: PASS 3 — exporting provenance")
vertex_count = export_vertex_provenance(
db,
output_vertices,
step,
)
if vertex_count != expected_vertices:
raise AssertionError(
"Provenance vertex count mismatch"
)
db.close()
Path(work_db).unlink(missing_ok=True)
elapsed = time.perf_counter() - started
print(
f"K={step}: COMPLETE | "
f"V={vertex_count:,} | "
f"T={produced:,} | "
f"Time={elapsed:.2f}s"
)
return {
"VERTICES": vertex_count,
"TETRAHEDRA": produced,
"ELAPSED_SECONDS": elapsed,
"FACET_SHA256": sha256_file(output_facets),
"VERTEX_SHA256": sha256_file(output_vertices),
}
except Exception:
db.close()
print(
f"\nK={step}: FAILED. "
f"Work database retained if present."
)
raise
# ============================================================
# EULER CHARACTERISTIC
# ============================================================
def euler_characteristic(f):
return f[0] - f[1] + f[2] - f[3]
# ============================================================
# INITIAL SIMPLEX
# ============================================================
def write_initial_tetrahedron(path):
"""
K=3 is the initial tetrahedral complex.
"""
write_facet_file(
path,
3,
[(1, 2, 3, 4)],
)
# ============================================================
# TRACE AND EXECUTION
# ============================================================
def run_machine(
steps=8,
output="fibonacci_ring_flat",
max_tetrahedra=10_000_000,
min_free_gb=2.0,
):
if steps < 1:
raise ValueError("Steps must be positive")
output_dir = Path(output)
output_dir.mkdir(parents=True, exist_ok=True)
phi = fixed_point_phi()
G = 2.0 * math.pi * (2.0 - phi)
trace = []
# --------------------------------------------------------
# INITIAL COMPLETE SIMPLICES
# --------------------------------------------------------
for k in range(min(steps, 4)):
ring = ring_resolve(k)
counts = complete_simplex_counts(k)
observation = {
**ring,
"OPERATION": f"COMPLETE_SIMPLEX({k + 1})",
"METRICS": {
"DIMENSION": k,
"VERTICES": counts[0],
"TOTAL_FACES": sum(counts),
"FACE_COUNTS": counts,
"EULER": euler_characteristic(
counts + [0] * (4 - len(counts))
),
},
"VERIFICATION": {
"SIMPLEX_VALID": True,
},
}
trace.append(observation)
save_json(
output_dir / f"manifest_K{k:04d}.json",
observation,
)
print(
f"K={k} | Ring={ring['RING_STATE']} | "
f"Faces={sum(counts):,} | "
f"Operation=COMPLETE_SIMPLEX"
)
if steps <= 4:
save_json(output_dir / "trace.json", trace)
return trace
# --------------------------------------------------------
# INITIAL TETRAHEDRAL STATE
# --------------------------------------------------------
current_facets = output_dir / "facets_K0003.bin"
write_initial_tetrahedron(current_facets)
face_counts = (4, 6, 4, 1)
assert euler_characteristic(face_counts) == 1
# --------------------------------------------------------
# SUCCESSIVE SUBDIVISIONS
# --------------------------------------------------------
for k in range(4, steps):
expected_counts = subdivided_face_counts(face_counts)
expected_tetrahedra = expected_counts[3]
if expected_tetrahedra > max_tetrahedra:
print(
f"\nSTOP BEFORE K={k}: "
f"{expected_tetrahedra:,} tetrahedra "
f"exceeds configured limit "
f"{max_tetrahedra:,}."
)
break
output_facets = (
output_dir / f"facets_K{k:04d}.bin"
)
output_vertices = (
output_dir / f"vertices_K{k:04d}.bin"
)
work_db = (
output_dir / f"work_K{k:04d}.sqlite"
)
result = subdivision_step(
input_facets=current_facets,
output_facets=output_facets,
output_vertices=output_vertices,
work_db=work_db,
step=k,
old_counts=face_counts,
min_free_gb=min_free_gb,
)
# ----------------------------------------------------
# INVARIANT VERIFICATION
# ----------------------------------------------------
chi = euler_characteristic(expected_counts)
if chi != 1:
raise AssertionError(
f"Euler characteristic changed at K={k}"
)
if expected_counts[3] != result["TETRAHEDRA"]:
raise AssertionError(
"Tetrahedral count does not match recurrence"
)
if expected_counts[0] != result["VERTICES"]:
raise AssertionError(
"Vertex count does not match recurrence"
)
ring = ring_resolve(k)
observation = {
**ring,
"OPERATION": "BARYCENTRIC_SUBDIVISION",
"METRICS": {
"DIMENSION": 3,
"VERTICES": expected_counts[0],
"EDGES": expected_counts[1],
"TRIANGLES": expected_counts[2],
"TETRAHEDRA": expected_counts[3],
"TOTAL_FACES": sum(expected_counts),
"FACE_COUNTS": expected_counts,
"EULER": chi,
"BETTI_Z2_EXPECTED": [1, 0, 0, 0],
},
"VERIFICATION": {
"VERTEX_COUNT_EXACT": True,
"TETRAHEDRA_COUNT_EXACT": True,
"EULER_PRESERVED": True,
"DIMENSION_PRESERVED": True,
"HOMOLOGY_BY_SUBDIVISION_INVARIANCE": True,
},
"STORAGE": {
"FACETS_FILE": output_facets.name,
"VERTICES_FILE": output_vertices.name,
**result,
},
}
save_json(
output_dir / f"manifest_K{k:04d}.json",
observation,
)
trace.append(observation)
print(
f"\nK={k} | "
f"Ring={ring['RING_STATE']} | "
f"Vertices={expected_counts[0]:,} | "
f"Edges={expected_counts[1]:,} | "
f"Triangles={expected_counts[2]:,} | "
f"Tetrahedra={expected_counts[3]:,} | "
f"Faces={sum(expected_counts):,} | "
f"Chi={chi}"
)
current_facets = output_facets
face_counts = expected_counts
save_json(
output_dir / "trace.json",
trace,
)
save_json(
output_dir / "trace.json",
trace,
)
print("\n========================================")
print("FIBONACCI RING — EXECUTION COMPLETE")
print("========================================")
print(f"Steps recorded: {len(trace)}")
print(f"Output: {output_dir.resolve()}")
return trace
# ============================================================
# COMMAND LINE
# ============================================================
def main():
parser = argparse.ArgumentParser(
description=(
"Fibonacci Ring with lossless flattened "
"barycentric topology."
)
)
parser.add_argument(
"--steps",
type=int,
default=8,
help="Number of states, starting at K=0.",
)
parser.add_argument(
"--output",
default="fibonacci_ring_flat",
help="Output directory.",
)
parser.add_argument(
"--max-tetrahedra",
type=int,
default=10_000_000,
help="Safety limit before starting a subdivision.",
)
parser.add_argument(
"--min-free-gb",
type=float,
default=2.0,
help="Minimum free disk space required per step.",
)
args = parser.parse_args()
if args.steps < 1:
parser.error("--steps must be positive")
if args.max_tetrahedra < 1:
parser.error("--max-tetrahedra must be positive")
if args.min_free_gb < 0:
parser.error("--min-free-gb cannot be negative")
try:
run_machine(
steps=args.steps,
output=args.output,
max_tetrahedra=args.max_tetrahedra,
min_free_gb=args.min_free_gb,
)
except KeyboardInterrupt:
print(
"\nInterrupted. Previously completed "
"flattened steps remain on disk."
)
sys.exit(130)
except Exception as exc:
print(
f"\nERROR: {type(exc).__name__}: {exc}",
file=sys.stderr,
)
sys.exit(1)
if __name__ == "__main__":
main()
