From 6c131240d951ac00457eb13b21d097620f2d0476 Mon Sep 17 00:00:00 2001 From: Ryan James Date: Sat, 10 Oct 2026 20:46:33 -0600 Subject: [PATCH 1/2] CSSR and transCSSR learn from independent Trials Co-authored-by: Cursor --- CHANGELOG.md | 5 ++ docs/inference/cssr.rst | 14 +++++ sofic/inference/__init__.py | 2 + sofic/inference/cssr/__init__.py | 2 + sofic/inference/cssr/counts.py | 85 ++++++++++++++++++++++++------ sofic/inference/cssr/process.py | 67 +++++++++++++---------- sofic/inference/cssr/subtree.py | 2 +- sofic/inference/cssr/transducer.py | 53 +++++++++++-------- tests/test_cssr_trials.py | 84 +++++++++++++++++++++++++++++ 9 files changed, 247 insertions(+), 67 deletions(-) create mode 100644 tests/test_cssr_trials.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 78d5b4b..9311573 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -115,6 +115,11 @@ brute-force reference or a closed-form value and has a regression test. ### New +- `sofic.inference.Trials`: `learn_epsilon_machine_cssr`, + `learn_epsilon_transducer_cssr`, and `suggest_max_history` learn from + independent trials, pooling counts without any history spanning two trials + (`SuffixCounts.from_sequences`; `JointSuffixCounts.from_sequences` accepts + paired `Trials`). A `dit.inference.Trials` is accepted too. - Hypothesis strategies in `sofic.testing` for NFAs, Büchi automata and lassos, Wheeler NFAs, Markov chains, Mealy HMMs, sofic shifts, SFTs, VPAs, NWAs, and Mealy transducers; `ci` and `nightly` Hypothesis profiles (`HYPOTHESIS_PROFILE`). diff --git a/docs/inference/cssr.rst b/docs/inference/cssr.rst index d8931ac..91ca009 100644 --- a/docs/inference/cssr.rst +++ b/docs/inference/cssr.rst @@ -94,10 +94,24 @@ apply directly. observations, method="cssr", max_history="auto", test="exact", correction="bonferroni" ) +Independent recordings of the same process go in a :class:`Trials`. Counts are +pooled across trials, but no suffix spans two trials, so the boundaries add no +spurious histories. :func:`learn_epsilon_transducer_cssr` and +:func:`suggest_max_history` accept :class:`Trials` too, as does a +``dit.inference.Trials``. + +.. code-block:: python + + from sofic.inference import Trials, learn_epsilon_machine_cssr + + machine = learn_epsilon_machine_cssr(Trials([run_0, run_1, run_2]), max_history=3) + .. autofunction:: learn_epsilon_machine_cssr .. autofunction:: suggest_max_history +.. autoclass:: Trials + After reconstruction, check the result with :func:`~sofic.inference.diagnostics.goodness_of_fit` and :func:`~sofic.inference.diagnostics.structure_stability` diff --git a/sofic/inference/__init__.py b/sofic/inference/__init__.py index aa0d8d2..b7c8935 100644 --- a/sofic/inference/__init__.py +++ b/sofic/inference/__init__.py @@ -9,6 +9,7 @@ from sofic.inference import bayesian, cssr, diagnostics, hmm, model_selection, spectral from sofic.inference.cssr import ( + Trials, learn_epsilon_machine_cssr, learn_epsilon_machine_subtree, learn_epsilon_transducer_cssr, @@ -69,6 +70,7 @@ "hmm", "model_selection", "spectral", + "Trials", "learn_epsilon_machine_cssr", "GoodnessOfFit", "StructureStability", diff --git a/sofic/inference/cssr/__init__.py b/sofic/inference/cssr/__init__.py index 56f6284..68d95a0 100644 --- a/sofic/inference/cssr/__init__.py +++ b/sofic/inference/cssr/__init__.py @@ -13,6 +13,7 @@ JointSuffixCounts, StackSuffixCounts, SuffixCounts, + Trials, ) from sofic.inference.cssr.process import learn_epsilon_machine_cssr, suggest_max_history from sofic.inference.cssr.significance import ( @@ -40,6 +41,7 @@ "StackSuffixCounts", "SuffixCounts", "TableTest", + "Trials", "aggregates_differ", "learn_epsilon_machine_cssr", "learn_stack_hmm_mle", diff --git a/sofic/inference/cssr/counts.py b/sofic/inference/cssr/counts.py index bc3c523..5e3d970 100644 --- a/sofic/inference/cssr/counts.py +++ b/sofic/inference/cssr/counts.py @@ -22,7 +22,33 @@ ConfigurationHistory = tuple[tuple[Any, ...], tuple[Any, ...]] -def infer_alphabet(tokens: Sequence[Any], alphabet: Sequence[Any] | None) -> tuple[Any, ...]: +class Trials(list): + """Independent realizations (trials) of the same process. + + Wrap a list of sequences in :class:`Trials` to tell the CSSR learners that they + are separate recordings: counts are pooled across trials, but no history ever + spans the boundary between two trials. Trials may have different lengths. A + ``dit.inference.Trials`` is accepted wherever this class is. + """ + + +def is_trials(data: Any) -> bool: + """Whether ``data`` is a :class:`Trials` (or a ``dit.inference.Trials``).""" + if isinstance(data, Trials): + return True + try: + from dit.inference import Trials as DitTrials + except ImportError: + return False + return isinstance(data, DitTrials) + + +def split_trials(data: Any) -> list[tuple[Any, ...]]: + """The trials of ``data`` as tuples: each trial of a :class:`Trials`, else ``data`` as one trial.""" + return [tuple(trial) for trial in data] if is_trials(data) else [tuple(data)] + + +def infer_alphabet(tokens: Iterable[Any], alphabet: Sequence[Any] | None) -> tuple[Any, ...]: """``alphabet`` as a tuple, or the distinct ``tokens`` sorted by ``repr``.""" return tuple(sorted(set(tokens), key=repr)) if alphabet is None else tuple(alphabet) @@ -56,17 +82,32 @@ def from_sequence( alphabet: Sequence[Any] | None = None, max_length: int | None = None, ) -> SuffixCounts: - seq = tuple(sequence) - if not seq: + if not is_trials(sequence) and not tuple(sequence): raise ValueError("sequence must be non-empty") - alphabet = infer_alphabet(seq, alphabet) - unknown = set(seq) - set(alphabet) + return cls.from_sequences(split_trials(sequence), alphabet=alphabet, max_length=max_length) + + @classmethod + def from_sequences( + cls, + sequences: Iterable[Sequence[Any]], + *, + alphabet: Sequence[Any] | None = None, + max_length: int | None = None, + ) -> SuffixCounts: + """Counts pooled over independent ``sequences``; no history spans two of them.""" + trials = [tuple(seq) for seq in sequences] + symbols = {symbol for seq in trials for symbol in seq} + if not symbols: + raise ValueError("sequences must be non-empty") + alphabet = infer_alphabet(symbols, alphabet) + unknown = symbols - set(alphabet) if unknown: raise ValueError(f"symbols {unknown!r} not in alphabet") counts = cls(alphabet=alphabet) - for t, history in iter_suffixes(seq, max_length if max_length is not None else len(seq)): - counts.history_counts[history] += 1 - counts.next_counts[history][seq[t]] += 1 + for seq in trials: + for t, history in iter_suffixes(seq, max_length if max_length is not None else len(seq)): + counts.history_counts[history] += 1 + counts.next_counts[history][seq[t]] += 1 return counts def morph(self, history: History, *, smoothing: float = 0.0) -> dict[Any, float]: @@ -138,18 +179,18 @@ def from_sequences( output_alphabet: Sequence[Any] | None = None, max_length: int, ) -> JointSuffixCounts: - xs = tuple(inputs) - ys = tuple(outputs) - if len(xs) != len(ys): - raise ValueError("inputs and outputs must have equal length") - if not xs: + """Counts of ``(inputs, outputs)``; both may be :class:`Trials` of matching lengths.""" + x_trials, y_trials = split_paired_trials(inputs, outputs) + if not any(x_trials): raise ValueError("sequences must be non-empty") counts = cls( - input_alphabet=infer_alphabet(xs, input_alphabet), output_alphabet=infer_alphabet(ys, output_alphabet) + input_alphabet=infer_alphabet((x for xs in x_trials for x in xs), input_alphabet), + output_alphabet=infer_alphabet((y for ys in y_trials for y in ys), output_alphabet), ) - for t, history in iter_suffixes(tuple(zip(xs, ys, strict=True)), max_length): - counts.history_counts[history] += 1 - counts.next_counts.setdefault(history, {}).setdefault(xs[t], Counter())[ys[t]] += 1 + for xs, ys in zip(x_trials, y_trials, strict=True): + for t, history in iter_suffixes(tuple(zip(xs, ys, strict=True)), max_length): + counts.history_counts[history] += 1 + counts.next_counts.setdefault(history, {}).setdefault(xs[t], Counter())[ys[t]] += 1 return counts def output_counts(self, histories: set[JointHistory], input_symbol: Any) -> Counter[Any]: @@ -172,6 +213,16 @@ def state_morph(self, histories: set[JointHistory], input_symbol: Any) -> dict[A return {symbol: observed.get(symbol, 0) / total for symbol in self.output_alphabet} +def split_paired_trials(inputs: Any, outputs: Any) -> tuple[list[tuple[Any, ...]], list[tuple[Any, ...]]]: + """The trials of paired ``inputs`` / ``outputs``: both :class:`Trials`, or both single sequences.""" + if is_trials(inputs) != is_trials(outputs): + raise ValueError("inputs and outputs must both be Trials, or both be single sequences") + x_trials, y_trials = split_trials(inputs), split_trials(outputs) + if len(x_trials) != len(y_trials) or any(len(xs) != len(ys) for xs, ys in zip(x_trials, y_trials, strict=True)): + raise ValueError("inputs and outputs must have equal length") + return x_trials, y_trials + + #: Aggregated output counts of a state: ``agg[input_symbol]`` is a Counter over outputs. StateAggregate = dict[Any, Counter[Any]] diff --git a/sofic/inference/cssr/process.py b/sofic/inference/cssr/process.py index 74c029a..4048635 100644 --- a/sofic/inference/cssr/process.py +++ b/sofic/inference/cssr/process.py @@ -14,7 +14,7 @@ from sofic.exceptions import StochasticValidationError from sofic.generators.epsilon_machine import EpsilonMachine from sofic.graph import ATTR_EMISSION, ATTR_PROB, TransitionGraph -from sofic.inference.cssr.counts import History, SuffixCounts +from sofic.inference.cssr.counts import History, SuffixCounts, is_trials, split_trials from sofic.inference.cssr.significance import ( MorphTest, _bonferroni_alpha, @@ -197,7 +197,7 @@ def _suffix_determinize( def _suffix_machine( states: list[set[History]], counts: SuffixCounts, - sequence: Sequence[Any], + trials: Sequence[tuple[Any, ...]], alive: set[int], *, max_history: int, @@ -210,13 +210,13 @@ def _suffix_machine( history_to_state = {h: index for index in alive for h in states[index]} visits: Counter[int] = Counter() - seq = tuple(sequence) - for t in range(len(seq) + 1): - for length in range(min(t, max_history), -1, -1): - state = history_to_state.get(seq[t - length : t]) - if state is not None: - visits[state] += 1 - break + for seq in trials: + for t in range(len(seq) + 1): + for length in range(min(t, max_history), -1, -1): + state = history_to_state.get(seq[t - length : t]) + if state is not None: + visits[state] += 1 + break classes = _recurrent_states(edges) if not classes: @@ -265,13 +265,16 @@ def _suffix_machine( def _suffix_reconstruct( states: list[set[History]], counts: SuffixCounts, - sequence: Sequence[Any], + trials: Sequence[tuple[Any, ...]], *, max_history: int, alpha: float, test: MorphTest, ) -> EpsilonMachine: - """Prune transient states, determinize, and build the machine from homogeneous ``states``.""" + """Prune transient states, determinize, and build the machine from homogeneous ``states``. + + ``trials`` are the observed sequences, one tuple per independent trial. + """ def reconstruct(resolve: bool) -> EpsilonMachine: everything = set(range(len(states))) @@ -283,13 +286,13 @@ def reconstruct(resolve: bool) -> EpsilonMachine: states, counts, alive, max_history=max_history, alpha=alpha, test=test, resolve=resolve ) return _suffix_machine( - split, counts, sequence, alive, max_history=max_history, alpha=alpha, test=test, resolve=resolve + split, counts, trials, alive, max_history=max_history, alpha=alpha, test=test, resolve=resolve ) machine = reconstruct(resolve=True) # Resolving truncated successors needs max_history at least the synchronization length. When it # is shorter, resolution can close off a state that never emits some observed symbol. - if {t.data[ATTR_EMISSION] for t in machine.transitions()} < set(sequence): + if {t.data[ATTR_EMISSION] for t in machine.transitions()} < {symbol for seq in trials for symbol in seq}: machine = reconstruct(resolve=False) return machine @@ -314,7 +317,7 @@ def suggest_max_history( Parameters ---------- sequence - Observed symbols. + Observed symbols, or independent :class:`~sofic.inference.cssr.Trials` of them. alpha Significance level of each order test. max_order @@ -341,11 +344,19 @@ def suggest_max_history( if select_markov_order is None: # pragma: no cover - depends on the installed dit raise ImportError("suggest_max_history requires a dit release with dit.inference.select_markov_order") codes: dict[Any, str] = {} - seq = [codes.setdefault(symbol, str(len(codes))) for symbol in sequence] + trials = [[codes.setdefault(symbol, str(len(codes))) for symbol in seq] for seq in split_trials(sequence)] + n = sum(len(seq) for seq in trials) if max_order is None: - k = max(2, len(set(seq))) - max_order = max(1, min(10, int(np.log(max(len(seq), 1) / 5) / np.log(k)) - 1)) - order = select_markov_order(seq, max_order, method=method, alpha=alpha, n_surrogates=n_surrogates, prng=seed) + k = max(2, len(codes)) + max_order = max(1, min(10, int(np.log(max(n, 1) / 5) / np.log(k)) - 1)) + if is_trials(sequence): + DitTrials = getattr(dit.inference, "Trials", None) + if DitTrials is None: # pragma: no cover - depends on the installed dit + raise ImportError("suggest_max_history on Trials requires a dit release with dit.inference.Trials") + data: Any = DitTrials(trials) + else: + data = trials[0] + order = select_markov_order(data, max_order, method=method, alpha=alpha, n_surrogates=n_surrogates, prng=seed) return max(1, int(order)) @@ -369,11 +380,12 @@ def learn_epsilon_machine_cssr( Parameters ---------- sequence - Observed symbols. + Observed symbols, or independent :class:`~sofic.inference.cssr.Trials` of them: + counts are pooled across trials, but no suffix spans two trials. alphabet Symbol alphabet; defaults to the symbols in ``sequence``. max_history - Longest suffix considered; by default a third of ``log_k len(sequence)``, + Longest suffix considered; by default a third of ``log_k n`` for ``n`` symbols in total, between 1 and 10. It should be at least the synchronization length of the source (for a Markov source, its order). Larger values run many more significance tests, and some of them split states by chance. ``"auto"`` @@ -403,21 +415,22 @@ def learn_epsilon_machine_cssr( reconstruction. CSSR then returns more states than the ε-machine, with an entropy rate that approaches the true one from above as ``max_history`` grows. """ - seq = tuple(sequence) - if len(seq) < 2: + trials = split_trials(sequence) + n = sum(len(seq) for seq in trials) + if n < 2: raise ValueError("sequence must contain at least two symbols") - alphabet_size = len(set(seq)) if alphabet is None else len(tuple(alphabet)) + alphabet_size = len({symbol for seq in trials for symbol in seq}) if alphabet is None else len(tuple(alphabet)) if max_history == "auto": - max_length = suggest_max_history(seq, alpha=alpha) + max_length = suggest_max_history(sequence if is_trials(sequence) else trials[0], alpha=alpha) else: - max_length = max_history if max_history is not None else _default_max_history(len(seq), alphabet_size) + max_length = max_history if max_history is not None else _default_max_history(n, alphabet_size) if max_length < 0: raise ValueError("max_history must be non-negative") - counts = SuffixCounts.from_sequence(seq, alphabet=alphabet, max_length=max_length + 1) + counts = SuffixCounts.from_sequences(trials, alphabet=alphabet, max_length=max_length + 1) if correction == "bonferroni": alpha = _bonferroni_alpha(counts, alpha, max_length=max_length, min_count=min_count, resolution_tests=True) elif correction is not None: raise ValueError(f"unknown correction {correction!r}") homogeneous = _suffix_homogenize(counts, max_history=max_length, alpha=alpha, test=test, min_count=min_count) - return _suffix_reconstruct(homogeneous, counts, seq, max_history=max_length, alpha=alpha, test=test) + return _suffix_reconstruct(homogeneous, counts, trials, max_history=max_length, alpha=alpha, test=test) diff --git a/sofic/inference/cssr/subtree.py b/sofic/inference/cssr/subtree.py index a0dd762..2105560 100644 --- a/sofic/inference/cssr/subtree.py +++ b/sofic/inference/cssr/subtree.py @@ -122,4 +122,4 @@ def learn_epsilon_machine_subtree( raise ValueError(f"unknown correction {correction!r}") states = list(_cluster_histories_by_morph(counts, histories, delta=delta, alpha=alpha, test=test).values()) - return _suffix_reconstruct(states, counts, seq, max_history=max_history, alpha=alpha, test=test) + return _suffix_reconstruct(states, counts, [seq], max_history=max_history, alpha=alpha, test=test) diff --git a/sofic/inference/cssr/transducer.py b/sofic/inference/cssr/transducer.py index 3d0484f..289fd0d 100644 --- a/sofic/inference/cssr/transducer.py +++ b/sofic/inference/cssr/transducer.py @@ -23,9 +23,12 @@ JointHistory, JointSuffixCounts, StateAggregate, + Trials, _history_aggregate, _merge_aggregate, _state_aggregate, + is_trials, + split_paired_trials, ) from sofic.inference.cssr.process import _recurrent_states, suggest_max_history from sofic.inference.cssr.significance import TableTest, _aggregate_score, aggregates_differ @@ -202,8 +205,8 @@ def _build_transducer( states: list[set[JointHistory]], counts: JointSuffixCounts, alive: set[int], - inputs: Sequence[Any], - outputs: Sequence[Any], + input_trials: Sequence[tuple[Any, ...]], + output_trials: Sequence[tuple[Any, ...]], *, max_history: int, alpha: float, @@ -213,13 +216,14 @@ def _build_transducer( history_to_state = {h: index for index in alive for h in states[index]} visits: Counter[int] = Counter() - pairs = tuple(zip(inputs, outputs, strict=True)) - for t in range(len(pairs) + 1): - for hist_len in range(min(t, max_history), -1, -1): - state = history_to_state.get(pairs[t - hist_len : t]) - if state is not None: - visits[state] += 1 - break + for xs, ys in zip(input_trials, output_trials, strict=True): + pairs = tuple(zip(xs, ys, strict=True)) + for t in range(len(pairs) + 1): + for hist_len in range(min(t, max_history), -1, -1): + state = history_to_state.get(pairs[t - hist_len : t]) + if state is not None: + visits[state] += 1 + break classes = _recurrent_states(edges) if not classes: @@ -293,6 +297,14 @@ def learn_epsilon_transducer_cssr( ) -> EpsilonTransducer: """Reconstruct an ε-transducer from paired input/output sequences (transCSSR). + ``inputs`` and ``outputs`` are paired sequences, or paired + :class:`~sofic.inference.cssr.Trials` of equal-length sequences: counts are pooled + across trials, but no joint history spans two trials. + + The transducer is Mealy: on reading input ``x_t`` it emits ``y_t``, so ``y_t`` may + depend on ``x_t`` and on the joint past. To model a channel with delay ``d``, pass + the input shifted so that ``inputs[t]`` is the source symbol at ``t - d``. + ``alpha`` is the per-test significance level for the causal-state split decision; the transCSSR/CSSR default of ``0.001`` favors fewer, more robust states. ``max_history`` bounds the joint-history depth and ``min_count`` the minimum @@ -306,26 +318,23 @@ def learn_epsilon_transducer_cssr( (history, input symbol) tests that can split a state; ``correction=None`` disables it. """ - xs = tuple(inputs) - ys = tuple(outputs) - if len(xs) != len(ys): - raise ValueError("inputs and outputs must have equal length") - if len(xs) < 2: + x_trials, y_trials = split_paired_trials(inputs, outputs) + n = sum(len(xs) for xs in x_trials) + if n < 2: raise ValueError("sequences must contain at least two symbols") joint_alphabet_size = ( - len(set(xs)) * len(set(ys)) + len({x for xs in x_trials for x in xs}) * len({y for ys in y_trials for y in ys}) if input_alphabet is None or output_alphabet is None else len(tuple(input_alphabet)) * len(tuple(output_alphabet)) ) if max_history == "auto": - max_length = suggest_max_history(list(zip(xs, ys, strict=True)), alpha=alpha) + joint = [list(zip(xs, ys, strict=True)) for xs, ys in zip(x_trials, y_trials, strict=True)] + max_length = suggest_max_history(Trials(joint) if is_trials(inputs) else joint[0], alpha=alpha) else: - max_length = ( - max_history if max_history is not None else _default_max_history(len(xs), joint_alphabet_size, min_count) - ) + max_length = max_history if max_history is not None else _default_max_history(n, joint_alphabet_size, min_count) counts = JointSuffixCounts.from_sequences( - xs, - ys, + Trials(x_trials), + Trials(y_trials), input_alphabet=input_alphabet, output_alphabet=output_alphabet, max_length=max_length + 1, @@ -345,4 +354,4 @@ def learn_epsilon_transducer_cssr( edges = _edges(states, counts, everything, max_history=max_length, alpha=alpha, test=test) alive = set().union(*_recurrent_states(edges)) or everything states, alive = _determinize(states, counts, alive, max_history=max_length, alpha=alpha, test=test) - return _build_transducer(states, counts, alive, xs, ys, max_history=max_length, alpha=alpha, test=test) + return _build_transducer(states, counts, alive, x_trials, y_trials, max_history=max_length, alpha=alpha, test=test) diff --git a/tests/test_cssr_trials.py b/tests/test_cssr_trials.py new file mode 100644 index 0000000..fa69757 --- /dev/null +++ b/tests/test_cssr_trials.py @@ -0,0 +1,84 @@ +"""CSSR and transCSSR on independent trials: pooled counts, no cross-boundary histories.""" + +from collections import Counter + +import numpy as np +import pytest +from hypothesis import given +from hypothesis import strategies as st + +from sofic.automata.transducer_operations import compose_transducer_generator +from sofic.examples.epsilon_machines import even_process, golden_mean +from sofic.examples.processes import binary_channel +from sofic.generators.sampling import sample +from sofic.inference import Trials, topology_key +from sofic.inference.cssr import ( + JointSuffixCounts, + SuffixCounts, + learn_epsilon_machine_cssr, + learn_epsilon_transducer_cssr, +) + +from .test_epsilon_transducer_inference import _iid_input + +trials_strategy = st.lists(st.lists(st.sampled_from("01"), min_size=1, max_size=12), min_size=1, max_size=5) + + +def _split(seq, k): + bounds = np.linspace(0, len(seq), k + 1).astype(int) + return Trials([seq[a:b] for a, b in zip(bounds[:-1], bounds[1:], strict=True)]) + + +@given(trials_strategy) +def test_pooled_counts_are_the_sum_of_per_trial_counts(trials): + """No history spans two trials: pooling equals summing each trial's own counts.""" + pooled = SuffixCounts.from_sequences(trials, alphabet="01", max_length=3) + history, following = Counter(), Counter() + for seq in trials: + single = SuffixCounts.from_sequence(seq, alphabet="01", max_length=3) + history.update(single.history_counts) + following.update({(h, s): c for h, counter in single.next_counts.items() for s, c in counter.items()}) + assert pooled.history_counts == history + assert Counter({(h, s): c for h, counter in pooled.next_counts.items() for s, c in counter.items()}) == following + + +@given(trials_strategy) +def test_joint_counts_never_span_trials(trials): + outputs = [seq[::-1] for seq in trials] + counts = JointSuffixCounts.from_sequences(Trials(trials), Trials(outputs), max_length=3) + pairs = [tuple(zip(xs, ys, strict=True)) for xs, ys in zip(trials, outputs, strict=True)] + for history in counts.history_counts: + assert any(pair[i : i + len(history)] == history for pair in pairs for i in range(len(pair) - len(history) + 1)) + assert counts.history_counts[()] == sum(len(p) for p in pairs) + + +def test_single_sequence_is_one_trial(): + seq = "0110100111" + assert SuffixCounts.from_sequence(seq, max_length=2) == SuffixCounts.from_sequences([seq], max_length=2) + assert SuffixCounts.from_sequence(Trials([seq]), max_length=2) == SuffixCounts.from_sequence(seq, max_length=2) + + +def test_mismatched_trials_raise(): + with pytest.raises(ValueError, match="both be Trials"): + JointSuffixCounts.from_sequences(Trials(["01"]), "01", max_length=1) + with pytest.raises(ValueError, match="equal length"): + JointSuffixCounts.from_sequences(Trials(["01", "1"]), Trials(["01", "10"]), max_length=1) + + +@pytest.mark.parametrize("factory", [golden_mean, even_process]) +def test_cssr_on_trials_matches_the_unsplit_sample(factory): + seq, _ = sample(factory(), 20000, np.random.default_rng(0)) + whole = learn_epsilon_machine_cssr(seq, max_history=3) + split = learn_epsilon_machine_cssr(_split(list(seq), 4), max_history=3) + assert topology_key(split) == topology_key(whole) + assert split.entropy_rate() == pytest.approx(whole.entropy_rate(), abs=0.01) + + +def test_transcssr_on_trials_matches_the_unsplit_sample(): + joint = compose_transducer_generator(binary_channel(0.1, 0.2), _iid_input(), joint=True) + observations, _ = joint.sample(20000, np.random.default_rng(1)) + xs = [pair[0] for pair in observations] + ys = [pair[1] for pair in observations] + whole = learn_epsilon_transducer_cssr(xs, ys, max_history=1) + split = learn_epsilon_transducer_cssr(_split(xs, 3), _split(ys, 3), max_history=1) + assert len(list(split.states())) == len(list(whole.states())) == 1 From e517a4dce30a2aa310a3e17aee9ed129f9bd5626 Mon Sep 17 00:00:00 2001 From: Ryan James Date: Sun, 11 Oct 2026 21:14:59 -0600 Subject: [PATCH 2/2] is_trials: look up dit.inference.Trials dynamically (absent in older dit) Co-authored-by: Cursor --- sofic/inference/cssr/counts.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/sofic/inference/cssr/counts.py b/sofic/inference/cssr/counts.py index 5e3d970..4be6def 100644 --- a/sofic/inference/cssr/counts.py +++ b/sofic/inference/cssr/counts.py @@ -37,10 +37,12 @@ def is_trials(data: Any) -> bool: if isinstance(data, Trials): return True try: - from dit.inference import Trials as DitTrials - except ImportError: + import dit.inference + except ImportError: # pragma: no cover - dit is a dependency return False - return isinstance(data, DitTrials) + # Older dit releases have no Trials. + dit_trials = getattr(dit.inference, "Trials", None) + return dit_trials is not None and isinstance(data, dit_trials) def split_trials(data: Any) -> list[tuple[Any, ...]]: