tanat.criterion package

Subpackages

• tanat.criterion.type package
  • Submodules
  • tanat.criterion.type.entity module
    • EntityCriterion
      • EntityCriterion.LEVELS
      • EntityCriterion.SETTINGS_CLASS
      • EntityCriterion.__init__()
    • EntityCriterionSettings
      • EntityCriterionSettings.__init__()
      • EntityCriterionSettings.model_dump()
      • EntityCriterionSettings.query
  • tanat.criterion.type.length module
    • LengthCriterion
      • LengthCriterion.LEVELS
      • LengthCriterion.SETTINGS_CLASS
      • LengthCriterion.__init__()
    • LengthCriterionSettings
      • LengthCriterionSettings.__init__()
      • LengthCriterionSettings.ge
      • LengthCriterionSettings.gt
      • LengthCriterionSettings.le
      • LengthCriterionSettings.lt
      • LengthCriterionSettings.model_dump()
  • tanat.criterion.type.pattern module
    • ANY
    • PatternCriterion
      • PatternCriterion.LEVELS
      • PatternCriterion.SETTINGS_CLASS
      • PatternCriterion.__init__()
    • PatternCriterionSettings
      • PatternCriterionSettings.__init__()
      • PatternCriterionSettings.case_sensitive
      • PatternCriterionSettings.feature
      • PatternCriterionSettings.model_dump()
      • PatternCriterionSettings.pattern
      • PatternCriterionSettings.present
      • PatternCriterionSettings.regex
    • WILDCARD
  • tanat.criterion.type.rank module
    • RankCriterion
      • RankCriterion.LEVELS
      • RankCriterion.SETTINGS_CLASS
      • RankCriterion.__init__()
    • RankCriterionSettings
      • RankCriterionSettings.__init__()
      • RankCriterionSettings.end
      • RankCriterionSettings.first
      • RankCriterionSettings.last
      • RankCriterionSettings.model_dump()
      • RankCriterionSettings.ranks
      • RankCriterionSettings.relative
      • RankCriterionSettings.start
      • RankCriterionSettings.step
  • tanat.criterion.type.static module
    • StaticCriterion
      • StaticCriterion.LEVELS
      • StaticCriterion.SETTINGS_CLASS
      • StaticCriterion.__init__()
    • StaticCriterionSettings
      • StaticCriterionSettings.__init__()
      • StaticCriterionSettings.model_dump()
      • StaticCriterionSettings.query
  • tanat.criterion.type.time module
    • TimeCriterion
      • TimeCriterion.LEVELS
      • TimeCriterion.SETTINGS_CLASS
      • TimeCriterion.__init__()
    • TimeCriterionSettings
      • TimeCriterionSettings.__init__()
      • TimeCriterionSettings.all_entities
      • TimeCriterionSettings.duration_within
      • TimeCriterionSettings.end_ge
      • TimeCriterionSettings.end_le
      • TimeCriterionSettings.model_dump()
      • TimeCriterionSettings.start_ge
      • TimeCriterionSettings.start_le
      • TimeCriterionSettings.validate_against()
  • Module contents
    • EntityCriterion
      • EntityCriterion.LEVELS
      • EntityCriterion.SETTINGS_CLASS
      • EntityCriterion.__init__()
    • LengthCriterion
      • LengthCriterion.LEVELS
      • LengthCriterion.SETTINGS_CLASS
      • LengthCriterion.__init__()
    • PatternCriterion
      • PatternCriterion.LEVELS
      • PatternCriterion.SETTINGS_CLASS
      • PatternCriterion.__init__()
    • RankCriterion
      • RankCriterion.LEVELS
      • RankCriterion.SETTINGS_CLASS
      • RankCriterion.__init__()
    • StaticCriterion
      • StaticCriterion.LEVELS
      • StaticCriterion.SETTINGS_CLASS
      • StaticCriterion.__init__()
    • TimeCriterion
      • TimeCriterion.LEVELS
      • TimeCriterion.SETTINGS_CLASS
      • TimeCriterion.__init__()

Submodules

tanat.criterion.base module

Criterion base: abstract class, level enum, and core exceptions.

class tanat.criterion.base.Criterion(settings: Any = None)

Bases: SettingsMixin, Registrable, ABC

Abstract base for all filtering criteria.

Subclasses must declare LEVELS and implement the three _*_impl hooks. All public methods enforce compatibility and delegate to those hooks.

LEVELS: ClassVar[frozenset[CriterionLevel]]

Declare which levels this criterion supports.

ensure_compatible(level: CriterionLevel) → None

Raise CriterionLevelError if level not in LEVELS.

filter_entities(target: Sequence | SequencePool, *, inplace: bool = False, verbose: bool = True)

Return a filtered view at entity level.

Parameters:

• target – A Sequence or SequencePool.

• inplace – If True, modify target in place.

• verbose – If True, emit a one-line report.

Returns:

Filtered target (or target itself when inplace=True).

Raises:

• CriterionLevelError – If ENTITY is not in LEVELS.

• TypeError – If target is not a Sequence or SequencePool.

• CriterionError – If the criterion fails probing.

match(target: Sequence | Trajectory) → bool

Return True if target satisfies this criterion.

Parameters:

target – A Sequence or Trajectory.

Returns:

True when target matches.

Raises:

• TypeError – If target is not a Sequence or Trajectory.

• CriterionLevelError – If the criterion is incompatible with target’s level.

which_ids(pool: SequencePool | TrajectoryPool, *, verbose: bool = False) → set

Return the set of IDs in pool that satisfy this criterion.

Parameters:

• pool – A SequencePool or TrajectoryPool.

• verbose – If True, emit a one-line report.

Returns:

Set of matching IDs.

Raises:

• TypeError – If pool is not a supported pool type.

• CriterionLevelError – If the criterion is incompatible with the pool’s level.

exception tanat.criterion.base.CriterionError

Bases: TanaTException

Raised when a criterion expression is invalid or fails schema probing.

class tanat.criterion.base.CriterionLevel(*values)

Bases: Enum

Compatibility level for a criterion.

ENTITY = 1

SEQUENCE = 2

TRAJECTORY = 3

exception tanat.criterion.base.CriterionLevelError

Bases: TanaTException

Raised when a criterion is applied at an incompatible level.

Module contents

Criterion module: filtering primitives for sequences, entities, and trajectories.

class tanat.criterion.Criterion(settings: Any = None)

Bases: SettingsMixin, Registrable, ABC

Abstract base for all filtering criteria.

Subclasses must declare LEVELS and implement the three _*_impl hooks. All public methods enforce compatibility and delegate to those hooks.

LEVELS: ClassVar[frozenset[CriterionLevel]]

Declare which levels this criterion supports.

ensure_compatible(level: CriterionLevel) → None

Raise CriterionLevelError if level not in LEVELS.

filter_entities(target: Sequence | SequencePool, *, inplace: bool = False, verbose: bool = True)

Return a filtered view at entity level.

Parameters:

• target – A Sequence or SequencePool.

• inplace – If True, modify target in place.

• verbose – If True, emit a one-line report.

Returns:

Filtered target (or target itself when inplace=True).

Raises:

• CriterionLevelError – If ENTITY is not in LEVELS.

• TypeError – If target is not a Sequence or SequencePool.

• CriterionError – If the criterion fails probing.

match(target: Sequence | Trajectory) → bool

Return True if target satisfies this criterion.

Parameters:

target – A Sequence or Trajectory.

Returns:

True when target matches.

Raises:

• TypeError – If target is not a Sequence or Trajectory.

• CriterionLevelError – If the criterion is incompatible with target’s level.

which_ids(pool: SequencePool | TrajectoryPool, *, verbose: bool = False) → set

Return the set of IDs in pool that satisfy this criterion.

Parameters:

• pool – A SequencePool or TrajectoryPool.

• verbose – If True, emit a one-line report.

Returns:

Set of matching IDs.

Raises:

• TypeError – If pool is not a supported pool type.

• CriterionLevelError – If the criterion is incompatible with the pool’s level.

exception tanat.criterion.CriterionError

Bases: TanaTException

Raised when a criterion expression is invalid or fails schema probing.

class tanat.criterion.CriterionLevel(*values)

Bases: Enum

Compatibility level for a criterion.

ENTITY = 1

SEQUENCE = 2

TRAJECTORY = 3

exception tanat.criterion.CriterionLevelError

Bases: TanaTException

Raised when a criterion is applied at an incompatible level.

class tanat.criterion.EntityCriterion(query: Expr)

Bases: Criterion

Filter entities or select sequences using a Polars expression.

Supported levels: ENTITY, SEQUENCE.

Example:

# entity-level pruning (keep only rows where diag_type == "DP")
pool2 = pool.filter_entities(EntityCriterion(query=pl.col("diag_type") == "DP"))

# sequence selection: IDs that have at least one such row
ids = pool.which(EntityCriterion(query=pl.col("diag_type") == "DP"))

# single sequence match
ok = seq.match(EntityCriterion(query=pl.col("diag_type") == "DP"))

LEVELS: ClassVar[frozenset[CriterionLevel]] = frozenset({CriterionLevel.ENTITY, CriterionLevel.SEQUENCE})

Declare which levels this criterion supports.

SETTINGS_CLASS

alias of EntityCriterionSettings

__init__(query: Expr) → None

class tanat.criterion.LengthCriterion(*, gt: int | None = None, ge: int | None = None, lt: int | None = None, le: int | None = None)

Bases: Criterion

Select sequences by their number of entities (rows).

Supported levels: SEQUENCE.

Example:

# sequences with more than 5 entities
ids = pool.which(LengthCriterion(gt=5))
pool2 = pool.subset(ids)

# a single sequence
ok = seq.match(LengthCriterion(ge=3, lt=20))

LEVELS: ClassVar[frozenset[CriterionLevel]] = frozenset({CriterionLevel.SEQUENCE})

Declare which levels this criterion supports.

SETTINGS_CLASS

alias of LengthCriterionSettings

__init__(*, gt: int | None = None, ge: int | None = None, lt: int | None = None, le: int | None = None) → None

class tanat.criterion.PatternCriterion(feature: str, pattern: str | list[str], present: bool = True, regex: bool = True, case_sensitive: bool = True)

Bases: Criterion

Filter entities or sequences by an ordered pattern of string values.

A sequence matches when its entities (in temporal order) contain the given pattern as an ordered sub-sequence: pattern element k must appear after element k-1 in the sequence.

Supported levels: ENTITY, SEQUENCE.

Entity level (filter_entities):

• present=True: keeps only the rows that are “witnesses” of the greedy first match. Sequences without a complete match → 0 rows.

• present=False: keeps all rows that are not witnesses (rows that don’t participate in the pattern). Sequences without a complete match → all their rows are kept.

Sequence level (which, match):

Keeps (or excludes, with present=False) whole sequences based on whether the ordered pattern is found.

Example:

# IDs where "A" appears directly before "B" (adjacent)
ids = pool.which(PatternCriterion(feature="code", pattern=["A", "B"]))

# Entity pruning: keep only the matched witness rows
pool2 = pool.filter_entities(
    PatternCriterion(feature="code", pattern=["A", "B"])
)

# Free gap: A before B with any rows in between
ids = pool.which(
    PatternCriterion(feature="code", pattern=["A", ANY, "B"])
)

# Exactly one element between A and B
ids = pool.which(
    PatternCriterion(feature="code", pattern=["A", WILDCARD, "B"])
)

# Single-element pattern: at least one row matching "ICU"
ids = pool.which(PatternCriterion(feature="label", pattern="ICU"))

# Exclusion: sequences that never contain adjacent A→B
ids = pool.which(
    PatternCriterion(feature="code", pattern=["A", "B"], present=False)
)

# Literal, case-insensitive
ids = pool.which(
    PatternCriterion(feature="code", pattern="icu", regex=False, case_sensitive=False)
)

# Single-sequence match
ok = seq.match(PatternCriterion(feature="code", pattern=["A", "B"]))

LEVELS: ClassVar[frozenset[CriterionLevel]] = frozenset({CriterionLevel.ENTITY, CriterionLevel.SEQUENCE})

Declare which levels this criterion supports.

SETTINGS_CLASS

alias of PatternCriterionSettings

__init__(feature: str, pattern: str | list[str], present: bool = True, regex: bool = True, case_sensitive: bool = True) → None

class tanat.criterion.RankCriterion(*, first: int | None = None, last: int | None = None, start: int | None = None, end: int | None = None, step: int | None = None, ranks: list[int] | int | None = None, relative: bool = False)

Bases: Criterion

Select entities by their positional rank within a sequence.

Supported levels: ENTITY only.

Entities are numbered 0-based within each sequence in their natural store order. Negative indices use Python-style semantics (from the end).

Example:

# keep the first 3 entities
pool2 = pool.filter_entities(RankCriterion(first=3))

# keep all except the last 2 entities
pool2 = pool.filter_entities(RankCriterion(first=-2))

# keep the last 2 entities
pool2 = pool.filter_entities(RankCriterion(last=2))

# keep all except the first 3 entities
pool2 = pool.filter_entities(RankCriterion(last=-3))

# keep entities at ranks 2, 3, 4 (Python slice semantics)
pool2 = pool.filter_entities(RankCriterion(start=2, end=5))

# keep from rank 5 to 2nd-from-end
pool2 = pool.filter_entities(RankCriterion(start=5, end=-2))

# keep every other entity
pool2 = pool.filter_entities(RankCriterion(step=2))

# keep specific ranks (first and last)
pool2 = pool.filter_entities(RankCriterion(ranks=[0, -1]))

# keep the 3 entities centered on T0 (requires set_t0())
pool2 = pool.filter_entities(RankCriterion(start=-1, end=2, relative=True))

LEVELS: ClassVar[frozenset[CriterionLevel]] = frozenset({CriterionLevel.ENTITY})

Declare which levels this criterion supports.

SETTINGS_CLASS

alias of RankCriterionSettings

__init__(*, first: int | None = None, last: int | None = None, start: int | None = None, end: int | None = None, step: int | None = None, ranks: list[int] | int | None = None, relative: bool = False) → None

class tanat.criterion.StaticCriterion(query: Expr)

Bases: Criterion

Select sequences or trajectories using a static-feature expression.

Supported levels: SEQUENCE, TRAJECTORY.

Example:

# sequence pool: keep IDs where age > 50
ids = seq_pool.which(StaticCriterion(query=pl.col("age") > 50))
seq_pool2 = seq_pool.subset(ids)

# trajectory pool
ids = traj_pool.which(StaticCriterion(query=pl.col("group") == "A"))
traj_pool2 = traj_pool.subset(ids)

# single match
ok = seq.match(StaticCriterion(query=pl.col("age") > 50))
ok = traj.match(StaticCriterion(query=pl.col("group") == "A"))

LEVELS: ClassVar[frozenset[CriterionLevel]] = frozenset({CriterionLevel.SEQUENCE, CriterionLevel.TRAJECTORY})

Declare which levels this criterion supports.

SETTINGS_CLASS

alias of StaticCriterionSettings

__init__(query: Expr) → None

class tanat.criterion.TimeCriterion(*, start_ge: datetime | date | int | float | None = None, start_le: datetime | date | int | float | None = None, end_ge: datetime | date | int | float | None = None, end_le: datetime | date | int | float | None = None, duration_within: bool = False, all_entities: bool = False)

Bases: Criterion

Filter entities or select sequences by temporal position.

Supported levels: ENTITY, SEQUENCE.

Example:

import datetime as dt
from tanat.criterion import TimeCriterion

t0 = dt.datetime(2020, 1, 1)
t1 = dt.datetime(2021, 1, 1)

# entity pruning: keep rows whose start time is in [t0, t1]
pool2 = pool.filter_entities(TimeCriterion(start_ge=t0, start_le=t1))

# entity pruning: interval must be fully contained in [t0, t1]
pool3 = pool.filter_entities(
    TimeCriterion(start_ge=t0, end_le=t1, duration_within=True)
)

# sequence selection: IDs with at least one row in the window (default)
ids = pool.which(TimeCriterion(start_ge=t0))

# sequence selection: IDs where ALL rows are in the window
ids = pool.which(TimeCriterion(start_ge=t0, end_le=t1, all_entities=True))

# match
ok = seq.match(TimeCriterion(start_le=t1))

LEVELS: ClassVar[frozenset[CriterionLevel]] = frozenset({CriterionLevel.ENTITY, CriterionLevel.SEQUENCE})

Declare which levels this criterion supports.

SETTINGS_CLASS

alias of TimeCriterionSettings

__init__(*, start_ge: datetime | date | int | float | None = None, start_le: datetime | date | int | float | None = None, end_ge: datetime | date | int | float | None = None, end_le: datetime | date | int | float | None = None, duration_within: bool = False, all_entities: bool = False) → None