API REFERENCE/omnibias-core

omnibias.core.proof.observe

0.5.0rc1Release candidatePython >=3.10Apache-2.0

Coefficient recurrences, activation contracts, jet algebra, and verified arithmetic shared by the numerical backends.

import omnibias.core.proof.observe

Module contract & mathematical conventions

Shared observation, class ranking, and frequency memory.

A tagged Observation is the one payload every condition-sort binder views. Ranking is a total order on *certified* hits. Soft RMSE never outranks an exact identity. ClassMemory / FrequencyGate only propose an order — they never write ~omnibias.core.proof.discovery.ExactCheck.

Shared formulas do not imply universal floating-point identity. Verify dtype, shape, derivative-order support, and the stated numerical assumptions for the path you use.

Module source ↗Source revision 9a2d73df0573Running reference examples →

CLASS

ClassGate

omnibias.core.proof.observe.ClassGateSource ↗

class ClassGate(Protocol)

Proposer over sorts. Must not write ExactCheck.

Fields & methods

proposemethod

propose(self, observation: Observation) -> Sequence[str]

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
observationpositional-or-keyword	ObservationRequired	See the operation contract and implementation below.

Returns: Sequence[str]

Method implementation

    def propose(self, observation: Observation) -> Sequence[str]: ...

View source ↗

Read the implementation

class ClassGate(Protocol):
    """Proposer over sorts. Must not write :class:`ExactCheck`."""

    def propose(self, observation: Observation) -> Sequence[str]: ...

CLASS

ClassHit

omnibias.core.proof.ClassHitSource ↗

class ClassHit

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

One certified (or empirical) class hit. Ranking ignores RMSE when exact exists.

Parameters

Name	Type & default	Contract
sortdataclass field	strRequired	See the operation contract and implementation below.
checkdataclass field	ExactCheckRequired	See the operation contract and implementation below.
tierdataclass field	ClassTierRequired	See the operation contract and implementation below.
tokensdataclass field	intRequired	See the operation contract and implementation below.
degreedataclass field	intRequired	See the operation contract and implementation below.
unique_in_familydataclass field	boolDefault: False	See the operation contract and implementation below.
rmsedataclass field	str | NoneDefault: None	See the operation contract and implementation below.

Regression example · rank two exact by token count

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_rank_two_exact_by_token_count() -> None:
    fat = ClassHit(
        sort="ore",
        check=ExactCheck(ok=True, payload={"tier": "exact"}),
        tier="exact",
        tokens=5,
        degree=0,
    )
    thin = ClassHit(
        sort="jet_monomial",
        check=ExactCheck(ok=True, payload={"tier": "exact"}),
        tier="exact",
        tokens=2,
        degree=4,
    )
    ranked = rank_class_hits([fat, thin])
    assert ranked[0].sort == "jet_monomial"
    assert ranked[0].tokens == 2

python -m pytest packages/omnibias-core/tests/test_observe.py -k test_rank_two_exact_by_token_count -q

Regression example · rank exact beats empirical

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_rank_exact_beats_empirical() -> None:
    exact = ClassHit(
        sort="jet_monomial",
        check=ExactCheck(ok=True, payload={"tier": "exact", "honesty": {}}),
        tier="exact",
        tokens=3,
        degree=2,
    )
    empirical = ClassHit(
        sort="pde_operator",
        check=ExactCheck(ok=False, payload={"mode": "empirical"}),
        tier="empirical",
        tokens=1,
        degree=0,
        rmse="0.001",
    )
    ranked = rank_class_hits([empirical, exact])
    assert ranked[0].sort == "jet_monomial"
    assert all(hit.tier != "empirical" for hit in ranked)

python -m pytest packages/omnibias-core/tests/test_observe.py -k test_rank_exact_beats_empirical -q

Fields & methods

sortfield

sort: str

Declared str field on ClassHit. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

checkfield

check: ExactCheck

Declared ExactCheck field on ClassHit. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

tierfield

tier: ClassTier

Declared ClassTier field on ClassHit. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

tokensfield

tokens: int

Declared int field on ClassHit. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

degreefield

degree: int

Declared int field on ClassHit. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

unique_in_familyfield

unique_in_family: bool = False

Declared bool field on ClassHit. Default expression: False.

View source ↗

rmsefield

rmse: str | None = None

Declared str | None field on ClassHit. Default expression: None.

View source ↗

as_dictmethod

as_dict(self) -> dict[str, Any]

See the class contract above and the exact method implementation below.

Returns: dict[str, Any]

Method implementation

    def as_dict(self) -> dict[str, Any]:
        return {
            "sort": self.sort,
            "tier": self.tier,
            "tokens": self.tokens,
            "degree": self.degree,
            "unique_in_family": self.unique_in_family,
            "rmse": self.rmse,
            "ok": self.check.ok,
        }

View source ↗

Read the implementation

class ClassHit:
    """One certified (or empirical) class hit. Ranking ignores RMSE when exact exists."""

    sort: str
    check: ExactCheck
    tier: ClassTier
    tokens: int
    degree: int
    unique_in_family: bool = False
    rmse: str | None = None

    def as_dict(self) -> dict[str, Any]:
        return {
            "sort": self.sort,
            "tier": self.tier,
            "tokens": self.tokens,
            "degree": self.degree,
            "unique_in_family": self.unique_in_family,
            "rmse": self.rmse,
            "ok": self.check.ok,
        }

Other public import paths

• omnibias.core.proof.ClassHit
• omnibias.core.proof.observe.ClassHit

CLASS

ClassMemory

omnibias.core.proof.ClassMemorySource ↗

class ClassMemory

In-memory features() → counts[sort]. Propose only; never a checker.

Executed regression · class memory json roundtrip

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_json_roundtrip(tmp_path) -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    memory.record(obs, "jet_monomial")
    path = tmp_path / "memory.json"
    memory.save(path)
    loaded = ClassMemory.load(path)
    assert loaded.propose(obs)[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_json_roundtrip" -q

Regression example · class memory reorders after record

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_reorders_after_record() -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    before = memory.propose(obs)
    memory.record(obs, "jet_monomial")
    after = memory.propose(obs)
    assert after[0] == "jet_monomial"
    assert FrequencyGate(memory=memory).propose(obs)[0] == "jet_monomial"
    assert before[0] != "jet_monomial" or after[0] == "jet_monomial"

python -m pytest packages/omnibias-core/tests/test_observe.py -k test_class_memory_reorders_after_record -q

Fields & methods

__init__method

__init__(self) -> None

See the class contract above and the exact method implementation below.

Returns: None

Executed regression · class memory json roundtrip

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_json_roundtrip(tmp_path) -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    memory.record(obs, "jet_monomial")
    path = tmp_path / "memory.json"
    memory.save(path)
    loaded = ClassMemory.load(path)
    assert loaded.propose(obs)[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_json_roundtrip" -q

Method implementation

    def __init__(self) -> None:
        self._counts: dict[tuple[int, ...], dict[str, int]] = {}

View source ↗

recordmethod

record(self, observation: Observation, sort: str) -> None

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
observationpositional-or-keyword	ObservationRequired	See the operation contract and implementation below.
sortpositional-or-keyword	strRequired	See the operation contract and implementation below.

Returns: None

Executed regression · class memory json roundtrip

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_json_roundtrip(tmp_path) -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    memory.record(obs, "jet_monomial")
    path = tmp_path / "memory.json"
    memory.save(path)
    loaded = ClassMemory.load(path)
    assert loaded.propose(obs)[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_json_roundtrip" -q

Method implementation

    def record(self, observation: Observation, sort: str) -> None:
        bucket = self._counts.setdefault(observation.features(), {})
        bucket[sort] = bucket.get(sort, 0) + 1

View source ↗

counts_formethod

counts_for(self, observation: Observation) -> dict[str, int]

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
observationpositional-or-keyword	ObservationRequired	See the operation contract and implementation below.

Returns: dict[str, int]

Method implementation

    def counts_for(self, observation: Observation) -> dict[str, int]:
        return dict(self._counts.get(observation.features(), {}))

View source ↗

proposemethod

propose(self, observation: Observation, sorts: Sequence[str] | None=None) -> tuple[str, ...]

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
observationpositional-or-keyword	ObservationRequired	See the operation contract and implementation below.
sortspositional-or-keyword	Sequence[str] | NoneDefault: None	See the operation contract and implementation below.

Returns: tuple[str, ...]

Executed regression · class memory json roundtrip

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_json_roundtrip(tmp_path) -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    memory.record(obs, "jet_monomial")
    path = tmp_path / "memory.json"
    memory.save(path)
    loaded = ClassMemory.load(path)
    assert loaded.propose(obs)[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_json_roundtrip" -q

Method implementation

    def propose(
        self,
        observation: Observation,
        sorts: Sequence[str] | None = None,
    ) -> tuple[str, ...]:
        names = tuple(sorts) if sorts is not None else _DEFAULT_SORTS
        counts = self._counts.get(observation.features(), {})
        return tuple(sorted(names, key=lambda name: (-counts.get(name, 0), name)))

View source ↗

as_dictmethod

as_dict(self) -> dict[str, Any]

See the class contract above and the exact method implementation below.

Returns: dict[str, Any]

Executed regression · class memory json roundtrip

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_json_roundtrip(tmp_path) -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    memory.record(obs, "jet_monomial")
    path = tmp_path / "memory.json"
    memory.save(path)
    loaded = ClassMemory.load(path)
    assert loaded.propose(obs)[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_json_roundtrip" -q

Method implementation

    def as_dict(self) -> dict[str, Any]:
        return {
            "counts": [
                {"features": list(key), "sorts": dict(value)}
                for key, value in self._counts.items()
            ]
        }

View source ↗

savemethod

save(self, path: str | Path) -> None

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
pathpositional-or-keyword	str | PathRequired	See the operation contract and implementation below.

Returns: None

Executed regression · class memory json roundtrip

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_json_roundtrip(tmp_path) -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    memory.record(obs, "jet_monomial")
    path = tmp_path / "memory.json"
    memory.save(path)
    loaded = ClassMemory.load(path)
    assert loaded.propose(obs)[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_json_roundtrip" -q

Method implementation

    def save(self, path: str | Path) -> None:
        Path(path).write_text(json.dumps(self.as_dict()), encoding="utf-8")

View source ↗

loadmethod

load(cls, path: str | Path) -> ClassMemory

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
pathpositional-or-keyword	str | PathRequired	See the operation contract and implementation below.

Returns: ClassMemory

Executed regression · class memory json roundtrip

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_json_roundtrip(tmp_path) -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    memory.record(obs, "jet_monomial")
    path = tmp_path / "memory.json"
    memory.save(path)
    loaded = ClassMemory.load(path)
    assert loaded.propose(obs)[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_json_roundtrip" -q

Method implementation

    def load(cls, path: str | Path) -> ClassMemory:
        data = json.loads(Path(path).read_text(encoding="utf-8"))
        memory = cls()
        for row in data.get("counts", ()):
            if not isinstance(row, Mapping):
                continue
            key = tuple(int(item) for item in row.get("features", ()))
            sorts = row.get("sorts", {})
            if isinstance(sorts, Mapping):
                memory._counts[key] = {str(name): int(count) for name, count in sorts.items()}
        return memory

View source ↗

Read the implementation

class ClassMemory:
    """In-memory ``features() → counts[sort]``. Propose only; never a checker."""

    def __init__(self) -> None:
        self._counts: dict[tuple[int, ...], dict[str, int]] = {}

    def record(self, observation: Observation, sort: str) -> None:
        bucket = self._counts.setdefault(observation.features(), {})
        bucket[sort] = bucket.get(sort, 0) + 1

    def counts_for(self, observation: Observation) -> dict[str, int]:
        return dict(self._counts.get(observation.features(), {}))

    def propose(
        self,
        observation: Observation,
        sorts: Sequence[str] | None = None,
    ) -> tuple[str, ...]:
        names = tuple(sorts) if sorts is not None else _DEFAULT_SORTS
        counts = self._counts.get(observation.features(), {})
        return tuple(sorted(names, key=lambda name: (-counts.get(name, 0), name)))

    def as_dict(self) -> dict[str, Any]:
        return {
            "counts": [
                {"features": list(key), "sorts": dict(value)}
                for key, value in self._counts.items()
            ]
        }

    def save(self, path: str | Path) -> None:
        Path(path).write_text(json.dumps(self.as_dict()), encoding="utf-8")

    @classmethod
    def load(cls, path: str | Path) -> ClassMemory:
        data = json.loads(Path(path).read_text(encoding="utf-8"))
        memory = cls()
        for row in data.get("counts", ()):
            if not isinstance(row, Mapping):
                continue
            key = tuple(int(item) for item in row.get("features", ()))
            sorts = row.get("sorts", {})
            if isinstance(sorts, Mapping):
                memory._counts[key] = {str(name): int(count) for name, count in sorts.items()}
        return memory

Other public import paths

• omnibias.core.proof.ClassMemory
• omnibias.core.proof.observe.ClassMemory

CLASS

ClassSelection

omnibias.core.proof.ClassSelectionSource ↗

class ClassSelection

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

Ranked class outcome. best is the total-order winner, not visit order.

Parameters

Name	Type & default	Contract
observationdataclass field	ObservationRequired	See the operation contract and implementation below.
hitsdataclass field	tuple[ClassHit, ...]Required	See the operation contract and implementation below.
bestdataclass field	ClassHit | NoneRequired	See the operation contract and implementation below.
honestydataclass field	Mapping[str, bool]Required	See the operation contract and implementation below.
search_incompletedataclass field	boolRequired	See the operation contract and implementation below.
grammar_completedataclass field	boolDefault: False	See the operation contract and implementation below.
resultdataclass field	Any | NoneDefault: None	See the operation contract and implementation below.

Fields & methods

observationfield

observation: Observation

Declared Observation field on ClassSelection. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

hitsfield

hits: tuple[ClassHit, ...]

Declared tuple[ClassHit, ...] field on ClassSelection. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

bestfield

best: ClassHit | None

Declared ClassHit | None field on ClassSelection. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

honestyfield

honesty: Mapping[str, bool]

Declared Mapping[str, bool] field on ClassSelection. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

search_incompletefield

search_incomplete: bool

Declared bool field on ClassSelection. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

grammar_completefield

grammar_complete: bool = False

Declared bool field on ClassSelection. Default expression: False.

View source ↗

resultfield

result: Any | None = None

Declared Any | None field on ClassSelection. Default expression: None.

View source ↗

as_dictmethod

as_dict(self) -> dict[str, Any]

See the class contract above and the exact method implementation below.

Returns: dict[str, Any]

Method implementation

    def as_dict(self) -> dict[str, Any]:
        return {
            "best": None if self.best is None else self.best.as_dict(),
            "hits": [hit.as_dict() for hit in self.hits],
            "honesty": dict(self.honesty),
            "search_incomplete": self.search_incomplete,
            "grammar_complete": self.grammar_complete,
            "observation": self.observation.as_dict(),
        }

View source ↗

Read the implementation

class ClassSelection:
    """Ranked class outcome. ``best`` is the total-order winner, not visit order."""

    observation: Observation
    hits: tuple[ClassHit, ...]
    best: ClassHit | None
    honesty: Mapping[str, bool]
    search_incomplete: bool
    grammar_complete: bool = False
    result: Any | None = None

    def as_dict(self) -> dict[str, Any]:
        return {
            "best": None if self.best is None else self.best.as_dict(),
            "hits": [hit.as_dict() for hit in self.hits],
            "honesty": dict(self.honesty),
            "search_incomplete": self.search_incomplete,
            "grammar_complete": self.grammar_complete,
            "observation": self.observation.as_dict(),
        }

Other public import paths

• omnibias.core.proof.ClassSelection
• omnibias.core.proof.observe.ClassSelection

CONSTANT

ClassTier

omnibias.core.proof.observe.ClassTierSource ↗

ClassTier = Literal['exact', 'enclosure', 'empirical']

This exported constant is defined by the expression above. Its exact value and annotation come from the pinned source; use it with the module contract.

Read the implementation

ClassTier = Literal["exact", "enclosure", "empirical"]

CONSTANT

FEATURE_DIM

omnibias.core.proof.FEATURE_DIMSource ↗

FEATURE_DIM = 16

This exported constant is defined by the expression above. Its exact value and annotation come from the pinned source; use it with the module contract.

Regression example · observation json and features

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_observation_json_and_features() -> None:
    obs = Observation(tag="fibonacci", sequence=("0", "1", "1", "2"))
    features = obs.features()
    assert len(features) == FEATURE_DIM
    assert features[0] == 1
    assert features[1] == 0
    assert features[5] == 4
    payload = obs.as_dict()
    restored = Observation.from_dict(payload)
    assert restored.tag == "fibonacci"
    assert restored.sequence == obs.sequence
    assert restored.sample_x == ()
    assert restored.poly_constraints == ()
    assert restored.features() == features

python -m pytest packages/omnibias-core/tests/test_observe.py -k test_observation_json_and_features -q

Read the implementation

FEATURE_DIM = 16

Other public import paths

• omnibias.core.proof.FEATURE_DIM
• omnibias.core.proof.observe.FEATURE_DIM

CLASS

FrequencyGate

omnibias.core.proof.FrequencyGateSource ↗

class FrequencyGate

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

Frequency proposer. Thin wrapper around ClassMemory.propose.

Parameters

Name	Type & default	Contract
memorydataclass field	ClassMemoryDefault: field(default_factory=ClassMemory)	See the operation contract and implementation below.

Executed regression · class memory reorders after record

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_reorders_after_record() -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    before = memory.propose(obs)
    memory.record(obs, "jet_monomial")
    after = memory.propose(obs)
    assert after[0] == "jet_monomial"
    assert FrequencyGate(memory=memory).propose(obs)[0] == "jet_monomial"
    assert before[0] != "jet_monomial" or after[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_reorders_after_record" -q

Fields & methods

memoryfield

memory: ClassMemory = field(default_factory=ClassMemory)

Declared ClassMemory field on FrequencyGate. Default expression: field(default_factory=ClassMemory).

View source ↗

proposemethod

propose(self, observation: Observation) -> tuple[str, ...]

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
observationpositional-or-keyword	ObservationRequired	See the operation contract and implementation below.

Returns: tuple[str, ...]

Executed regression · class memory reorders after record

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_reorders_after_record() -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    before = memory.propose(obs)
    memory.record(obs, "jet_monomial")
    after = memory.propose(obs)
    assert after[0] == "jet_monomial"
    assert FrequencyGate(memory=memory).propose(obs)[0] == "jet_monomial"
    assert before[0] != "jet_monomial" or after[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_reorders_after_record" -q

Method implementation

    def propose(self, observation: Observation) -> tuple[str, ...]:
        return self.memory.propose(observation)

View source ↗

Read the implementation

class FrequencyGate:
    """Frequency proposer. Thin wrapper around :class:`ClassMemory.propose`."""

    memory: ClassMemory = field(default_factory=ClassMemory)

    def propose(self, observation: Observation) -> tuple[str, ...]:
        return self.memory.propose(observation)

Other public import paths

• omnibias.core.proof.FrequencyGate
• omnibias.core.proof.observe.FrequencyGate

CLASS

LinearSpanFamily

omnibias.core.proof.LinearSpanFamilySource ↗

class LinearSpanFamily

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

Bitmask family over an exact-Q design. Nullity-one is the accept gate.

Parameters

Name	Type & default	Contract
designdataclass field	tuple[tuple[int, ...], ...]Required	See the operation contract and implementation below.
term_namesdataclass field	tuple[str, ...]Required	See the operation contract and implementation below.
sortdataclass field	strDefault: 'jet_monomial'	See the operation contract and implementation below.
targetdataclass field	tuple[int, ...] | NoneDefault: None	See the operation contract and implementation below.
namedataclass field	strDefault: 'linear_span'	See the operation contract and implementation below.
completedataclass field	boolDefault: False	See the operation contract and implementation below.
empty_miss_detaildataclass field	strDefault: 'no witness in enumerated grammar'	See the operation contract and implementation below.
statementdataclass field	StatementDefault: field(default_factory=lambda: Statement(name='linear_span', obligation='a unique linear relation in the supplied exact-Q columns', parent='observation class loop', parent_status='already_true'))	See the operation contract and implementation below.

Executed regression · linear span nullity one

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_linear_span_nullity_one() -> None:
    family = LinearSpanFamily(
        design=((1, 1), (2, 2)),
        term_names=("a", "b"),
        sort="jet_monomial",
    )
    both = family.check(3)
    assert both is not None
    assert both.ok is True
    assert both.payload["tier"] == "exact"
    single = family.check(1)
    assert single is not None
    assert single.ok is False

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_linear_span_nullity_one" -q

Regression example · linear span product growth

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_linear_span_product_growth() -> None:
    family = LinearSpanFamily(
        design=((1, 2, 2), (4, 4, 2), (9, 6, 2), (16, 8, 2)),
        term_names=("y", "yp", "ypp"),
        sort="jet_monomial",
    )
    assert family.check(7) is None or family.check(7).ok is False
    grown = family.grown_product_check()
    assert grown is not None
    assert grown.ok is True
    assert grown.payload["grown"] is True

python -m pytest packages/omnibias-core/tests/test_observe.py -k test_linear_span_product_growth -q

Fields & methods

designfield

design: tuple[tuple[int, ...], ...]

Declared tuple[tuple[int, ...], ...] field on LinearSpanFamily. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

term_namesfield

term_names: tuple[str, ...]

Declared tuple[str, ...] field on LinearSpanFamily. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

sortfield

sort: str = 'jet_monomial'

Declared str field on LinearSpanFamily. Default expression: 'jet_monomial'.

View source ↗

targetfield

target: tuple[int, ...] | None = None

Declared tuple[int, ...] | None field on LinearSpanFamily. Default expression: None.

View source ↗

namefield

name: str = 'linear_span'

Declared str field on LinearSpanFamily. Default expression: 'linear_span'.

View source ↗

completefield

complete: bool = False

Declared bool field on LinearSpanFamily. Default expression: False.

View source ↗

empty_miss_detailfield

empty_miss_detail: str = 'no witness in enumerated grammar'

Declared str field on LinearSpanFamily. Default expression: 'no witness in enumerated grammar'.

View source ↗

statementfield

statement: Statement = field(default_factory=lambda: Statement(name='linear_span', obligation='a unique linear relation in the supplied exact-Q columns', parent='observation class loop', parent_status='already_true'))

Declared Statement field on LinearSpanFamily. Default expression: field(default_factory=lambda: Statement(name='linear_span', obligation='a unique linear relation in the supplied exact-Q columns', parent='observation class loop', parent_status='already_true')).

View source ↗

from_observationmethod

from_observation(cls, observation: Observation, *, sort: str) -> LinearSpanFamily | None

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
observationpositional-or-keyword	ObservationRequired	See the operation contract and implementation below.
sortkeyword-only	strRequired	See the operation contract and implementation below.

Returns: LinearSpanFamily | None

Method implementation

    def from_observation(
        cls,
        observation: Observation,
        *,
        sort: str,
    ) -> LinearSpanFamily | None:
        if observation.jet_rows and observation.jet_names:
            rows = tuple(
                tuple(as_fraction(Fraction(item)) for item in row)
                for row in observation.jet_rows
            )
            names = observation.jet_names
            target_frac: tuple[Fraction, ...] | None = None
        elif observation.design and observation.term_names:
            rows = tuple(
                tuple(as_fraction(Fraction(item)) for item in row)
                for row in observation.design
            )
            names = observation.term_names
            target_frac = (
                tuple(as_fraction(Fraction(item)) for item in observation.target)
                if observation.target
                else None
            )
        else:
            return None
        if target_frac is not None:
            augmented = tuple(row + (value,) for row, value in zip(rows, target_frac, strict=True))
            ints = fractions_to_ints(augmented)
            design = tuple(row[:-1] for row in ints)
            target = tuple(row[-1] for row in ints)
        else:
            design = fractions_to_ints(rows)
            target = None
        return cls(design=design, term_names=names, sort=sort, target=target)

View source ↗

cardinalitymethod

cardinality(self) -> int

See the class contract above and the exact method implementation below.

Returns: int

Executed regression · linear span nullity one

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_linear_span_nullity_one() -> None:
    family = LinearSpanFamily(
        design=((1, 1), (2, 2)),
        term_names=("a", "b"),
        sort="jet_monomial",
    )
    both = family.check(3)
    assert both is not None
    assert both.ok is True
    assert both.payload["tier"] == "exact"
    single = family.check(1)
    assert single is not None
    assert single.ok is False

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_linear_span_nullity_one" -q

Method implementation

    def cardinality(self) -> int:
        width = len(self.term_names)
        return 0 if width == 0 else (1 << width) - 1

View source ↗

originmethod

origin(self) -> int

See the class contract above and the exact method implementation below.

Returns: int

Executed regression · select class grows linear span products

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_select_class_grows_linear_span_products() -> None:
    saved = _isolated()
    try:
        from omnibias.core.proof.condition import register_condition_sort

        family = LinearSpanFamily(
            design=((1, 2, 2), (4, 4, 2), (9, 6, 2), (16, 8, 2)),
            term_names=("y", "yp", "ypp"),
            sort="jet_monomial",
        )
        register_condition_sort("jet_monomial", lambda obs: family)
        obs = Observation(
            jet_names=("y", "yp", "ypp"),
            jet_rows=(("1", "2", "2"), ("4", "4", "2"), ("9", "6", "2"), ("16", "8", "2")),
        )
        selection = select_class(obs, sorts=("jet_monomial",), grow=True, inner_budget=2)
        assert selection.best is not None
        assert selection.best.tier == "exact"
        assert selection.grammar_complete is False
    finally:
        _restore(saved)

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_select_class_grows_linear_span_products" -q

Method implementation

    def origin(self) -> int:
        names = self.term_names
        if "yp" in names:
            return 1 << names.index("yp")
        if "u_xx" in names:
            return 1 << names.index("u_xx")
        if "rho_t" in names and "j_x" in names:
            return (1 << names.index("rho_t")) | (1 << names.index("j_x"))
        width = len(names)
        return (1 << width) - 1 if width else 1

View source ↗

neighborsmethod

neighbors(self, candidate: Candidate) -> Sequence[int]

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
candidatepositional-or-keyword	CandidateRequired	See the operation contract and implementation below.

Returns: Sequence[int]

Executed regression · select class grows linear span products

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_select_class_grows_linear_span_products() -> None:
    saved = _isolated()
    try:
        from omnibias.core.proof.condition import register_condition_sort

        family = LinearSpanFamily(
            design=((1, 2, 2), (4, 4, 2), (9, 6, 2), (16, 8, 2)),
            term_names=("y", "yp", "ypp"),
            sort="jet_monomial",
        )
        register_condition_sort("jet_monomial", lambda obs: family)
        obs = Observation(
            jet_names=("y", "yp", "ypp"),
            jet_rows=(("1", "2", "2"), ("4", "4", "2"), ("9", "6", "2"), ("16", "8", "2")),
        )
        selection = select_class(obs, sorts=("jet_monomial",), grow=True, inner_budget=2)
        assert selection.best is not None
        assert selection.best.tier == "exact"
        assert selection.grammar_complete is False
    finally:
        _restore(saved)

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_select_class_grows_linear_span_products" -q

Method implementation

    def neighbors(self, candidate: Candidate) -> Sequence[int]:
        if not isinstance(candidate, int):
            return ()
        width = len(self.term_names)
        limit = self.cardinality()
        out: list[int] = []
        for bit in range(width):
            nxt = int(candidate) ^ (1 << bit)
            if 0 < nxt <= limit:
                out.append(nxt)
        return out

View source ↗

scoremethod

score(self, candidate: Candidate) -> int

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
candidatepositional-or-keyword	CandidateRequired	See the operation contract and implementation below.

Returns: int

Executed regression · select class grows linear span products

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_select_class_grows_linear_span_products() -> None:
    saved = _isolated()
    try:
        from omnibias.core.proof.condition import register_condition_sort

        family = LinearSpanFamily(
            design=((1, 2, 2), (4, 4, 2), (9, 6, 2), (16, 8, 2)),
            term_names=("y", "yp", "ypp"),
            sort="jet_monomial",
        )
        register_condition_sort("jet_monomial", lambda obs: family)
        obs = Observation(
            jet_names=("y", "yp", "ypp"),
            jet_rows=(("1", "2", "2"), ("4", "4", "2"), ("9", "6", "2"), ("16", "8", "2")),
        )
        selection = select_class(obs, sorts=("jet_monomial",), grow=True, inner_budget=2)
        assert selection.best is not None
        assert selection.best.tier == "exact"
        assert selection.grammar_complete is False
    finally:
        _restore(saved)

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_select_class_grows_linear_span_products" -q

Method implementation

    def score(self, candidate: Candidate) -> int:
        if not isinstance(candidate, int):
            return 0
        value = int(candidate)
        bonus = 0
        names = self.term_names
        if "yp" in names and value & (1 << names.index("yp")):
            bonus += 5
        if "u_xx" in names and value & (1 << names.index("u_xx")):
            bonus += 3
        return bonus - value.bit_count()

View source ↗

checkmethod

check(self, candidate: Candidate) -> ExactCheck | None

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
candidatepositional-or-keyword	CandidateRequired	See the operation contract and implementation below.

Returns: ExactCheck | None

Executed regression · linear span nullity one

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_linear_span_nullity_one() -> None:
    family = LinearSpanFamily(
        design=((1, 1), (2, 2)),
        term_names=("a", "b"),
        sort="jet_monomial",
    )
    both = family.check(3)
    assert both is not None
    assert both.ok is True
    assert both.payload["tier"] == "exact"
    single = family.check(1)
    assert single is not None
    assert single.ok is False

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_linear_span_nullity_one" -q

Method implementation

    def check(self, candidate: Candidate) -> ExactCheck | None:
        if not isinstance(candidate, int):
            return None
        mask = int(candidate)
        width = len(self.term_names)
        if width == 0 or not 0 < mask <= self.cardinality():
            return None
        cols = [index for index in range(width) if mask & (1 << index)]
        if not cols:
            return None
        sub = [[row[index] for index in cols] for row in self.design]
        if self.target is not None and any(value != 0 for value in self.target):
            matrix = [list(row) + [-value] for row, value in zip(sub, self.target, strict=True)]
        else:
            matrix = [list(row) for row in sub]
        null = integer_null_space(matrix)
        ok = len(null) == 1
        from omnibias.core.proof.condition import condition_honesty

        pretty = None
        coeffs: tuple[str, ...] = ()
        if ok:
            vector = null[0]
            pieces: list[str] = []
            coeff_list: list[str] = []
            design_coefs = vector[: len(cols)]
            for name, coef in zip((self.term_names[i] for i in cols), design_coefs, strict=True):
                coeff_list.append(str(coef))
                if coef == 0:
                    continue
                pieces.append(f"({coef})*{name}")
            if self.target is not None and len(vector) > len(cols):
                coeff_list.append(str(vector[-1]))
            pretty = " + ".join(pieces) + " = 0" if pieces else "0 = 0"
            coeffs = tuple(coeff_list)
        return ExactCheck(
            ok=ok,
            payload={
                "sort": self.sort,
                "mask": mask,
                "tier": "exact",
                "tokens": len(cols),
                "annihilator": pretty,
                "equation": (
                    None
                    if pretty is None
                    else {
                        "kind": "polynomial_identity",
                        "pretty": pretty,
                        "coefficients": list(coeffs),
                    }
                ),
                "honesty": condition_honesty(discovered=ok),
                "no_condition_exists_claim": False,
            },
        )

View source ↗

grown_product_checkmethod

grown_product_check(self) -> ExactCheck | None

One-step column products. Used by select_class(..., grow=True) on a miss.

Returns: ExactCheck | None

Executed regression · linear span product growth

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_linear_span_product_growth() -> None:
    family = LinearSpanFamily(
        design=((1, 2, 2), (4, 4, 2), (9, 6, 2), (16, 8, 2)),
        term_names=("y", "yp", "ypp"),
        sort="jet_monomial",
    )
    assert family.check(7) is None or family.check(7).ok is False
    grown = family.grown_product_check()
    assert grown is not None
    assert grown.ok is True
    assert grown.payload["grown"] is True

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_linear_span_product_growth" -q

Method implementation

    def grown_product_check(self) -> ExactCheck | None:
        """One-step column products. Used by ``select_class(..., grow=True)`` on a miss."""

        from omnibias.core.proof.condition import condition_honesty

        width = len(self.term_names)
        if width == 0:
            return None
        for left in range(width):
            for right in range(left, width):
                product = tuple(row[left] * row[right] for row in self.design)
                product_name = f"{self.term_names[left]}*{self.term_names[right]}"
                for other in range(width):
                    if self.term_names[other] == product_name:
                        continue
                    matrix = [
                        [row[other], product[index]]
                        for index, row in enumerate(self.design)
                    ]
                    if self.target is not None and any(value != 0 for value in self.target):
                        matrix = [
                            list(row) + [-value]
                            for row, value in zip(matrix, self.target, strict=True)
                        ]
                    null = integer_null_space(matrix)
                    if len(null) != 1:
                        continue
                    vector = null[0]
                    if self.target is not None and any(value != 0 for value in self.target):
                        if len(vector) < 3 or vector[-1] == 0:
                            continue
                    pretty = (
                        f"({vector[0]})*{self.term_names[other]} + ({vector[1]})*{product_name} = 0"
                    )
                    return ExactCheck(
                        ok=True,
                        payload={
                            "sort": self.sort,
                            "tier": "exact",
                            "tokens": 2,
                            "grown": True,
                            "annihilator": pretty,
                            "equation": {
                                "kind": "polynomial_identity",
                                "pretty": pretty,
                                "coefficients": [str(vector[0]), str(vector[1])],
                            },
                            "honesty": condition_honesty(discovered=True),
                            "no_condition_exists_claim": False,
                        },
                    )
        return None

View source ↗

Read the implementation

class LinearSpanFamily:
    """Bitmask family over an exact-``Q`` design. Nullity-one is the accept gate."""

    design: tuple[tuple[int, ...], ...]
    term_names: tuple[str, ...]
    sort: str = "jet_monomial"
    target: tuple[int, ...] | None = None
    name: str = "linear_span"
    complete: bool = False
    empty_miss_detail: str = "no witness in enumerated grammar"
    statement: Statement = field(
        default_factory=lambda: Statement(
            name="linear_span",
            obligation="a unique linear relation in the supplied exact-Q columns",
            parent="observation class loop",
            parent_status="already_true",
        )
    )

    @classmethod
    def from_observation(
        cls,
        observation: Observation,
        *,
        sort: str,
    ) -> LinearSpanFamily | None:
        if observation.jet_rows and observation.jet_names:
            rows = tuple(
                tuple(as_fraction(Fraction(item)) for item in row)
                for row in observation.jet_rows
            )
            names = observation.jet_names
            target_frac: tuple[Fraction, ...] | None = None
        elif observation.design and observation.term_names:
            rows = tuple(
                tuple(as_fraction(Fraction(item)) for item in row)
                for row in observation.design
            )
            names = observation.term_names
            target_frac = (
                tuple(as_fraction(Fraction(item)) for item in observation.target)
                if observation.target
                else None
            )
        else:
            return None
        if target_frac is not None:
            augmented = tuple(row + (value,) for row, value in zip(rows, target_frac, strict=True))
            ints = fractions_to_ints(augmented)
            design = tuple(row[:-1] for row in ints)
            target = tuple(row[-1] for row in ints)
        else:
            design = fractions_to_ints(rows)
            target = None
        return cls(design=design, term_names=names, sort=sort, target=target)

    def cardinality(self) -> int:
        width = len(self.term_names)
        return 0 if width == 0 else (1 << width) - 1

    def origin(self) -> int:
        names = self.term_names
        if "yp" in names:
            return 1 << names.index("yp")
        if "u_xx" in names:
            return 1 << names.index("u_xx")
        if "rho_t" in names and "j_x" in names:
            return (1 << names.index("rho_t")) | (1 << names.index("j_x"))
        width = len(names)
        return (1 << width) - 1 if width else 1

    def neighbors(self, candidate: Candidate) -> Sequence[int]:
        if not isinstance(candidate, int):
            return ()
        width = len(self.term_names)
        limit = self.cardinality()
        out: list[int] = []
        for bit in range(width):
            nxt = int(candidate) ^ (1 << bit)
            if 0 < nxt <= limit:
                out.append(nxt)
        return out

    def score(self, candidate: Candidate) -> int:
        if not isinstance(candidate, int):
            return 0
        value = int(candidate)
        bonus = 0
        names = self.term_names
        if "yp" in names and value & (1 << names.index("yp")):
            bonus += 5
        if "u_xx" in names and value & (1 << names.index("u_xx")):
            bonus += 3
        return bonus - value.bit_count()

    def check(self, candidate: Candidate) -> ExactCheck | None:
        if not isinstance(candidate, int):
            return None
        mask = int(candidate)
        width = len(self.term_names)
        if width == 0 or not 0 < mask <= self.cardinality():
            return None
        cols = [index for index in range(width) if mask & (1 << index)]
        if not cols:
            return None
        sub = [[row[index] for index in cols] for row in self.design]
        if self.target is not None and any(value != 0 for value in self.target):
            matrix = [list(row) + [-value] for row, value in zip(sub, self.target, strict=True)]
        else:
            matrix = [list(row) for row in sub]
        null = integer_null_space(matrix)
        ok = len(null) == 1
        from omnibias.core.proof.condition import condition_honesty

        pretty = None
        coeffs: tuple[str, ...] = ()
        if ok:
            vector = null[0]
            pieces: list[str] = []
            coeff_list: list[str] = []
            design_coefs = vector[: len(cols)]
            for name, coef in zip((self.term_names[i] for i in cols), design_coefs, strict=True):
                coeff_list.append(str(coef))
                if coef == 0:
                    continue
                pieces.append(f"({coef})*{name}")
            if self.target is not None and len(vector) > len(cols):
                coeff_list.append(str(vector[-1]))
            pretty = " + ".join(pieces) + " = 0" if pieces else "0 = 0"
            coeffs = tuple(coeff_list)
        return ExactCheck(
            ok=ok,
            payload={
                "sort": self.sort,
                "mask": mask,
                "tier": "exact",
                "tokens": len(cols),
                "annihilator": pretty,
                "equation": (
                    None
                    if pretty is None
                    else {
                        "kind": "polynomial_identity",
                        "pretty": pretty,
                        "coefficients": list(coeffs),
                    }
                ),
                "honesty": condition_honesty(discovered=ok),
                "no_condition_exists_claim": False,
            },
        )

    def grown_product_check(self) -> ExactCheck | None:
        """One-step column products. Used by ``select_class(..., grow=True)`` on a miss."""

        from omnibias.core.proof.condition import condition_honesty

        width = len(self.term_names)
        if width == 0:
            return None
        for left in range(width):
            for right in range(left, width):
                product = tuple(row[left] * row[right] for row in self.design)
                product_name = f"{self.term_names[left]}*{self.term_names[right]}"
                for other in range(width):
                    if self.term_names[other] == product_name:
                        continue
                    matrix = [
                        [row[other], product[index]]
                        for index, row in enumerate(self.design)
                    ]
                    if self.target is not None and any(value != 0 for value in self.target):
                        matrix = [
                            list(row) + [-value]
                            for row, value in zip(matrix, self.target, strict=True)
                        ]
                    null = integer_null_space(matrix)
                    if len(null) != 1:
                        continue
                    vector = null[0]
                    if self.target is not None and any(value != 0 for value in self.target):
                        if len(vector) < 3 or vector[-1] == 0:
                            continue
                    pretty = (
                        f"({vector[0]})*{self.term_names[other]} + ({vector[1]})*{product_name} = 0"
                    )
                    return ExactCheck(
                        ok=True,
                        payload={
                            "sort": self.sort,
                            "tier": "exact",
                            "tokens": 2,
                            "grown": True,
                            "annihilator": pretty,
                            "equation": {
                                "kind": "polynomial_identity",
                                "pretty": pretty,
                                "coefficients": [str(vector[0]), str(vector[1])],
                            },
                            "honesty": condition_honesty(discovered=True),
                            "no_condition_exists_claim": False,
                        },
                    )
        return None

Related APIsomnibias.core.proof.ExactCheck↗omnibias.core.proof.Statement↗omnibias.core.proof.as_fraction↗omnibias.core.proof.integer_null_space↗omnibias.core.proof.observe.fractions_to_ints↗

Other public import paths

• omnibias.core.proof.LinearSpanFamily
• omnibias.core.proof.observe.LinearSpanFamily

CLASS

Observation

omnibias.core.proof.ObservationSource ↗

class Observation

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

Tagged, JSON-able payload. Values are fraction strings or ints. No numpy.

Parameters

Name	Type & default	Contract
tagdataclass field	strDefault: ''	See the operation contract and implementation below.
sequencedataclass field	tuple[str, ...]Default: ()	See the operation contract and implementation below.
jet_namesdataclass field	tuple[str, ...]Default: ()	See the operation contract and implementation below.
jet_rowsdataclass field	tuple[tuple[str, ...], ...]Default: ()	See the operation contract and implementation below.
designdataclass field	tuple[tuple[str, ...], ...]Default: ()	See the operation contract and implementation below.
targetdataclass field	tuple[str, ...]Default: ()	See the operation contract and implementation below.
term_namesdataclass field	tuple[str, ...]Default: ()	See the operation contract and implementation below.
graph_ndataclass field	intDefault: 0	See the operation contract and implementation below.
graph_edgesdataclass field	tuple[tuple[int, int], ...]Default: ()	See the operation contract and implementation below.
poly_n_varsdataclass field	intDefault: 0	See the operation contract and implementation below.
poly_termsdataclass field	PolyTermsDefault: ()	See the operation contract and implementation below.
sample_xdataclass field	tuple[str, ...]Default: ()	See the operation contract and implementation below.
poly_constraintsdataclass field	tuple[PolyTerms, ...]Default: ()	See the operation contract and implementation below.
extradataclass field	tuple[tuple[str, str], ...]Default: ()	See the operation contract and implementation below.

Executed regression · class memory json roundtrip

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_json_roundtrip(tmp_path) -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    memory.record(obs, "jet_monomial")
    path = tmp_path / "memory.json"
    memory.save(path)
    loaded = ClassMemory.load(path)
    assert loaded.propose(obs)[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_json_roundtrip" -q

Regression example · class memory reorders after record

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_reorders_after_record() -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    before = memory.propose(obs)
    memory.record(obs, "jet_monomial")
    after = memory.propose(obs)
    assert after[0] == "jet_monomial"
    assert FrequencyGate(memory=memory).propose(obs)[0] == "jet_monomial"
    assert before[0] != "jet_monomial" or after[0] == "jet_monomial"

python -m pytest packages/omnibias-core/tests/test_observe.py -k test_class_memory_reorders_after_record -q

Fields & methods

tagfield

tag: str = ''

Declared str field on Observation. Default expression: ''.

View source ↗

sequencefield

sequence: tuple[str, ...] = ()

Declared tuple[str, ...] field on Observation. Default expression: ().

Regression example · observation json and features

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_observation_json_and_features() -> None:
    obs = Observation(tag="fibonacci", sequence=("0", "1", "1", "2"))
    features = obs.features()
    assert len(features) == FEATURE_DIM
    assert features[0] == 1
    assert features[1] == 0
    assert features[5] == 4
    payload = obs.as_dict()
    restored = Observation.from_dict(payload)
    assert restored.tag == "fibonacci"
    assert restored.sequence == obs.sequence
    assert restored.sample_x == ()
    assert restored.poly_constraints == ()
    assert restored.features() == features

python -m pytest packages/omnibias-core/tests/test_observe.py -k test_observation_json_and_features -q

View source ↗

jet_namesfield

jet_names: tuple[str, ...] = ()

Declared tuple[str, ...] field on Observation. Default expression: ().

View source ↗

jet_rowsfield

jet_rows: tuple[tuple[str, ...], ...] = ()

Declared tuple[tuple[str, ...], ...] field on Observation. Default expression: ().

View source ↗

designfield

design: tuple[tuple[str, ...], ...] = ()

Declared tuple[tuple[str, ...], ...] field on Observation. Default expression: ().

View source ↗

targetfield

target: tuple[str, ...] = ()

Declared tuple[str, ...] field on Observation. Default expression: ().

View source ↗

term_namesfield

term_names: tuple[str, ...] = ()

Declared tuple[str, ...] field on Observation. Default expression: ().

View source ↗

graph_nfield

graph_n: int = 0

Declared int field on Observation. Default expression: 0.

View source ↗

graph_edgesfield

graph_edges: tuple[tuple[int, int], ...] = ()

Declared tuple[tuple[int, int], ...] field on Observation. Default expression: ().

View source ↗

poly_n_varsfield

poly_n_vars: int = 0

Declared int field on Observation. Default expression: 0.

View source ↗

poly_termsfield

poly_terms: PolyTerms = ()

Declared PolyTerms field on Observation. Default expression: ().

View source ↗

sample_xfield

sample_x: tuple[str, ...] = ()

Declared tuple[str, ...] field on Observation. Default expression: ().

View source ↗

poly_constraintsfield

poly_constraints: tuple[PolyTerms, ...] = ()

Declared tuple[PolyTerms, ...] field on Observation. Default expression: ().

View source ↗

extrafield

extra: tuple[tuple[str, str], ...] = ()

Declared tuple[tuple[str, str], ...] field on Observation. Default expression: ().

View source ↗

extra_mapmethod

extra_map(self) -> dict[str, str]

See the class contract above and the exact method implementation below.

Returns: dict[str, str]

Method implementation

    def extra_map(self) -> dict[str, str]:
        return dict(self.extra)

View source ↗

featuresmethod

features(self) -> tuple[int, ...]

Length-FEATURE_DIM int tuple for memory / gate keys.

Returns: tuple[int, ...]

Executed regression · class memory json roundtrip

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_class_memory_json_roundtrip(tmp_path) -> None:
    memory = ClassMemory()
    obs = Observation(tag="tanh", jet_names=("y",), jet_rows=(("1",),))
    memory.record(obs, "jet_monomial")
    path = tmp_path / "memory.json"
    memory.save(path)
    loaded = ClassMemory.load(path)
    assert loaded.propose(obs)[0] == "jet_monomial"

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_class_memory_json_roundtrip" -q

Method implementation

    def features(self) -> tuple[int, ...]:
        """Length-:data:`FEATURE_DIM` int tuple for memory / gate keys."""

        names = self.term_names + self.jet_names
        has_second = 1 if any(name in _SECOND_DERIV for name in names) else 0
        has_cont = 1 if ("rho_t" in names and "j_x" in names) else 0
        return (
            1 if self.sequence else 0,
            1 if self.jet_rows or self.jet_names else 0,
            1 if self.design or self.target else 0,
            1 if self.graph_edges or self.graph_n else 0,
            1 if self.poly_terms or self.poly_n_vars else 0,
            len(self.sequence),
            len(self.jet_rows),
            len(self.design),
            int(self.graph_n),
            int(self.poly_n_vars),
            has_second,
            has_cont,
            _sign_change_count(self),
            _poly_degree_estimate(self.poly_terms),
            len(self.poly_constraints),
            1 if self.sample_x else 0,
        )

View source ↗

as_dictmethod

as_dict(self) -> dict[str, Any]

See the class contract above and the exact method implementation below.

Returns: dict[str, Any]

Executed regression · observation json and features

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_observation_json_and_features() -> None:
    obs = Observation(tag="fibonacci", sequence=("0", "1", "1", "2"))
    features = obs.features()
    assert len(features) == FEATURE_DIM
    assert features[0] == 1
    assert features[1] == 0
    assert features[5] == 4
    payload = obs.as_dict()
    restored = Observation.from_dict(payload)
    assert restored.tag == "fibonacci"
    assert restored.sequence == obs.sequence
    assert restored.sample_x == ()
    assert restored.poly_constraints == ()
    assert restored.features() == features

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_observation_json_and_features" -q

Method implementation

    def as_dict(self) -> dict[str, Any]:
        return {
            "tag": self.tag,
            "sequence": list(self.sequence),
            "jet_names": list(self.jet_names),
            "jet_rows": [list(row) for row in self.jet_rows],
            "design": [list(row) for row in self.design],
            "target": list(self.target),
            "term_names": list(self.term_names),
            "graph_n": self.graph_n,
            "graph_edges": [list(edge) for edge in self.graph_edges],
            "poly_n_vars": self.poly_n_vars,
            "poly_terms": [[list(exp), coeff] for exp, coeff in self.poly_terms],
            "sample_x": list(self.sample_x),
            "poly_constraints": [
                [[list(exp), coeff] for exp, coeff in poly]
                for poly in self.poly_constraints
            ],
            "extra": dict(self.extra),
            "features": list(self.features()),
        }

View source ↗

from_dictmethod

from_dict(cls, data: Mapping[str, Any]) -> Observation

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
datapositional-or-keyword	Mapping[str, Any]Required	See the operation contract and implementation below.

Returns: Observation

Executed regression · observation json and features

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_observation_json_and_features() -> None:
    obs = Observation(tag="fibonacci", sequence=("0", "1", "1", "2"))
    features = obs.features()
    assert len(features) == FEATURE_DIM
    assert features[0] == 1
    assert features[1] == 0
    assert features[5] == 4
    payload = obs.as_dict()
    restored = Observation.from_dict(payload)
    assert restored.tag == "fibonacci"
    assert restored.sequence == obs.sequence
    assert restored.sample_x == ()
    assert restored.poly_constraints == ()
    assert restored.features() == features

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_observation_json_and_features" -q

Method implementation

    def from_dict(cls, data: Mapping[str, Any]) -> Observation:
        extra_raw = data.get("extra", ())
        if isinstance(extra_raw, Mapping):
            extra = tuple((str(key), str(value)) for key, value in extra_raw.items())
        else:
            extra = tuple((str(key), str(value)) for key, value in extra_raw)
        constraints_raw = data.get("poly_constraints", ())
        return cls(
            tag=str(data.get("tag", "")),
            sequence=tuple(str(item) for item in data.get("sequence", ())),
            jet_names=tuple(str(item) for item in data.get("jet_names", ())),
            jet_rows=tuple(
                tuple(str(item) for item in row) for row in data.get("jet_rows", ())
            ),
            design=tuple(
                tuple(str(item) for item in row) for row in data.get("design", ())
            ),
            target=tuple(str(item) for item in data.get("target", ())),
            term_names=tuple(str(item) for item in data.get("term_names", ())),
            graph_n=int(data.get("graph_n", 0)),
            graph_edges=tuple(
                (int(edge[0]), int(edge[1])) for edge in data.get("graph_edges", ())
            ),
            poly_n_vars=int(data.get("poly_n_vars", 0)),
            poly_terms=_parse_poly_terms(data.get("poly_terms", ())),
            sample_x=tuple(str(item) for item in data.get("sample_x", ())),
            poly_constraints=tuple(_parse_poly_terms(poly) for poly in constraints_raw),
            extra=extra,
        )

View source ↗

Read the implementation

class Observation:
    """Tagged, JSON-able payload. Values are fraction strings or ints. No numpy."""

    tag: str = ""
    sequence: tuple[str, ...] = ()
    jet_names: tuple[str, ...] = ()
    jet_rows: tuple[tuple[str, ...], ...] = ()
    design: tuple[tuple[str, ...], ...] = ()
    target: tuple[str, ...] = ()
    term_names: tuple[str, ...] = ()
    graph_n: int = 0
    graph_edges: tuple[tuple[int, int], ...] = ()
    poly_n_vars: int = 0
    poly_terms: PolyTerms = ()
    sample_x: tuple[str, ...] = ()
    poly_constraints: tuple[PolyTerms, ...] = ()
    extra: tuple[tuple[str, str], ...] = ()

    def extra_map(self) -> dict[str, str]:
        return dict(self.extra)

    def features(self) -> tuple[int, ...]:
        """Length-:data:`FEATURE_DIM` int tuple for memory / gate keys."""

        names = self.term_names + self.jet_names
        has_second = 1 if any(name in _SECOND_DERIV for name in names) else 0
        has_cont = 1 if ("rho_t" in names and "j_x" in names) else 0
        return (
            1 if self.sequence else 0,
            1 if self.jet_rows or self.jet_names else 0,
            1 if self.design or self.target else 0,
            1 if self.graph_edges or self.graph_n else 0,
            1 if self.poly_terms or self.poly_n_vars else 0,
            len(self.sequence),
            len(self.jet_rows),
            len(self.design),
            int(self.graph_n),
            int(self.poly_n_vars),
            has_second,
            has_cont,
            _sign_change_count(self),
            _poly_degree_estimate(self.poly_terms),
            len(self.poly_constraints),
            1 if self.sample_x else 0,
        )

    def as_dict(self) -> dict[str, Any]:
        return {
            "tag": self.tag,
            "sequence": list(self.sequence),
            "jet_names": list(self.jet_names),
            "jet_rows": [list(row) for row in self.jet_rows],
            "design": [list(row) for row in self.design],
            "target": list(self.target),
            "term_names": list(self.term_names),
            "graph_n": self.graph_n,
            "graph_edges": [list(edge) for edge in self.graph_edges],
            "poly_n_vars": self.poly_n_vars,
            "poly_terms": [[list(exp), coeff] for exp, coeff in self.poly_terms],
            "sample_x": list(self.sample_x),
            "poly_constraints": [
                [[list(exp), coeff] for exp, coeff in poly]
                for poly in self.poly_constraints
            ],
            "extra": dict(self.extra),
            "features": list(self.features()),
        }

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> Observation:
        extra_raw = data.get("extra", ())
        if isinstance(extra_raw, Mapping):
            extra = tuple((str(key), str(value)) for key, value in extra_raw.items())
        else:
            extra = tuple((str(key), str(value)) for key, value in extra_raw)
        constraints_raw = data.get("poly_constraints", ())
        return cls(
            tag=str(data.get("tag", "")),
            sequence=tuple(str(item) for item in data.get("sequence", ())),
            jet_names=tuple(str(item) for item in data.get("jet_names", ())),
            jet_rows=tuple(
                tuple(str(item) for item in row) for row in data.get("jet_rows", ())
            ),
            design=tuple(
                tuple(str(item) for item in row) for row in data.get("design", ())
            ),
            target=tuple(str(item) for item in data.get("target", ())),
            term_names=tuple(str(item) for item in data.get("term_names", ())),
            graph_n=int(data.get("graph_n", 0)),
            graph_edges=tuple(
                (int(edge[0]), int(edge[1])) for edge in data.get("graph_edges", ())
            ),
            poly_n_vars=int(data.get("poly_n_vars", 0)),
            poly_terms=_parse_poly_terms(data.get("poly_terms", ())),
            sample_x=tuple(str(item) for item in data.get("sample_x", ())),
            poly_constraints=tuple(_parse_poly_terms(poly) for poly in constraints_raw),
            extra=extra,
        )

Other public import paths

• omnibias.core.proof.Observation
• omnibias.core.proof.observe.Observation

FUNCTION

class_hit_from_check

omnibias.core.proof.class_hit_from_checkSource ↗

class_hit_from_check(sort: str, check: ExactCheck, *, unique_in_family: bool=False) -> ClassHit

Build a ClassHit from an exact-check payload.

Parameters

Name	Type & default	Contract
sortpositional-or-keyword	strRequired	See the operation contract and implementation below.
checkpositional-or-keyword	ExactCheckRequired	See the operation contract and implementation below.
unique_in_familykeyword-only	boolDefault: False	See the operation contract and implementation below.

ReturnsClassHit

Executed regression · select class grows linear span products

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_select_class_grows_linear_span_products() -> None:
    saved = _isolated()
    try:
        from omnibias.core.proof.condition import register_condition_sort

        family = LinearSpanFamily(
            design=((1, 2, 2), (4, 4, 2), (9, 6, 2), (16, 8, 2)),
            term_names=("y", "yp", "ypp"),
            sort="jet_monomial",
        )
        register_condition_sort("jet_monomial", lambda obs: family)
        obs = Observation(
            jet_names=("y", "yp", "ypp"),
            jet_rows=(("1", "2", "2"), ("4", "4", "2"), ("9", "6", "2"), ("16", "8", "2")),
        )
        selection = select_class(obs, sorts=("jet_monomial",), grow=True, inner_budget=2)
        assert selection.best is not None
        assert selection.best.tier == "exact"
        assert selection.grammar_complete is False
    finally:
        _restore(saved)

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_select_class_grows_linear_span_products" -q

Read the implementation

def class_hit_from_check(
    sort: str,
    check: ExactCheck,
    *,
    unique_in_family: bool = False,
) -> ClassHit:
    """Build a :class:`ClassHit` from an exact-check payload."""

    return ClassHit(
        sort=sort,
        check=check,
        tier=_tier_of(check),
        tokens=_tokens_of(check),
        degree=_degree_of(check),
        unique_in_family=unique_in_family or _unique_of(check),
        rmse=_rmse_of(check),
    )

Related APIsomnibias.core.proof.ClassHit↗

Other public import paths

• omnibias.core.proof.class_hit_from_check
• omnibias.core.proof.observe.class_hit_from_check

FUNCTION

fractions_to_ints

omnibias.core.proof.observe.fractions_to_intsSource ↗

fractions_to_ints(rows: Sequence[Sequence[Fraction]]) -> tuple[tuple[int, ...], ...]

Clear a common denominator so an exact-Q table becomes integers.

Parameters

Name	Type & default	Contract
rowspositional-or-keyword	Sequence[Sequence[Fraction]]Required	See the operation contract and implementation below.

Returnstuple[tuple[int, ...], ...]

Read the implementation

def fractions_to_ints(
    rows: Sequence[Sequence[Fraction]],
) -> tuple[tuple[int, ...], ...]:
    """Clear a common denominator so an exact-``Q`` table becomes integers."""

    denoms = [value.denominator for row in rows for value in row]
    scale = 1
    for denom in denoms:
        scale = scale * denom // gcd(scale, denom)
    return tuple(tuple(int(value * scale) for value in row) for row in rows)

FUNCTION

rank_class_hits

omnibias.core.proof.rank_class_hitsSource ↗

rank_class_hits(hits: Sequence[ClassHit]) -> list[ClassHit]

Total order on certified hits. Exact beats enclosure beats empirical.

RMSE is ignored whenever any exact hit exists. Empirical hits are dropped from the returned list in that case so they can never be best.

Parameters

Name	Type & default	Contract
hitspositional-or-keyword	Sequence[ClassHit]Required	See the operation contract and implementation below.

Returnslist[ClassHit]

Executed regression · rank two exact by token count

Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_rank_two_exact_by_token_count() -> None:
    fat = ClassHit(
        sort="ore",
        check=ExactCheck(ok=True, payload={"tier": "exact"}),
        tier="exact",
        tokens=5,
        degree=0,
    )
    thin = ClassHit(
        sort="jet_monomial",
        check=ExactCheck(ok=True, payload={"tier": "exact"}),
        tier="exact",
        tokens=2,
        degree=4,
    )
    ranked = rank_class_hits([fat, thin])
    assert ranked[0].sort == "jet_monomial"
    assert ranked[0].tokens == 2

python -m pytest "packages/omnibias-core/tests/test_observe.py::test_rank_two_exact_by_token_count" -q

Regression example · rank exact beats empirical

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_rank_exact_beats_empirical() -> None:
    exact = ClassHit(
        sort="jet_monomial",
        check=ExactCheck(ok=True, payload={"tier": "exact", "honesty": {}}),
        tier="exact",
        tokens=3,
        degree=2,
    )
    empirical = ClassHit(
        sort="pde_operator",
        check=ExactCheck(ok=False, payload={"mode": "empirical"}),
        tier="empirical",
        tokens=1,
        degree=0,
        rmse="0.001",
    )
    ranked = rank_class_hits([empirical, exact])
    assert ranked[0].sort == "jet_monomial"
    assert all(hit.tier != "empirical" for hit in ranked)

python -m pytest packages/omnibias-core/tests/test_observe.py -k test_rank_exact_beats_empirical -q

Read the implementation

def rank_class_hits(hits: Sequence[ClassHit]) -> list[ClassHit]:
    """Total order on certified hits. Exact beats enclosure beats empirical.

    RMSE is ignored whenever any exact hit exists. Empirical hits are dropped
    from the returned list in that case so they can never be ``best``.
    """

    has_exact = any(hit.tier == "exact" for hit in hits)
    pool = [hit for hit in hits if hit.tier != "empirical"] if has_exact else list(hits)
    return sorted(
        pool,
        key=lambda hit: (
            _TIER_RANK[hit.tier],
            hit.tokens,
            hit.degree,
            0 if hit.unique_in_family else 1,
            hit.sort,
        ),
    )

Other public import paths

• omnibias.core.proof.rank_class_hits
• omnibias.core.proof.observe.rank_class_hits

← Package overviewRunnable learning examples →