Listing: Listing

def choose_match(rows, target, tolerance, key="d_w"):
    if not math.isfinite(target) or target < 0:
        raise ValueError("invalid target")
    if not math.isfinite(tolerance) or tolerance < 0:
        raise ValueError("invalid tolerance")
    eligible = []
    for row in rows:
        value = float(row["budget_acute"][key])
        if not math.isfinite(value) or value < 0:
            raise ValueError("invalid displacement")
        distance = abs(value - target)
        if distance <= tolerance:
            eligible.append((distance, str(row["candidate_id"]), row))
    if not eligible:
        return None
    return min(eligible, key=lambda item: (item[0], item[1]))[2]

def matched_comparisons(sweeps, targets, tolerance, key="d_w"):
    reports = []
    for target in targets:
        matches = {
            family: choose_match(rows, target, tolerance, key)
            for family, rows in sweeps.items()
        }
        reports.append({
            "target": float(target),
            "metric": key,
            "tolerance": tolerance,
            "status": (
                "matched" if all(row is not None
                                 for row in matches.values())
                else "unmatched"
            ),
            "matches": matches,
        })
    return reports

def displacement_efficiency(improvement_fraction, displacement,
                            minimum_displacement=1e-6):
    if minimum_displacement <= 0:
        raise ValueError("positive denominator guard required")
    if not math.isfinite(displacement) or displacement < 0:
        raise ValueError("invalid displacement")
    if displacement < minimum_displacement:
        return None
    return improvement_fraction / displacement

def endpoint_tests():
    # These tests are included but are not reported as executed.
    model = CheckedNetwork(widths=(2, 2, 2), seed=7)
    ledger = MaskLedger(model)
    ledger.prune_to_density(0.5)
    pre = phase_snapshot(model, ledger)

    zero_scores = {
        name: torch.zeros_like(model.get_parameter(name))
        for name in ledger.masks
    }
    before_global = torch.random.get_rng_state().clone()
    regrown = ledger.regrow(0.0, zero_scores, 0.03, seed=99)
    assert all(value == 0 for value in regrown.values())
    assert torch.equal(before_global, torch.random.get_rng_state())
    for name, tensor in pre["weights_biases"].items():
        assert torch.equal(model.state_dict()[name], tensor)
    for name, mask in pre["masks"].items():
        assert torch.equal(ledger.masks[name], mask)

    # Modulation can have zero weight displacement.
    model.a = 0.7
    changed = budget(pre, model, ledger)
    assert changed["d_w"] == 0.0
    assert changed["d_a"] > 0.0

    row = {
        "candidate_id": "candidate-0",
        "budget_acute": {"d_w": 0.02},
    }
    assert choose_match([row], 0.01, 0.001) is None
    assert choose_match([row], 0.02, 0.001) is row

    # Delete all active positions exactly.
    ledger.secondary_prune(1.0)
    assert ledger.density() == 0.0
    for name in ledger.masks:
        assert int(torch.count_nonzero(
            model.get_parameter(name)
        )) == 0

    # Restore all inactive positions, not "at least one".
    scores = {
        name: torch.zeros_like(model.get_parameter(name))
        for name in ledger.masks
    }
    ledger.regrow(1.0, scores, 0.0, seed=1)
    assert ledger.density() == 1.0
    # Structural activity may coexist with numerically zero weights.