Research source · python

verify_horizon_capacity.py

site/public/research-artifacts/finite-horizon-scaffold-capacity/verify_horizon_capacity.py

218 lines. Source is displayed for inspection; it is not executed by this page.

File fingerprint

SHA-256: 291deb4b7469fdbf30aac1fae38ad85ef45ad5bbc895298bed72c1ebd4fbac71

#!/usr/bin/env python3
"""Finite audit for readable-cone pullback and its sharp radius."""

from __future__ import annotations

import itertools
import json
from dataclasses import dataclass
from pathlib import Path


MISSING = -1
SELF = -2


@dataclass(frozen=True)
class Scaffold:
    labels: tuple[int, ...]
    edges: tuple[tuple[int, ...], ...]

    @property
    def top(self) -> int:
        return len(self.labels) - 1


def view(scaffold: Scaffold, node: int, radius: int):
    if node == MISSING:
        return ("missing",)
    if radius == 0:
        return ("label", scaffold.labels[node])
    return (
        "node",
        scaffold.labels[node],
        tuple(view(scaffold, target, radius - 1) for target in scaffold.edges[node]),
    )


def words(degree: int, distance: int) -> list[tuple[int, ...]]:
    return [
        word
        for length in range(distance + 1)
        for word in itertools.product(range(degree), repeat=length)
    ]


def follow(scaffold: Scaffold, start: int, descriptor: tuple[int, ...]) -> int:
    node = start
    for port in descriptor:
        if node == MISSING:
            return MISSING
        node = scaffold.edges[node][port]
    return node


def extend(scaffold: Scaffold, label: int, descriptors: tuple[object, ...]) -> Scaffold:
    new = len(scaffold.labels)
    targets: list[int] = []
    for descriptor in descriptors:
        if descriptor == SELF:
            targets.append(new)
        elif descriptor == MISSING:
            targets.append(MISSING)
        else:
            targets.append(follow(scaffold, scaffold.top, descriptor))
    return Scaffold(scaffold.labels + (label,), scaffold.edges + (tuple(targets),))


def scaffolds(node_count: int, degree: int):
    label_choices = itertools.product(range(2), repeat=node_count)
    edge_options = [
        list(itertools.product([MISSING, *range(node)], repeat=degree))
        for node in range(node_count)
    ]
    for labels in label_choices:
        for edges in itertools.product(*edge_options):
            yield Scaffold(tuple(labels), tuple(edges))


def extension_signature(scaffold: Scaffold, degree: int, distance: int, radius: int):
    descriptor_options: list[object] = [MISSING, SELF, *words(degree, distance)]
    return tuple(
        view(extend(scaffold, label, descriptor_tuple), len(scaffold.labels), radius)
        for label in range(2)
        for descriptor_tuple in itertools.product(descriptor_options, repeat=degree)
    )


def audit_pullback() -> dict[str, int]:
    families = [
        # degree, nodes, maximum distance, maximum requested new radius
        (1, 5, 3, 3),
        (2, 4, 2, 2),
    ]
    scaffold_count = 0
    extension_cases = 0
    view_classes = 0
    comparisons = 0

    for degree, nodes, max_distance, max_radius in families:
        material = list(scaffolds(nodes, degree))
        scaffold_count += len(material)
        for distance in range(1, max_distance + 1):
            option_count = 2 + len(words(degree, distance))
            for radius in range(1, max_radius + 1):
                old_radius = distance + radius - 1
                representatives: dict[object, object] = {}
                for scaffold in material:
                    old_view = view(scaffold, scaffold.top, old_radius)
                    signature = extension_signature(scaffold, degree, distance, radius)
                    extension_cases += 2 * (option_count**degree)
                    if old_view in representatives:
                        comparisons += 1
                        assert representatives[old_view] == signature
                    else:
                        representatives[old_view] = signature
                view_classes += len(representatives)

    return {
        "old_scaffolds": scaffold_count,
        "one_step_extension_cases": extension_cases,
        "readable_view_classes": view_classes,
        "same_view_signature_comparisons": comparisons,
    }


def chain(radius: int, target: int) -> Scaffold:
    labels = tuple([target] + [0] * radius)
    # Root is node radius; port zero walks toward node zero.
    edges = tuple((MISSING,) if node == 0 else (node - 1,) for node in range(radius + 1))
    return Scaffold(labels, edges)


def endpoint_label(scaffold: Scaffold, descriptor: tuple[int, ...]) -> int | None:
    node = follow(scaffold, scaffold.top, descriptor)
    return None if node == MISSING else scaffold.labels[node]


def run_sharpness(scaffold: Scaffold, distance: int, horizon: int) -> int:
    zero_path = (0,) * distance
    current = scaffold
    for _ in range(horizon - 1):
        current = extend(current, 0, (zero_path,))
    return int(endpoint_label(current, zero_path) == 1)


def audit_sharpness() -> dict[str, int]:
    witnesses = 0
    equal_smaller_views = 0
    separated_answers = 0
    for distance in range(1, 5):
        for horizon in range(1, 7):
            radius = distance + (horizon - 1) * (distance - 1)
            left = chain(radius, 0)
            right = chain(radius, 1)
            assert view(left, left.top, radius - 1) == view(right, right.top, radius - 1)
            equal_smaller_views += 1
            assert run_sharpness(left, distance, horizon) == 0
            assert run_sharpness(right, distance, horizon) == 1
            separated_answers += 1
            witnesses += 1
    return {
        "sharp_radius_witnesses": witnesses,
        "equal_radius_minus_one_views": equal_smaller_views,
        "separated_horizon_answers": separated_answers,
    }


def view_capacity(degree: int, labels: int, radius: int) -> int:
    # At radius zero, a missing node and an unlabelled node have the same
    # observation; the labels ordinary nodes may expose contribute `labels`.
    total = labels + 1
    for _ in range(radius):
        total = 1 + (labels + 1) * (total**degree)
    return total


def audit_capacity_recurrence() -> dict[str, object]:
    cases = []
    checks = 0
    for degree in range(1, 4):
        for labels in range(1, 4):
            previous = labels + 1
            for radius in range(1, 5):
                current = view_capacity(degree, labels, radius)
                assert current == 1 + (labels + 1) * (previous**degree)
                assert current > previous
                previous = current
                checks += 2
            cases.append(
                {
                    "degree": degree,
                    "working_labels": labels,
                    "view_counts_r0_to_r4": [
                        view_capacity(degree, labels, radius) for radius in range(5)
                    ],
                }
            )
    return {"capacity_recurrence_checks": checks, "capacity_examples": cases}


def main() -> None:
    result = {
        "scope": {
            "pullback": "all binary-labelled backward scaffolds: degree 1 on 5 nodes and degree 2 on 4 nodes",
            "sharpness": "distance 1..4 and continuation horizon 1..6",
        },
        **audit_pullback(),
        **audit_sharpness(),
        **audit_capacity_recurrence(),
        "status": "pass",
    }
    output = Path(__file__).with_name("checker-output.json")
    output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()