"""Label records: validated integer labels, explicit identities and uniform sampling segments.
A :class:`LabelRecord` is one tracked trajectory of per-frame group labels for a fixed set of
entities (for example atoms). ``labels[t, i]`` is the label of entity ``i`` at frame ``t``;
``-1`` means rejected/abstained, every other value is an accepted group id ``>= 0``.
Sampling contract
-----------------
* Frames are the time unit of every duration. ``frame_interval``/``time_unit`` are metadata
for conversion and for refusing to pool records sampled differently.
* ``segments`` are explicit, sorted, non-overlapping ``[start, stop)`` row ranges of uniformly
sampled consecutive frames. Rows outside every segment are unobserved. A segment boundary
(a gap, a restart, dropped frames) censors both sides: runs never cross it.
* Optional ``timestamps`` are validated, never used to invent durations: within a segment the
step must equal ``frame_interval`` (exactly, or within an explicit ``timestamp_rtol``) and
timestamps must increase strictly across the whole record. Anything else is rejected.
* The early/late split is always an explicit argument; there is no default split.
* Two records are independent: there is no concatenation API, and the same entity id in two
records is never treated as the same tracked identity (see :func:`runs.pool_runs`).
Identities (kept separate)
--------------------------
* **Source population** -- WHAT was observed: an optional caller-declared
:class:`SourceSelection` (an opaque ``source_id`` of the physical record, e.g. a trajectory
name plus its file hash, and the affine frame map ``source frame = frame_start + frame_step *
row``) together with the entity ids. It is never inferred from shapes, labels, record ids or
file paths; undeclared records have ``population_key is None``.
* **Sampling/layout** -- :attr:`LabelRecord.layout_key`: frame count, entity order, segments,
sampling metadata. Two different trajectories (or two slices of one) can share a layout.
* **Content** -- :attr:`LabelRecord.fingerprint` (labels included): integrity of early/late
halves only. Equal content never proves the same population, and distinct candidates of
one population have different labels.
"""
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import math
import numpy as np
REJECTED = -1
_INT64_MAX = np.iinfo(np.int64).max
[docs]
class SamplingError(ValueError):
"""Irregular, ambiguous or unsupported sampling metadata."""
def _as_window(window, n_frames):
try:
bounds = (window[0], window[1])
start, stop = (int(bounds[0]), int(bounds[1]))
except (TypeError, ValueError, IndexError) as exc:
raise ValueError(f'window must be a (start, stop) pair of ints, got {window!r}') from exc
if any(isinstance(x, (bool, np.bool_)) for x in bounds):
raise ValueError(f'window bounds must be integers, not booleans; got {window!r}')
if (start, stop) != (window[0], window[1]):
raise ValueError(f'window bounds must be integers, got {window!r}')
if not 0 <= start < stop <= n_frames:
raise ValueError(f'window {window!r} is not a non-empty range inside [0, {n_frames})')
return start, stop
def _read_only_copy(a):
out = np.array(a) # always a copy: later changes to the caller's array cannot bypass validation
out.flags.writeable = False
return out
def _validate_labels(labels, n_groups):
lab = _read_only_copy(labels)
if lab.ndim != 2:
raise ValueError(f'labels must be 2-D (frames, entities); got shape {lab.shape}')
if lab.dtype.kind not in 'iu':
raise TypeError(f'labels must have an integer dtype; got {lab.dtype}')
if lab.shape[0] == 0 or lab.shape[1] == 0:
raise ValueError(f'labels must have at least one frame and one entity; got shape {lab.shape}')
lo = int(lab.min())
hi = int(lab.max())
if lo < REJECTED:
raise ValueError(f'labels must be -1 (rejected) or >= 0; found {lo}')
if hi > _INT64_MAX:
raise ValueError('labels must fit in int64')
if n_groups is not None:
if int(n_groups) != n_groups or n_groups < 1:
raise ValueError(f'n_groups must be a positive int, got {n_groups!r}')
if hi >= n_groups:
raise ValueError(f'label {hi} is outside [0, n_groups={n_groups})')
return lab
def _validate_segments(segments, n_frames):
if segments is None:
return ((0, n_frames),)
out = []
for seg in segments:
out.append(_as_window(seg, n_frames))
if not out:
raise ValueError('segments must contain at least one (start, stop) range')
for (a0, a1), (b0, b1) in zip(out, out[1:]):
if b0 < a1:
raise SamplingError(f'segments must be sorted and non-overlapping; ({a0}, {a1}) then ({b0}, {b1})')
return tuple(out)
def _validate_timestamps(timestamps, n_frames, segments, frame_interval, rtol):
ts = _read_only_copy(timestamps)
if ts.shape != (n_frames,):
raise ValueError(f'timestamps must have shape ({n_frames},); got {ts.shape}')
if ts.dtype.kind not in 'iuf':
raise TypeError(f'timestamps must be numeric; got {ts.dtype}')
exact = ts.dtype.kind in 'iu' and float(frame_interval).is_integer() and rtol == 0
covered = np.concatenate([np.arange(a, b) for a, b in segments])
if ts.dtype.kind == 'f' and not np.all(np.isfinite(ts[covered])):
raise SamplingError('timestamps of observed rows must be finite')
for k, (a, b) in enumerate(segments):
if b - a < 2:
continue
if exact:
step = np.diff(ts[a:b].astype(object)) if ts.dtype == np.uint64 else np.diff(ts[a:b].astype(np.int64))
bad = np.flatnonzero(step != int(frame_interval))
else:
step = np.diff(ts[a:b].astype(np.float64))
bad = np.flatnonzero(np.abs(step - float(frame_interval)) > rtol * float(frame_interval))
if bad.size:
i = int(bad[0])
raise SamplingError(
f'irregular sampling in segment {k} ({a}, {b}): timestamp step {step[i]!r} at row {a + i + 1} '
f'!= frame_interval {frame_interval!r}; declare an explicit segment boundary instead of '
'interpolating or inventing elapsed durations')
for k in range(1, len(segments)):
prev_last = ts[segments[k - 1][1] - 1]
first = ts[segments[k][0]]
if not first > prev_last:
raise SamplingError(
f'timestamps are not strictly increasing across segments {k - 1} and {k} '
f'({prev_last!r} then {first!r}): overlapping restarts are ambiguous')
return ts
[docs]
@dataclass(frozen=True)
class SourceSelection:
"""Declared source population of a record: which physical record, which of its frames.
``source_id`` is an opaque, caller-declared identity of the physical record (for example
``'run7/traj.xyz#sha256=...'``); it is never a candidate id or a path chosen by this
package. Row ``r`` of the label array is source frame ``frame_start + frame_step * r``.
Only affine selections are expressible: a non-affine selection needs its own ``source_id``,
and overlap between differently declared sources cannot be detected.
"""
source_id: str
frame_start: int = 0
frame_step: int = 1
def __post_init__(self):
if not isinstance(self.source_id, str) or not self.source_id:
raise ValueError(f'source_id must be a non-empty string, got {self.source_id!r}')
for name, low in (('frame_start', 0), ('frame_step', 1)):
v = getattr(self, name)
if isinstance(v, (bool, np.bool_)) or not isinstance(v, (int, np.integer)) or int(v) < low:
raise ValueError(f'{name} must be an int >= {low}, got {v!r}')
object.__setattr__(self, name, int(v))
[docs]
def frames(self, rows):
"""Source frame numbers of label rows ``rows`` (int64)."""
return self.frame_start + self.frame_step * np.asarray(rows, np.int64)
[docs]
def as_dict(self):
return {'source_id': self.source_id, 'frame_start': self.frame_start, 'frame_step': self.frame_step}
def _canonical_ids(ids):
return [x.item() if isinstance(x, np.generic) else x for x in np.asarray(ids).tolist()]
def _key(payload):
import json
return hashlib.sha256(json.dumps(payload, separators=(',', ':'), sort_keys=True).encode()).hexdigest()
[docs]
@dataclass(frozen=True, eq=False)
class LabelRecord:
"""One validated label record; construct with :meth:`from_labels`.
:meth:`from_labels` stores read-only COPIES of labels, identities and timestamps, so changing
the caller's arrays afterwards cannot bypass validation. Direct ``LabelRecord(...)``
construction is internal: it neither validates nor copies.
"""
labels: np.ndarray
entity_ids: np.ndarray
segments: tuple
frame_interval: float
time_unit: str
record_id: str
timestamps: np.ndarray | None = None
fingerprint: str | None = None
source: SourceSelection | None = None
[docs]
@classmethod
def from_labels(cls, labels, *, entity_ids=None, segments=None, frame_interval=1, time_unit='frame',
timestamps=None, timestamp_rtol=0.0, record_id='record', n_groups=None, source=None):
"""Validate and wrap ``labels`` (frames x entities, integer, -1 = rejected).
``entity_ids`` are the original identities (default ``0..N-1``), integer or unicode
strings, and must be unique. ``segments`` defaults to one segment spanning every row.
``timestamps`` (optional) are checked against ``frame_interval``; ``timestamp_rtol`` is
an explicit tolerance for floating-point timestamps (default exact). ``n_groups``
(optional) bounds the labels. ``source`` (optional :class:`SourceSelection`) declares
the observed source population; it does not change :attr:`fingerprint`.
"""
lab = _validate_labels(labels, n_groups)
n_frames, n_entities = lab.shape
if entity_ids is None:
ids = np.arange(n_entities, dtype=np.int64)
else:
ids = _read_only_copy(entity_ids)
if ids.dtype.kind not in 'iuU':
raise TypeError(f'entity_ids must be integers or unicode strings, got dtype {ids.dtype}; convert '
'them explicitly (e.g. np.asarray(ids, dtype=np.int64) or dtype=str)')
if ids.shape != (n_entities,):
raise ValueError(f'entity_ids must have shape ({n_entities},); got {ids.shape}')
if len(np.unique(ids)) != n_entities:
raise ValueError('entity_ids must be unique: a repeated id would make identity ambiguous')
ids.flags.writeable = False
if isinstance(frame_interval, bool) or not isinstance(frame_interval, (int, float, np.integer, np.floating)):
raise TypeError(f'frame_interval must be a real number, got {frame_interval!r}')
if not (math.isfinite(float(frame_interval)) and float(frame_interval) > 0):
raise SamplingError(f'frame_interval must be positive and finite, got {frame_interval!r}')
if not isinstance(time_unit, str) or not time_unit:
raise ValueError('time_unit must be a non-empty string')
if not isinstance(record_id, str) or not record_id:
raise ValueError('record_id must be a non-empty string')
if not (timestamp_rtol >= 0 and math.isfinite(timestamp_rtol)):
raise ValueError('timestamp_rtol must be a finite non-negative number')
if source is not None and not isinstance(source, SourceSelection):
raise TypeError(f'source must be a SourceSelection or None, got {type(source).__name__}')
segs = _validate_segments(segments, n_frames)
ts = None
if timestamps is not None:
ts = _validate_timestamps(timestamps, n_frames, segs, frame_interval, float(timestamp_rtol))
return cls(lab, ids, segs, frame_interval, time_unit, record_id, ts,
_fingerprint(lab, ids, segs, frame_interval, time_unit, ts), source)
@property
def n_frames(self):
return int(self.labels.shape[0])
@property
def n_entities(self):
return int(self.labels.shape[1])
@property
def population_key(self):
"""Hash of the declared source population (source id, frame map, frame count, entity
ids), or ``None`` when no :class:`SourceSelection` was declared. Labels are excluded."""
if self.source is None:
return None
return _key({'source': self.source.as_dict(), 'n_frames': self.n_frames,
'entity_ids': _canonical_ids(self.entity_ids)})
@property
def layout_key(self):
"""Hash of the sampling/layout (frame count, entity order, segments, frame interval, time
unit, timestamps). Equal layouts do NOT imply the same population."""
payload = {'n_frames': self.n_frames, 'entity_ids': _canonical_ids(self.entity_ids),
'segments': [list(s) for s in self.segments], 'frame_interval': repr(float(self.frame_interval)),
'time_unit': self.time_unit}
if self.timestamps is not None:
ts = np.ascontiguousarray(self.timestamps)
payload['timestamps_sha256'] = hashlib.sha256(ts.dtype.str.encode() + ts.tobytes()).hexdigest()
return _key(payload)
[docs]
def full_window(self):
"""The whole row range ``(0, n_frames)``."""
return (0, self.n_frames)
[docs]
def check_window(self, window):
"""Validated ``(start, stop)`` row window inside the record."""
return _as_window(window, self.n_frames)
[docs]
def half_windows(self, split):
"""``((0, split), (split, n_frames))`` for an explicit split frame (no default)."""
if isinstance(split, bool) or int(split) != split:
raise ValueError(f'split must be an integer frame index, got {split!r}')
split = int(split)
if not 0 < split < self.n_frames:
raise ValueError(f'split {split} must lie strictly inside (0, {self.n_frames})')
return (0, split), (split, self.n_frames)
[docs]
def observed_intervals(self, window):
"""Non-empty ``segment ∩ window`` pieces, shape (k, 2), in time order.
Every piece boundary censors: a run touching one is left/right-censored.
"""
start, stop = self.check_window(window)
out = [(max(a, start), min(b, stop)) for a, b in self.segments if max(a, start) < min(b, stop)]
return np.asarray(out, dtype=np.int64).reshape(-1, 2)
[docs]
def observed_frames(self, window):
"""Number of observed frames per entity inside ``window``."""
iv = self.observed_intervals(window)
return int(np.sum(iv[:, 1] - iv[:, 0]))
def _fingerprint(labels, ids, segments, frame_interval, time_unit, timestamps):
"""Content identity of a record (labels, identities, segments, sampling), independent of
the caller-chosen ``record_id``: two records are halves of one record only if this matches."""
h = hashlib.sha256()
for a in (labels, ids) + (() if timestamps is None else (timestamps,)):
a = np.ascontiguousarray(a)
h.update(a.dtype.str.encode() + repr(a.shape).encode())
h.update(a.view(np.uint8).tobytes())
h.update(repr((tuple(tuple(int(x) for x in s) for s in segments), float(frame_interval), str(time_unit))).encode())
return h.hexdigest()