# resolve_ties.py
"""Resolve ties between IIT objects."""
from collections.abc import Callable
from collections.abc import Iterable
from collections.abc import Iterator
from collections.abc import Mapping
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any
from typing import Literal
from typing import Protocol
from typing import TypeVar
from . import numerics
from .conf import config
from .conf import fallback
from .registry import Registry
from .utils import NO_DEFAULT
from .utils import iter_with_default
T = TypeVar("T")
# ---------------------------------------------------------------------------
# Cascade primitive
# ---------------------------------------------------------------------------
#
# See ``docs/superpowers/specs/2026-05-13-cascade-execution-model.md`` for
# the design. The cascade walks the IIT postulate hierarchy
# (Existence → Intrinsicality → Information → Integration → Exclusion →
# Composition, plus a pyphi-specific Determinism level) and resolves ties
# at the lowest sufficient postulate. A ``ResolutionContext`` bounds the
# walk's escalation budget and memoizes per-candidate intermediate values.
Postulate = Literal[
"Existence",
"Intrinsicality",
"Information",
"Integration",
"Exclusion",
"Composition",
"Determinism",
]
# Postulate order. Determinism sits after Composition as a pyphi-specific
# canonicalization fallback for ties that don't violate any postulate
# (e.g., mechanism MIPs tied at unnormalized phi with different
# partitions — extrinsic labeling ties).
_POSTULATE_ORDER: tuple[Postulate, ...] = (
"Existence",
"Intrinsicality",
"Information",
"Integration",
"Exclusion",
"Composition",
"Determinism",
)
def _postulate_rank(postulate: Postulate) -> int:
return _POSTULATE_ORDER.index(postulate)
CascadeOp = Literal["argmax", "argmin", "filter"]
[docs]
@dataclass(frozen=True)
class CascadeLevel:
"""A single named step in a cascade.
``postulate`` names the IIT postulate this step instantiates;
used to consult the ``ResolutionContext``'s escalation budget.
``op`` is the reduction (argmax / argmin / filter). ``key`` is a
callable returning the comparison value for a candidate (or a
bool for ``filter``).
"""
postulate: Postulate
op: CascadeOp
key: Callable[[Any], Any]
CascadeOutcomeStatus = Literal[
"RESOLVED",
"UNRESOLVED_WITHIN_BUDGET",
]
[docs]
@dataclass(frozen=True)
class CascadeOutcome[U]:
"""The outcome of running a cascade.
``resolved`` is the unique winner when ``outcome == 'RESOLVED'``;
``None`` otherwise. ``tied_set`` is the set of candidates that
survived to the final level the cascade reached. ``cascade_level``
names that postulate.
"""
resolved: U | None
tied_set: tuple[U, ...]
cascade_level: Postulate
outcome: CascadeOutcomeStatus
failure_reason: str | None = None
[docs]
class NotAComplex(Exception):
"""Raised when a cascade reaches its final level with a tie and
``on_unresolved='fail'``. Indicates that the substrate fails the
postulate at which the tie persists; downstream callers convert
to a Null* sentinel for the public API.
"""
def __init__(
self,
tied_set: Sequence[Any],
cascade_level: Postulate,
failure_reason: str | None = None,
) -> None:
super().__init__(
f"Cascade exhausted at {cascade_level}: "
f"{len(tied_set)} tied candidates remain"
)
self.tied_set: tuple[Any, ...] = tuple(tied_set)
self.cascade_level: Postulate = cascade_level
self.failure_reason: str | None = failure_reason
[docs]
class ResolutionContext:
"""Per-computation context for cascade tie resolution.
Carries the entry-point function's escalation budget and a
memoization cache shared across nested cascade calls.
"""
def __init__(
self,
max_escalation_level: Postulate,
memo: dict[Any, Any] | None = None,
) -> None:
self.max_escalation_level: Postulate = max_escalation_level
self._memo: dict[Any, Any] = memo if memo is not None else {}
[docs]
def can_escalate_to(self, postulate: Postulate) -> bool:
"""True iff ``postulate`` is at or below ``max_escalation_level``
in the postulate order."""
return _postulate_rank(postulate) <= _postulate_rank(self.max_escalation_level)
[docs]
def memoize[V](self, key: Any, fn: Callable[[], V]) -> V:
"""Return ``memo[key]``, computing via ``fn()`` if absent."""
if key not in self._memo:
self._memo[key] = fn()
return self._memo[key]
[docs]
def child(self) -> "ResolutionContext":
"""Return a child context inheriting parent budget and memo cache."""
return ResolutionContext(
max_escalation_level=self.max_escalation_level,
memo=self._memo,
)
OnUnresolved = Literal["fail", "defer", "warn"]
def _tied_with_extremum[U](
objects: Sequence[U], keys: Sequence[Any], extremum: Any
) -> tuple[U, ...]:
"""Return the objects whose key ties the extremum.
Float keys tie up to ``config.numerics.precision`` (via
:func:`pyphi.numerics.eq`); all other key types (integers, bytes,
tuples) compare exactly.
Parameters
----------
objects : Sequence
The candidate objects, aligned with ``keys``.
keys : Sequence
The comparison key of each object.
extremum : Any
The extremal key against which each object's key is compared.
Returns
-------
tuple
The objects whose key ties ``extremum``.
"""
if isinstance(extremum, float):
return tuple(
o for o, k in zip(objects, keys, strict=True) if numerics.eq(k, extremum)
)
return tuple(o for o, k in zip(objects, keys, strict=True) if k == extremum)
def _apply_level[U](
candidates: Sequence[U],
level: CascadeLevel,
) -> tuple[U, ...]:
"""Apply ``level``'s op to ``candidates`` and return the winners."""
if level.op == "filter":
return tuple(c for c in candidates if level.key(c))
keys = [level.key(c) for c in candidates]
extremum = max(keys) if level.op == "argmax" else min(keys)
return _tied_with_extremum(candidates, keys, extremum)
[docs]
def cascade[U](
candidates: Iterable[U],
levels: Sequence[CascadeLevel],
*,
context: ResolutionContext,
on_unresolved: OnUnresolved = "defer",
) -> CascadeOutcome[U]:
"""Walk a cascade of postulate-level reductions.
At each level, apply ``level.op`` with ``level.key`` to identify
surviving candidates. If a single candidate remains, the cascade
resolves. If multiple remain and the next level is within budget,
recurse on that level. Otherwise:
- ``on_unresolved='defer'`` (default): return
``UNRESOLVED_WITHIN_BUDGET`` carrying the tied set for downstream
surfacing.
- ``on_unresolved='fail'``: raise :class:`NotAComplex`.
- ``on_unresolved='warn'``: emit a warning and return as 'defer'.
"""
survivors: tuple[U, ...] = tuple(candidates)
if not survivors:
raise ValueError("cascade requires at least one candidate")
# Track the most recently *processed* level (one whose op was applied).
# When budget blocks or all levels exhaust, this is the level reported.
last_processed_level: Postulate | None = None
def _unresolved(level: Postulate) -> CascadeOutcome[U]:
# Terminated with a genuine tie — budget-blocked or exhausted alike.
if on_unresolved == "fail":
raise NotAComplex(survivors, level)
if on_unresolved == "warn":
import warnings
warnings.warn(
f"Cascade unresolved at {level} with {len(survivors)} tied candidates",
stacklevel=3,
)
return CascadeOutcome(
resolved=None,
tied_set=survivors,
cascade_level=level,
outcome="UNRESOLVED_WITHIN_BUDGET",
)
for level in levels:
if len(survivors) == 1:
return CascadeOutcome(
resolved=survivors[0],
tied_set=survivors,
cascade_level=last_processed_level or level.postulate,
outcome="RESOLVED",
)
if not context.can_escalate_to(level.postulate):
return _unresolved(last_processed_level or level.postulate)
pre_apply = survivors
survivors = _apply_level(pre_apply, level)
last_processed_level = level.postulate
if len(survivors) == 1:
return CascadeOutcome(
resolved=survivors[0],
tied_set=pre_apply,
cascade_level=level.postulate,
outcome="RESOLVED",
)
final_level: Postulate = last_processed_level or "Determinism"
if len(survivors) == 1:
return CascadeOutcome(
resolved=survivors[0],
tied_set=survivors,
cascade_level=final_level,
outcome="RESOLVED",
)
# Cascade exhausted all levels with a tie.
return _unresolved(final_level)
class _StateMIP(Protocol):
"""Structural type for a per-state MIP result consumed by the state-tie cascade.
Requires ``phi`` (the Integration-level cascade key). ``big_phi``
(Composition-level key) is read only when the cascade escalates
past Integration.
"""
@property
def phi(self) -> float: ...
class _ComplexCandidate(Protocol):
"""Structural type for a candidate complex consumed by the
substrate-exclusion cascade. Requires ``big_phi`` (Composition-level
integrated information of the cause-effect structure).
"""
@property
def big_phi(self) -> float: ...
[docs]
def resolve_complex_tie[V: _ComplexCandidate](
candidates: "Iterable[V]",
*,
context: ResolutionContext,
on_unresolved: OnUnresolved = "defer",
) -> CascadeOutcome[V]:
"""Resolve a substrate-exclusion tie via the Composition cascade.
Per Albantakis et al. 2023 S1 Text: when overlapping substrates tie
at maximum ``φ_s``, the exclusion postulate is resolved by
escalating to ``Φ`` (Composition). The substrate with maximum ``Φ``
qualifies as the complex; overlapping losers are excluded. If
multiple substrates also tie at ``Φ``, the exclusion postulate
fails for that group and they do not qualify as complexes — under
``on_unresolved='fail'`` this raises :exc:`NotAComplex`; under
``'defer'`` the outcome is ``UNRESOLVED_WITHIN_BUDGET`` carrying
the tied set so the caller can proceed to the next-best by
``φ_s``.
Callers are expected to pre-filter candidates to a single
``φ_s``-tied overlap-clique; this function only walks the
Composition step.
"""
return cascade(
candidates,
levels=[
CascadeLevel(
postulate="Composition",
op="argmax",
key=lambda c: c.big_phi,
),
],
context=context,
on_unresolved=on_unresolved,
)
class _CongruentMice(Protocol):
"""Structural type for a MICE consumed by the distinction-state cascade.
Requires ``is_congruent`` (against a per-direction state spec) and
``purview`` (used by the cross-purview heuristic).
"""
@property
def purview(self) -> Sequence[int]: ...
def is_congruent(self, other: Any) -> bool: ...
[docs]
def congruent_distinction_readings[V: _CongruentMice](
state_ties: "Sequence[V] | None",
purview_ties: "Sequence[V] | None",
system_state_spec: Any,
) -> list[V]:
"""Every tied reading of a distinction's cause or effect that is
congruent with the system's specified state.
The candidate set is every tied reading: the union of the state ties
carried by each purview-tied MICE (state ties are same-purview
readings tied at maximum ``ii(m, z)``; purview ties are cross-purview
readings tied at maximum ``φ_d(m, Z)``).
Congruence with ``system_state_spec`` — the direction-specific
component of the system's specified cause-effect state — is a
requirement, not a tie-break (Albantakis et al. 2023 S1 Text): a
non-congruent reading cannot enter the cause-effect structure, and a
distinction with no congruent reading is excluded from it. Selection
among the congruent readings is the Composition appeal — performed
jointly across the structure's distinctions by
:meth:`pyphi.models.distinctions.Distinctions.resolve_congruence`,
which selects the readings that maximize the structure integrated
information Φ, per S1's principle of maximal existence.
"""
candidates: list[V] = []
seen: set[int] = set()
for group in (state_ties or (), purview_ties or ()):
for mice in group:
for reading in getattr(mice, "state_ties", None) or (mice,):
if id(reading) not in seen:
seen.add(id(reading))
candidates.append(reading)
return [m for m in candidates if m.is_congruent(system_state_spec)]
class _AcRIALike(Protocol):
"""Structural type for an AC repertoire-irreducibility analysis."""
@property
def alpha(self) -> float: ...
@property
def purview(self) -> Sequence[int]: ...
@property
def partition(self) -> Any: ...
[docs]
def resolve_ac_partition_tie[V: _AcRIALike](
rias: "Iterable[V]",
*,
context: ResolutionContext,
on_unresolved: OnUnresolved = "defer",
) -> CascadeOutcome[V]:
"""Resolve a tie among per-partition AcRIAs at min ``|alpha|``.
The 2019 AC paper defines the MIP as ``argmin over partitions of
(rho - rho_psi)``; tie-break behavior at this level is unspecified by the
paper. The cascade walks Integration (argmin ``|alpha|``) and falls
through to a pyphi-specific Determinism level (lex-canonical
partition) so identical-alpha partitions resolve reproducibly across
iteration orderings.
"""
return cascade(
rias,
levels=[
CascadeLevel(
postulate="Integration",
op="argmin",
key=lambda r: abs(r.alpha),
),
CascadeLevel(
postulate="Determinism",
op="argmin",
key=lambda r: r.partition.lex_key(),
),
],
context=context,
on_unresolved=on_unresolved,
)
def _ac_minimal_purviews[V: _AcRIALike](rias: "Sequence[V]") -> tuple[V, ...]:
"""Filter to minimal-purview AcRIAs — drop strict supersets.
Implements the Exclusion postulate's minimality condition from
Albantakis et al. 2019 (Definition 1 condition 2 / "AC3" clause):
among candidates tied at alpha_max, an actual cause cannot contain
another tied candidate as a strict subset.
"""
purview_sets = [set(r.purview) for r in rias]
keep = []
for i, r in enumerate(rias):
p_i = purview_sets[i]
is_strict_superset_of_any = any(
p_i > p_j for j, p_j in enumerate(purview_sets) if i != j
)
if not is_strict_superset_of_any:
keep.append(r)
return tuple(keep)
[docs]
def resolve_ac_causal_link_tie[V: _AcRIALike](
rias: "Iterable[V]",
*,
context: ResolutionContext, # noqa: ARG001
) -> CascadeOutcome[V]:
"""Resolve a causal-link tie per the 2019 AC paper Definition 1.
Three-step cascade:
1. **Information**: ``argmax`` over ``alpha`` selects candidates at
``alpha_max``.
2. **Exclusion**: minimality filter drops any candidate whose
purview is a strict superset of another tied candidate's — no
actual cause contains another actual cause as a strict subset.
3. **Determinism** (pyphi canonicalization): when multiple minimal
candidates remain (genuine symmetric over-determination per
Def. 1 outcome 2), pick the lex-smallest purview as the
representative; the full minimal set is the tied set.
A surviving tied set of size > 1 signals an "indeterminate" actual
cause in paper terms; the cascade outcome is
``UNRESOLVED_WITHIN_BUDGET`` with ``tied_set`` carrying every
minimal candidate.
"""
survivors = tuple(rias)
if not survivors:
raise ValueError("resolve_ac_causal_link_tie requires at least one AcRIA")
alphas = [r.alpha for r in survivors]
info_tied = _tied_with_extremum(survivors, alphas, max(alphas))
minimal = _ac_minimal_purviews(info_tied)
if len(minimal) == 1:
return CascadeOutcome(
resolved=minimal[0],
tied_set=minimal,
cascade_level="Information",
outcome="RESOLVED",
)
representative = min(minimal, key=lambda r: tuple(sorted(r.purview)))
return CascadeOutcome(
resolved=representative,
tied_set=minimal,
cascade_level="Exclusion",
outcome="UNRESOLVED_WITHIN_BUDGET",
)
class _AcSIALike(Protocol):
"""Structural type for an AC system-irreducibility analysis."""
@property
def alpha(self) -> float: ...
@property
def size(self) -> int: ...
@property
def partition(self) -> Any: ...
[docs]
def resolve_ac_sia_tie[V: _AcSIALike](
sias: "Iterable[V]",
*,
context: ResolutionContext,
on_unresolved: OnUnresolved = "defer",
) -> CascadeOutcome[V]:
"""Resolve a tie among per-partition AC system analyses at min 𝒜.
The system-level MIP is the partition of minimum 𝒜 (Albantakis et
al. 2019, Eq. 20); tie-break behavior is unspecified by the paper.
The cascade walks Integration (argmin 𝒜) and falls through to a
pyphi-specific Determinism level (lex-canonical partition) so
equal-𝒜 partitions resolve reproducibly across iteration orderings.
"""
return cascade(
sias,
levels=[
CascadeLevel(
postulate="Integration",
op="argmin",
key=lambda s: s.alpha,
),
CascadeLevel(
postulate="Determinism",
op="argmin",
key=lambda s: s.partition.lex_key(),
),
],
context=context,
on_unresolved=on_unresolved,
)
[docs]
def resolve_ac_nexus_tie[V: _AcSIALike](
sias: "Iterable[V]",
*,
context: ResolutionContext,
on_unresolved: OnUnresolved = "defer",
) -> CascadeOutcome[V]:
"""Resolve a tie among candidate transitions for the causal nexus.
The causal nexus is the transition of maximal 𝒜. Ties escalate to
the larger transition, then to a pyphi-specific Determinism level
(lex-smallest cause/effect index sets) for reproducibility.
"""
return cascade(
sias,
levels=[
CascadeLevel(
postulate="Integration",
op="argmax",
key=lambda s: s.alpha,
),
CascadeLevel(
postulate="Integration",
op="argmax",
key=lambda s: s.size,
),
CascadeLevel(
postulate="Determinism",
op="argmin",
key=lambda s: (
tuple(sorted(s.cause_indices)),
tuple(sorted(s.effect_indices)),
),
),
],
context=context,
on_unresolved=on_unresolved,
)
class _IIT3SiaLike(Protocol):
"""Structural type for an IIT 3.0 SIA consumed by the cross-subsystem cascade."""
@property
def node_indices(self) -> Sequence[int]: ...
[docs]
def resolve_iit3_complex_tie[V: _IIT3SiaLike](
sias: "Iterable[V]",
*,
context: ResolutionContext, # noqa: ARG001
on_unresolved: OnUnresolved = "defer",
) -> CascadeOutcome[V]:
"""Resolve a cross-subsystem complex tie on the IIT 3.0 path.
The IIT 3.0 formalism (Oizumi et al. 2014) does not prescribe a
system-level tie-break, and no further postulate provides a
paper-canonical escalation. Returns a lex-smallest candidate as
a diagnostic representative and flags
``UNRESOLVED_WITHIN_BUDGET`` so callers treat the clique as
failing the exclusion postulate.
"""
survivors = tuple(sias)
if not survivors:
raise ValueError("resolve_iit3_complex_tie requires at least one SIA")
if len(survivors) == 1:
return CascadeOutcome(
resolved=survivors[0],
tied_set=survivors,
cascade_level="Exclusion",
outcome="RESOLVED",
)
representative = min(survivors, key=lambda s: tuple(sorted(s.node_indices)))
if on_unresolved == "fail":
raise NotAComplex(
tied_set=survivors,
cascade_level="Exclusion",
failure_reason=(
f"IIT 3.0 cross-subsystem complex tie with {len(survivors)} candidates; "
"no paper-canonical escalation available"
),
)
if on_unresolved == "warn":
import warnings
warnings.warn(
f"IIT 3.0 cross-subsystem complex tie with {len(survivors)} candidates; "
"no paper-canonical escalation available",
stacklevel=2,
)
return CascadeOutcome(
resolved=representative,
tied_set=survivors,
cascade_level="Exclusion",
outcome="UNRESOLVED_WITHIN_BUDGET",
)
[docs]
def resolve_state_tie[K, V: _StateMIP](
per_state_mips: "Mapping[K, V]",
*,
context: ResolutionContext,
on_unresolved: OnUnresolved = "defer",
) -> CascadeOutcome[K]:
"""Resolve a state tie via the per-state max-min cascade.
Per Albantakis et al. 2023 S1 Text + Eq 20 parenthetical: among states
tied at maximum intrinsic information ``ii``, the canonical winner is
the state whose per-state ``φ_s`` (the value of integrated information
at that state's MIP) is greatest. If ``φ_s`` ties too, the cascade
escalates to per-state ``Φ`` (the cause-effect structure's integrated
information at Composition). If ``Φ`` ties as well, the substrate
fails the information postulate — :exc:`NotAComplex` is raised under
``on_unresolved='fail'``.
``per_state_mips`` maps each candidate state spec to a per-state MIP
result object (typically a :class:`SystemIrreducibilityAnalysis` or
similar). The object must expose ``.phi``; ``.big_phi`` is consulted
only when Composition escalation fires.
Returns a :class:`CascadeOutcome` whose ``resolved`` field is the
winning key from ``per_state_mips`` (or ``None`` when budget caps
escalation short of resolution).
"""
return cascade(
list(per_state_mips.keys()),
levels=[
CascadeLevel(
postulate="Integration",
op="argmax",
key=lambda spec: per_state_mips[spec].phi,
),
CascadeLevel(
postulate="Composition",
op="argmax",
key=lambda spec: per_state_mips[spec].big_phi, # pyright: ignore[reportAttributeAccessIssue]
),
],
context=context,
on_unresolved=on_unresolved,
)
# Suppress "unused" warning for field — used in CascadeOutcome subclasses
# that may extend with diagnostic tables.
[docs]
class PhiObjectTieResolutionRegistry(Registry):
"""Storage for functions for resolving ties among phi-objects."""
desc = "functions for resolving ties among phi-objects"
phi_object_tie_resolution_strategies = PhiObjectTieResolutionRegistry()
@phi_object_tie_resolution_strategies.register("PURVIEW_SIZE")
def _(m):
return len(m.purview)
@phi_object_tie_resolution_strategies.register("NEGATIVE_PURVIEW_SIZE")
def _(m):
return -len(m.purview)
@phi_object_tie_resolution_strategies.register("PHI")
def _(m):
return m.phi
@phi_object_tie_resolution_strategies.register("NEGATIVE_PHI")
def _(m):
return -m.phi
@phi_object_tie_resolution_strategies.register("NORMALIZED_PHI")
def _(m):
return m.normalized_phi
@phi_object_tie_resolution_strategies.register("NEGATIVE_NORMALIZED_PHI")
def _(m):
return -m.normalized_phi
@phi_object_tie_resolution_strategies.register("NONE")
def _(m):
raise NotImplementedError(
'tie resolution strategy "NONE" should never be called; '
"it must be special-cased in the resolve() function"
)
@phi_object_tie_resolution_strategies.register("PARTITION_LEX")
def _(m):
return m.partition.lex_key()
[docs]
def resolve[T](
objects: Iterable[T],
strategy: str | list[str],
operation: Callable[..., Any],
default: Any = NO_DEFAULT,
) -> Iterator[T]:
"""Filter φ-objects to those extremal under ``strategy``.
Strategy components apply lexicographically: for each component in
order, the exact extremum of the component's key is computed over
the surviving objects, and every object whose key ties it survives
(float keys tie up to ``config.numerics.precision``; other key
types compare exactly). Later components only see the survivors of
earlier ones. The surviving set is independent of input order.
Parameters
----------
objects : Iterable
The φ-objects to filter.
strategy : str or list of str
A tie-resolution strategy name, or a list of names applied
lexicographically. The special value ``"NONE"`` disables
filtering and yields every object.
operation : callable
The extremum function applied per component — :func:`max` or
:func:`min`.
default : Any, optional
Yielded when ``objects`` is empty.
Yields
------
object
Each φ-object extremal under ``strategy``.
"""
if strategy == "NONE":
yield from iter_with_default(objects, default=default)
return
if isinstance(strategy, str):
strategy = [strategy]
survivors = list(objects)
if not survivors:
yield from iter_with_default(survivors, default=default)
return
for name in strategy:
if len(survivors) == 1:
break
if name == "NONE":
# "NONE" filters nothing, as in the bare-string form.
continue
key_function = phi_object_tie_resolution_strategies[name]
keys = [key_function(obj) for obj in survivors]
extremum = operation(keys)
survivors = list(_tied_with_extremum(survivors, keys, extremum))
yield from survivors
[docs]
def states[T](
rias: Iterable[T], strategy: str | list[str] | None = None, **kwargs: Any
) -> Iterator[T]:
"""Resolve ties among states (RIAs).
Controlled by the ``state_tie_resolution`` configuration option.
"""
strategy = fallback(strategy, config.formalism.iit.state_tie_resolution)
assert strategy is not None, "STATE_TIE_RESOLUTION config must be set"
return resolve(rias, strategy, operation=max, **kwargs)
[docs]
def partitions[T](
mips: Iterable[T], strategy: str | list[str] | None = None, **kwargs: Any
) -> Iterator[T]:
"""Resolve ties among mechanism partitions (MIPs).
Controlled by the ``mip_tie_resolution`` configuration option.
"""
strategy = fallback(strategy, config.formalism.iit.mip_tie_resolution)
assert strategy is not None, "MIP_TIE_RESOLUTION config must be set"
return resolve(mips, strategy, operation=min, **kwargs)
[docs]
def purviews[T](
mice: Iterable[T], strategy: str | list[str] | None = None, **kwargs: Any
) -> Iterator[T]:
"""Resolve ties among purviews (MICEs).
Controlled by the ``purview_tie_resolution`` configuration option.
"""
strategy = fallback(strategy, config.formalism.iit.purview_tie_resolution)
assert strategy is not None, "purview_tie_resolution config must be set"
yield from resolve(mice, strategy, operation=max, **kwargs)
[docs]
def sias[T](
analyses: Iterable[T], strategy: str | list[str] | None = None, **kwargs: Any
) -> Iterator[T]:
"""Resolve ties among system-level SIAs.
Controlled by the ``sia_tie_resolution`` configuration option.
"""
strategy = fallback(strategy, config.formalism.iit.sia_tie_resolution)
assert strategy is not None, "sia_tie_resolution config must be set"
return resolve(analyses, strategy, operation=min, **kwargs)
