tanat.metric.trajectory package

Subpackages

• tanat.metric.trajectory.type package
  • Subpackages
    • tanat.metric.trajectory.type.aggregation package
      • Submodules
      • tanat.metric.trajectory.type.aggregation.metric module
      • Module contents
  • Module contents

Submodules

tanat.metric.trajectory.base module

TrajectoryMetric ABC: base class for all trajectory-level distance metrics.

class tanat.metric.trajectory.base.TrajectoryMetric(settings=None, storage: StorageOptions | dict | None = None)

Bases: SettingsMixin, Registrable, DisplayMixin, ABC

Abstract base for trajectory-level distance metrics.

Computes a scalar distance between two Trajectory objects and a full pairwise DistanceMatrix over a TrajectoryPool.

MEMMAP_SUPPORT: bool = False

Set to True in subclasses that implement disk-backed (memmap) computation.

__init__(settings=None, storage: StorageOptions | dict | None = None) → None

compute_cross_matrix(pool_rows: TrajectoryPool, pool_cols: TrajectoryPool) → ndarray

Compute an asymmetric (n × k) distance matrix between two pools.

Row i ↔ trajectory i in pool_rows; column j ↔ trajectory j in pool_cols.

Validates both pools, then delegates to _compute_cross_matrix_impl(). Subclasses override _compute_cross_matrix_impl() to use optimised kernels.

Parameters:

• pool_rows – Pool whose trajectories form the rows (n items).

• pool_cols – Pool whose trajectories form the columns (k items).

Returns:

float32 numpy array of shape (n, k).

compute_matrix(pool: TrajectoryPool, *, store_path: str | Path | None = None, chunk_size: int = 500, resume: bool = True, dtype: str = 'float32') → DistanceMatrix

Compute full pairwise trajectory distance matrix.

Storage kwargs are forwarded to StorageOptions.

Parameters:

• pool – A TrajectoryPool.

• store_path – Storage directory (None → in-memory).

• chunk_size – Rows per flush chunk (default 500).

• resume – Skip already-computed chunks (default True).

• dtype – Numpy dtype for the matrix (default "float32").

Returns:

A DistanceMatrix.

Module contents

Trajectory metric sub-package.

class tanat.metric.trajectory.AggregationSettings(*, default_metric: SequenceMetric = 'linearpairwise', sequence_metrics: dict[str, SequenceMetric] | None = None, agg_fun: str = 'mean', weights: dict[str, float] | None = None)

Bases: object

Settings for AggregationTrajectoryMetric.

agg_fun and weights are orthogonal: "mean" + weights computes a weighted mean, "sum" + weights a weighted sum. Aliases absent from weights default to 1.0.

__init__(*args: Any, **kwargs: Any) → None

agg_fun: str = 'mean'

default_metric: SequenceMetric = 'linearpairwise'

model_dump(*, mode='python', **dump_kwargs)

Dump settings to a dict via Pydantic serialization.

sequence_metrics: dict[str, SequenceMetric] | None = None

weights: dict[str, float] | None = None

class tanat.metric.trajectory.AggregationTrajectoryMetric(default_metric: SequenceMetric | str = 'linearpairwise', sequence_metrics: dict[str, SequenceMetric | str] | None = None, agg_fun: str = 'mean', weights: dict[str, float] | None = None, *, store_path: str | Path | None = None, chunk_size: int = 500, resume: bool = True, dtype: str = 'float32')

Bases: TrajectoryMetric

Trajectory distance by per-alias sequence distances, then weighted aggregation.

For each store alias visible on both trajectories, computes the sequence-level distance using the configured SequenceMetric. The resulting per-alias distances are aggregated (weighted mean/sum) into a scalar trajectory distance.

Example:

hamming = HammingEntityMetric(entity_feature="status")
lp = LinearPairwiseSequenceMetric(entity_metric=hamming)

agg = AggregationTrajectoryMetric(
    default_metric=lp,
    agg_fun="mean",
    weights={"events": 1.0, "states": 0.5},
)

dist = agg(traj_a, traj_b)
dm   = agg.compute_matrix(traj_pool)

MEMMAP_SUPPORT: bool = True

Set to True in subclasses that implement disk-backed (memmap) computation.

SETTINGS_CLASS

alias of AggregationSettings

__init__(default_metric: SequenceMetric | str = 'linearpairwise', sequence_metrics: dict[str, SequenceMetric | str] | None = None, agg_fun: str = 'mean', weights: dict[str, float] | None = None, *, store_path: str | Path | None = None, chunk_size: int = 500, resume: bool = True, dtype: str = 'float32') → None

class tanat.metric.trajectory.TrajectoryMetric(settings=None, storage: StorageOptions | dict | None = None)

Bases: SettingsMixin, Registrable, DisplayMixin, ABC

Abstract base for trajectory-level distance metrics.

Computes a scalar distance between two Trajectory objects and a full pairwise DistanceMatrix over a TrajectoryPool.

MEMMAP_SUPPORT: bool = False

Set to True in subclasses that implement disk-backed (memmap) computation.

__init__(settings=None, storage: StorageOptions | dict | None = None) → None

compute_cross_matrix(pool_rows: TrajectoryPool, pool_cols: TrajectoryPool) → ndarray

Compute an asymmetric (n × k) distance matrix between two pools.

Row i ↔ trajectory i in pool_rows; column j ↔ trajectory j in pool_cols.

Validates both pools, then delegates to _compute_cross_matrix_impl(). Subclasses override _compute_cross_matrix_impl() to use optimised kernels.

Parameters:

• pool_rows – Pool whose trajectories form the rows (n items).

• pool_cols – Pool whose trajectories form the columns (k items).

Returns:

float32 numpy array of shape (n, k).

compute_matrix(pool: TrajectoryPool, *, store_path: str | Path | None = None, chunk_size: int = 500, resume: bool = True, dtype: str = 'float32') → DistanceMatrix

Compute full pairwise trajectory distance matrix.

Storage kwargs are forwarded to StorageOptions.

Parameters:

• pool – A TrajectoryPool.

• store_path – Storage directory (None → in-memory).

• chunk_size – Rows per flush chunk (default 500).

• resume – Skip already-computed chunks (default True).

• dtype – Numpy dtype for the matrix (default "float32").

Returns:

A DistanceMatrix.