"""Literal runs/visits, exclusive censoring, one-pass bridge1, accounting and a lossless runpack.
A run is a maximal contiguous sequence of one label for one entity inside one observed piece
(``segment ∩ window``), rejected runs included. Coordinates are absolute row indices with an
EXCLUSIVE exit, so ``duration = exit - entry``. A run touching a piece boundary (window edge or
segment boundary) is left- and/or right-censored; only accepted runs with neither flag are
complete visits. Ported from the accepted v2 ``extract_runs``/``bridge1``/``accounting``/
``pack_runs`` and the PF ``stats_from_runs`` (same values and run order; vectorised over
entities).
"""
from __future__ import annotations
import math
from dataclasses import dataclass, field
import numpy as np
from .record import SourceSelection
COMPLETE, LEFT_ONLY, RIGHT_ONLY, BOTH_CENSORED = 0, 1, 2, 3
CENSORING_CLASSES = ('complete', 'left_only', 'right_only', 'both_censored')
PREPROCESSING = ('raw', 'bridge1')
BRIDGE_FIELDS = ('frames_changed', 'accepted_to_accepted_changes', 'rejected_to_accepted_changes',
'overlapping_candidate_pairs')
FULL_SCOPE, HALF_SCOPE = 'full_primary', 'secondary_half'
[docs]
@dataclass(frozen=True)
class RunSource:
"""Where a block of runs came from: one record and one window.
``record_frames`` is the record's frame count (its full window is ``(0, record_frames)``);
``None`` when unknown, e.g. for a run table rebuilt from an archived v2 runpack. ``source``
is the record's declared :class:`~.record.SourceSelection` (``None`` when undeclared).
"""
record_id: str
window: tuple
intervals: np.ndarray = field(repr=False)
entity_offset: int
n_entities: int
frame_interval: float
time_unit: str
record_frames: int | None = None
record_fingerprint: str | None = None
source: SourceSelection | None = None
@property
def observed_frames(self):
return int(np.sum(self.intervals[:, 1] - self.intervals[:, 0]))
[docs]
@dataclass(frozen=True)
class BridgeChanges:
"""Change categories of one bridge1 application (counts over all entities)."""
frames_changed: int
accepted_to_accepted_changes: int
rejected_to_accepted_changes: int
overlapping_candidate_pairs: int
[docs]
@dataclass(frozen=True, eq=False)
class Runs:
"""Run table (v2 order: entity-major, then entry) plus provenance.
``entity`` indexes ``entity_ids``; ``source`` indexes ``sources`` (one per record/window).
``bridge``/``bridge_mask`` exist for ``preprocessing='bridge1'``: the mask marks changed
labels over the observed rows of the window stacked in time order (frames x entities).
``bridge is None`` for a bridge1 table means the change counts are UNKNOWN (never zero).
``qualified_ids`` is True when ``entity_ids`` are already ``(record_id, entity_id)`` pairs
(output of :func:`pool_runs`). ``notes`` carries pooling remarks (never a silent change).
Tables from :func:`extract_runs` are maximal by construction and :func:`unpack_runs` validates
encoded (untrusted) run data; a ``Runs`` constructed directly is trusted as given.
"""
group: np.ndarray
duration: np.ndarray
entity: np.ndarray
entry: np.ndarray
exit: np.ndarray
left_censored: np.ndarray
right_censored: np.ndarray
entry_bin: np.ndarray
source: np.ndarray
sources: tuple
entity_ids: np.ndarray
n_bins: int
preprocessing: str
bridge: BridgeChanges | None = None
bridge_mask: np.ndarray | None = field(default=None, repr=False)
qualified_ids: bool = False
notes: tuple = ()
@property
def n_entities(self):
return int(sum(s.n_entities for s in self.sources))
def __len__(self):
return int(len(self.group))
[docs]
def complete(self):
"""Accepted runs with observed entry and exit inside their piece."""
return (self.group >= 0) & ~self.left_censored & ~self.right_censored
[docs]
def censoring_class(self):
"""Mutually exclusive class per run: -1 rejected, else COMPLETE/LEFT_ONLY/RIGHT_ONLY/BOTH_CENSORED."""
cls = np.full(len(self.group), -1, np.int8)
acc = self.group >= 0
cls[acc] = (self.left_censored[acc].astype(np.int8) + 2 * self.right_censored[acc].astype(np.int8))
return cls
[docs]
def bridge1(block):
"""v2 one-pass simultaneous bridge on a contiguous block (frames x entities).
For interior t, the label becomes A iff the ORIGINAL neighbours at t-1 and t+1 both equal an
accepted A and the original middle differs. Returns ``(bridged, change)`` with ``change`` of
shape ``(len(block) - 2, entities)`` (empty if the block has fewer than three frames).
"""
src = np.asarray(block)
out = src.copy()
if len(src) < 3:
return out, np.zeros((0,) + src.shape[1:], bool)
change = (src[:-2] == src[2:]) & (src[:-2] >= 0) & (src[1:-1] != src[:-2])
out[1:-1][change] = src[:-2][change]
return out, change
def _bridge_pieces(x, piece_starts, piece_stops):
"""bridge1 applied independently inside each piece of the stacked rows ``x``."""
out = x.copy()
mask = np.zeros(x.shape, bool)
counts = np.zeros(4, np.int64)
for p0, p1 in zip(piece_starts.tolist(), piece_stops.tolist()):
src = x[p0:p1]
bridged, change = bridge1(src)
if not change.size:
continue
out[p0:p1] = bridged
mask[p0 + 1:p1 - 1] = change
mid = src[1:-1]
counts += (int(change.sum()), int(np.count_nonzero(change & (mid >= 0))),
int(np.count_nonzero(change & (mid < 0))), int(np.count_nonzero(change[:-1] & change[1:])))
return out, mask, BridgeChanges(*(int(c) for c in counts))
def _empty_runs(dtype, sources, entity_ids, n_bins, preprocessing):
z = np.zeros(0, np.int64)
f = np.zeros(0, bool)
return Runs(np.zeros(0, dtype), z, z, z, z, f, f, z, z, sources, entity_ids, n_bins, preprocessing,
BridgeChanges(0, 0, 0, 0) if preprocessing == 'bridge1' else None)
def _source_entities(table, src):
ids = table.entity_ids[src.entity_offset:src.entity_offset + src.n_entities]
raw = [e[1] if table.qualified_ids else e for e in ids.tolist()]
return {x.item() if isinstance(x, np.generic) else x for x in raw}
def _source_frames(src):
rows = np.concatenate([np.arange(a, b) for a, b in src.intervals.tolist()]) if len(src.intervals) else \
np.zeros(0, np.int64)
return src.source.frames(rows)
def _pool_checks(tables):
"""Alias refusal and identical-content notes over every pair of sources from DIFFERENT input
tables (pairs inside an already pooled table were checked when it was pooled)."""
flat = [(k, t, s) for k, t in enumerate(tables) for s in t.sources]
notes = [n for t in tables for n in t.notes]
for i in range(len(flat)):
ki, ti, si = flat[i]
for j in range(i + 1, len(flat)):
kj, tj, sj = flat[j]
if ki == kj:
continue
if si.source is not None and sj.source is not None and si.source.source_id == sj.source.source_id:
ents = _source_entities(ti, si) & _source_entities(tj, sj)
if ents and len(np.intersect1d(_source_frames(si), _source_frames(sj))):
raise ValueError(
f'pool_runs refuses records {si.record_id!r} and {sj.record_id!r}: both declare source '
f'{si.source.source_id!r} and their observed source frames and entities overlap, so the same '
'source observations under two aliases would be counted twice')
same = (si.record_fingerprint is not None and si.record_fingerprint == sj.record_fingerprint
and tuple(si.window) == tuple(sj.window))
if same and (si.source is None or sj.source is None):
notes.append(f'records {si.record_id!r} and {sj.record_id!r} have identical content (labels, ids, '
'sampling) but no declared distinct sources; both are kept as distinct records')
return tuple(notes)
[docs]
def pool_runs(tables):
"""Explicitly pool run tables of DIFFERENT records without linking identities.
Entities are re-indexed per record (``(record_id, entity_id)`` identities, qualified once:
re-pooling a pooled table keeps its pairs), no run is joined across records, and records must
share ``frame_interval``/``time_unit``/``preprocessing``/``n_bins``; duplicate record ids are
refused as ambiguous. Pooled bridge1 change counts are unknown if any input's are.
Duplicates are defined by declared provenance, never by content: two records declaring the
same ``source_id`` whose observed source frames and entities intersect are refused (the same
observations under two aliases). Equal label content is NOT duplication -- distinct records
may legitimately coincide -- so such records are kept, and a remark is added to ``notes``
when at least one of them has no declared source.
"""
tables = list(tables)
if not tables:
raise ValueError('pool_runs needs at least one run table')
first = tables[0]
srcs = [s for t in tables for s in t.sources]
if len({s.record_id for s in srcs}) != len(srcs):
raise ValueError('pool_runs refuses repeated record ids: the same record/window would be double counted '
'and identical ids would make identity ambiguous')
if len({(s.frame_interval, s.time_unit) for s in srcs}) != 1:
raise ValueError('pool_runs refuses records with different sampling intervals or time units')
if len({(t.preprocessing, t.n_bins) for t in tables}) != 1:
raise ValueError('pool_runs refuses tables with different preprocessing or n_bins')
notes = _pool_checks(tables)
sources, ids, parts, offset = [], [], [], 0
for t in tables:
base = len(sources)
for s in t.sources:
sources.append(RunSource(s.record_id, s.window, s.intervals, s.entity_offset + offset, s.n_entities,
s.frame_interval, s.time_unit, s.record_frames, s.record_fingerprint, s.source))
for i in range(s.n_entities):
eid = t.entity_ids[s.entity_offset + i]
eid = eid.item() if isinstance(eid, np.generic) else eid
ids.append(eid if t.qualified_ids else (s.record_id, eid))
parts.append((t, base, offset))
offset += t.n_entities
ident = np.empty(len(ids), dtype=object)
ident[:] = ids
cat = lambda name: np.concatenate([getattr(t, name) for t, _, _ in parts])
bridge = None
if first.preprocessing == 'bridge1' and all(t.bridge is not None for t in tables):
bridge = BridgeChanges(*(sum(getattr(t.bridge, f) for t, _, _ in parts) for f in BRIDGE_FIELDS))
return Runs(group=cat('group'), duration=cat('duration'),
entity=np.concatenate([t.entity + off for t, _, off in parts]),
entry=cat('entry'), exit=cat('exit'), left_censored=cat('left_censored'),
right_censored=cat('right_censored'), entry_bin=cat('entry_bin'),
source=np.concatenate([t.source + base for t, base, _ in parts]), sources=tuple(sources),
entity_ids=ident, n_bins=first.n_bins, preprocessing=first.preprocessing, bridge=bridge,
qualified_ids=True, notes=notes)
[docs]
def native_acceptance(record):
"""``(accepted, observed)`` raw entity-frame counts over the record's full window.
``accepted`` counts labels ``>= 0`` on observed rows (every segment), ``observed`` all
observed entity-frames; identical to the ``accepted_occupancy_frames`` and
``total_entity_frames`` of :func:`accounting` on the raw full-window runs. Not reduced, so
the exact acceptance is ``Fraction(accepted, observed)``. Preprocessing never changes it.
"""
accepted = sum(int(np.count_nonzero(record.labels[a:b] >= 0)) for a, b in record.segments)
observed = sum(b - a for a, b in record.segments) * record.n_entities
return accepted, observed
def _derived_scope(runs):
"""Sample scope from the windows themselves (never from a window NAME), or None if not derivable.
Every source spans its record's full window -> ``'full_primary'``; every source is one half
``(0, s)``/``(s, n_frames)`` of its record -> ``'secondary_half'``; anything else (an interior
window, mixed sources, an unknown record length) has no derivable scope.
"""
scopes = set()
for s in runs.sources:
n = s.record_frames
start, stop = (int(x) for x in s.window)
if n is None:
return None
scopes.add(FULL_SCOPE if (start, stop) == (0, n) else HALF_SCOPE if start == 0 or stop == n else None)
return scopes.pop() if len(scopes) == 1 else None
def _metadata(runs, resolution, window_name, sample_scope):
scope = sample_scope if sample_scope is not None else _derived_scope(runs)
if scope is None:
raise ValueError(f'sample_scope of window {window_name!r} cannot be derived (not the full record window '
'or one half of it, mixed sources, or unknown record length); pass sample_scope explicitly')
return {'resolution': resolution, 'preprocessing': runs.preprocessing, 'window': window_name, 'sample_scope': scope}
[docs]
def accounting(runs, *, resolution, window_name, sample_scope=None):
"""Exhaustive run/occupancy accounting (v2 definitions, generic names).
Verifies that complete, left-only, right-only and both-censored visits partition the
accepted visits and that runs cover every observed entity-frame exactly once.
``window_name`` is a label only; ``sample_scope`` defaults to the scope derived from the
windows (``'full_primary'`` for the record's full window, ``'secondary_half'`` for one half)
and must be given explicitly for any other window.
"""
g = runs.group
d = runs.duration
left = runs.left_censored
right = runs.right_censored
a = g >= 0
co = a & ~left & ~right
lo = a & left & ~right
ro = a & right & ~left
bo = a & left & right
if int(np.sum(co | lo | ro | bo)) != int(np.sum(a)):
raise AssertionError('censoring classes do not partition the accepted visits')
expected = sum(s.observed_frames * s.n_entities for s in runs.sources)
if int(d.sum()) != expected:
raise AssertionError(f'runs cover {int(d.sum())} entity-frames, expected {expected}')
accepted = int(d[a].sum())
complete_occ = int(d[co].sum())
out = _metadata(runs, resolution, window_name, sample_scope)
out.update({
'all_runs': len(g), 'accepted_visits': int(a.sum()), 'rejected_runs': int((~a).sum()),
'complete_visits': int(co.sum()), 'left_only_censored_visits': int(lo.sum()),
'right_only_censored_visits': int(ro.sum()), 'both_censored_visits': int(bo.sum()),
'left_inclusive_censored_visits': int((a & left).sum()),
'right_inclusive_censored_visits': int((a & right).sum()),
'censored_union_visits': int((lo | ro | bo).sum()),
'complete_occupancy_frames': complete_occ, 'accepted_occupancy_frames': accepted,
'left_only_occupancy_frames': int(d[lo].sum()), 'right_only_occupancy_frames': int(d[ro].sum()),
'both_censored_occupancy_frames': int(d[bo].sum()),
'censored_occupancy_frames': accepted - complete_occ,
'rejected_occupancy_frames': int(d[~a].sum()), 'total_entity_frames': int(d.sum()),
'complete_event_coverage': float(co.sum() / a.sum()) if a.any() else np.nan,
'complete_time_coverage': complete_occ / accepted if accepted else np.nan,
'rejected_fraction': float(d[~a].sum() / d.sum()) if len(d) else np.nan})
return out
[docs]
def half_stats(runs, n_groups):
"""Exact per-group integers of one window (PF ``stats_from_runs`` definitions).
Returns ``n`` (complete visits), ``s`` (their summed duration), ``occ``, ``one``,
``left_only``/``right_only``/``both``, ``ogrp`` (entity x group complete-visit indicator) and
lossless frequency rows ``fg, fd, fc`` sorted by (group, duration), plus rejected and
bridge-change counts. These are the inputs of :func:`separation.separation`. Bridge-change
counts are 0 for raw tables and ``None`` (unknown, reason in ``bridge_changes_reason``) for a
bridge1 table whose counts were not kept, e.g. one rebuilt from an archived v2 runpack.
"""
ng = int(n_groups)
g = np.asarray(runs.group, np.int64)
d = np.asarray(runs.duration, np.int64)
o = np.asarray(runs.entity, np.int64)
if len(g) and int(g.max()) >= ng:
raise ValueError(f'group id {int(g.max())} is outside [0, n_groups={ng})')
left = np.asarray(runs.left_censored, bool)
right = np.asarray(runs.right_censored, bool)
acc = g >= 0
comp = acc & ~left & ~right
out = {'n': np.bincount(g[comp], minlength=ng).astype(np.int64)}
s = np.zeros(ng, np.int64)
np.add.at(s, g[comp], d[comp])
out['s'] = s
occ = np.zeros(ng, np.int64)
np.add.at(occ, g[acc], d[acc])
out['occ'] = occ
out['one'] = np.bincount(g[comp & (d == 1)], minlength=ng).astype(np.int64)
out['left_only'] = np.bincount(g[acc & left & ~right], minlength=ng).astype(np.int64)
out['right_only'] = np.bincount(g[acc & right & ~left], minlength=ng).astype(np.int64)
out['both'] = np.bincount(g[acc & left & right], minlength=ng).astype(np.int64)
ogrp = np.zeros((runs.n_entities, ng), np.uint8)
ogrp[o[comp], g[comp]] = 1
out['ogrp'] = ogrp
if comp.any():
width = int(d[comp].max()) + 1
if ng * width < 2 ** 62:
key, cnt = np.unique(g[comp] * width + d[comp], return_counts=True)
out['fg'] = (key // width).astype(np.int64)
out['fd'] = (key % width).astype(np.int64)
out['fc'] = cnt.astype(np.int64)
else: # same rows without the packed key
gc, dc = g[comp], d[comp]
order = np.lexsort((dc, gc))
gc, dc = gc[order], dc[order]
new = np.r_[True, (gc[1:] != gc[:-1]) | (dc[1:] != dc[:-1])]
first = np.flatnonzero(new)
out['fg'], out['fd'] = gc[first], dc[first]
out['fc'] = np.diff(np.r_[first, len(gc)]).astype(np.int64)
else:
out['fg'] = out['fd'] = out['fc'] = np.zeros(0, np.int64)
out['rejected_occupancy_frames'] = int(d[~acc].sum())
out['rejected_runs'] = int((~acc).sum())
if runs.bridge is not None:
out.update({key: int(getattr(runs.bridge, key)) for key in BRIDGE_FIELDS}, bridge_changes_reason='')
elif runs.preprocessing == 'raw':
out.update(dict.fromkeys(BRIDGE_FIELDS, 0), bridge_changes_reason='')
else:
out.update(dict.fromkeys(BRIDGE_FIELDS, None),
bridge_changes_reason=f'unknown: bridge change counts not available for {runs.preprocessing!r} '
'runs (e.g. rebuilt from a runpack that does not store them)')
return out
def _fit_dtype(values, ladder):
lo = int(values.min()) if len(values) else 0
hi = int(values.max()) if len(values) else 0
for dt in ladder:
info = np.iinfo(dt)
if info.min <= lo and hi <= info.max:
return dt
raise OverflowError('values do not fit a 64-bit integer')
[docs]
def pack_runs(runs):
"""Lossless array pack of one record/window run table (v2 layout, generalised).
Keys ``group, duration, offsets, start, stop`` follow v2 (smallest of int16/32/64,
uint16/32/64, int32/64 that holds the values; PIT data give v2's dtypes) plus
``intervals`` (observed pieces), ``n_bins``, ``entity_ids``, ``record_id``,
``preprocessing``, ``frame_interval``, ``time_unit`` and, when known, ``record_frames``,
``record_fingerprint``, the four bridge1 change counts ``bridge_changes`` (in
:data:`BRIDGE_FIELDS` order) and the declared source (``source_id``, ``source_frame_start``,
``source_frame_step``). The per-frame ``bridge_mask`` is not packed. Save with
``np.savez_compressed``.
"""
if len(runs.sources) != 1 or runs.qualified_ids:
raise ValueError('a runpack holds one unpooled record/window; pack the per-record tables before pooling')
src = runs.sources[0]
d = runs.duration
if len(d) and (np.any(d <= 0) or not np.array_equal(d, runs.exit - runs.entry)):
raise AssertionError('durations must be positive and equal exit - entry')
count = np.bincount(runs.entity, minlength=src.n_entities)
if len(count) != src.n_entities or not np.array_equal(runs.entity, np.repeat(np.arange(src.n_entities), count)):
raise AssertionError('runs must be ordered by entity')
sums = np.zeros(src.n_entities, np.int64)
np.add.at(sums, runs.entity, d)
if not np.all(sums == src.observed_frames):
raise AssertionError('runs of every entity must cover every observed frame exactly once')
offsets = np.r_[0, np.cumsum(count)]
entry, piece = _entries_from_durations(d.astype(np.int64), offsets, src.intervals)
if not (np.array_equal(entry, runs.entry) and np.all(runs.exit <= src.intervals[piece, 1])):
raise AssertionError('runs of every entity must tile the observed pieces contiguously')
pos_dt = _fit_dtype(np.asarray([src.window[0], src.window[1]]), (np.int32, np.int64))
pack = {'group': runs.group.astype(_fit_dtype(runs.group, (np.int16, np.int32, np.int64))),
'duration': d.astype(_fit_dtype(d, (np.uint16, np.uint32, np.uint64))),
'offsets': offsets.astype(_fit_dtype(offsets, (np.int32, np.int64))),
'start': pos_dt(src.window[0]), 'stop': pos_dt(src.window[1]),
'intervals': np.asarray(src.intervals, np.int64), 'n_bins': np.int64(runs.n_bins),
'entity_ids': np.asarray(runs.entity_ids), 'record_id': np.str_(src.record_id),
'preprocessing': np.str_(runs.preprocessing), 'frame_interval': np.float64(src.frame_interval),
'time_unit': np.str_(src.time_unit)}
if src.record_frames is not None:
pack['record_frames'] = np.int64(src.record_frames)
if src.record_fingerprint is not None:
pack['record_fingerprint'] = np.str_(src.record_fingerprint)
if runs.bridge is not None:
pack['bridge_changes'] = np.asarray([getattr(runs.bridge, f) for f in BRIDGE_FIELDS], np.int64)
if src.source is not None:
pack.update({'source_id': np.str_(src.source.source_id),
'source_frame_start': np.int64(src.source.frame_start),
'source_frame_step': np.int64(src.source.frame_step)})
return pack
def _entries_from_durations(d, offsets, intervals):
"""Absolute entries (and piece index) of runs tiling the observed pieces entity by entity."""
if len(d) == 0:
return np.zeros(0, np.int64), np.zeros(0, np.int64)
excl = np.cumsum(d) - d
entity = np.repeat(np.arange(len(offsets) - 1), np.diff(offsets))
local = excl - excl[offsets[:-1]][entity]
lengths = intervals[:, 1] - intervals[:, 0]
cstart = np.cumsum(lengths) - lengths
piece = np.searchsorted(cstart, local, side='right') - 1
return intervals[piece, 0] + (local - cstart[piece]), piece
_INT64_MAX = int(np.iinfo(np.int64).max)
_V2_KEYS = frozenset(('group', 'duration', 'offsets', 'start', 'stop'))
_PACK_KEYS = _V2_KEYS | {'intervals', 'n_bins', 'entity_ids', 'record_id', 'preprocessing', 'frame_interval',
'time_unit'}
_SOURCE_KEYS = frozenset(('source_id', 'source_frame_start', 'source_frame_step'))
_OPTIONAL_KEYS = frozenset(('record_frames', 'record_fingerprint', 'bridge_changes')) | _SOURCE_KEYS
def _pack_int(a, key, low, high=_INT64_MAX):
"""A 0-d integer pack member as a Python int; floats, booleans and strings are refused, never cast."""
if a.ndim != 0 or a.dtype.kind not in 'iu':
raise ValueError(f'runpack {key} must be an integer scalar; got dtype {a.dtype}, shape {a.shape}')
v = int(a)
if not low <= v <= high:
raise ValueError(f'runpack {key} = {v} is outside [{low}, {high}]')
return v
def _pack_str(a, key):
if a.ndim != 0 or a.dtype.kind != 'U' or not str(a):
raise ValueError(f'runpack {key} must be a non-empty string; got dtype {a.dtype}, shape {a.shape}')
return str(a)
def _pack_ints(a, key, ndim):
if a.ndim != ndim or a.dtype.kind not in 'iu':
raise ValueError(f'runpack {key} must be a {ndim}-D integer array; got dtype {a.dtype}, shape {a.shape} '
'(floating, boolean, string and object arrays are refused, never cast)')
return a
def _pack_is_v2(keys):
"""True for the v2 layout, False for the complete :func:`pack_runs` layout; anything else is refused."""
unknown = sorted(keys - _PACK_KEYS - _OPTIONAL_KEYS)
if unknown:
raise ValueError(f'runpack has unknown keys {unknown}; expected the pack_runs or the v2 layout')
if keys == _V2_KEYS:
return True
missing = sorted(_PACK_KEYS - keys)
if missing:
raise ValueError(f'runpack lacks {missing}: it is neither the complete pack_runs layout nor the v2 layout '
f'{sorted(_V2_KEYS)}, so its observed pieces cannot be trusted')
if keys & _SOURCE_KEYS and not _SOURCE_KEYS <= keys:
raise ValueError(f'runpack declares a partial source {sorted(keys & _SOURCE_KEYS)}; need all of '
f'{sorted(_SOURCE_KEYS)}')
return False
def _pack_pieces(a, start, stop):
"""Observed pieces ``(k, 2)``: integers inside ``[start, stop]``, non-empty, in time order (touching allowed)."""
iv = _pack_ints(a, 'intervals', 2)
if iv.shape[1] != 2:
raise ValueError(f'runpack intervals must have shape (k, 2); got {iv.shape}')
if len(iv) and (int(iv.min()) < start or int(iv.max()) > stop):
raise ValueError(f'runpack intervals {iv.tolist()} leave the window [{start}, {stop})')
iv = iv.astype(np.int64)
bad = np.flatnonzero(iv[:, 1] <= iv[:, 0])
if len(bad):
raise ValueError(f'runpack piece {int(bad[0])} {iv[bad[0]].tolist()} is empty or inverted')
bad = np.flatnonzero(iv[1:, 0] < iv[:-1, 1])
if len(bad):
raise ValueError(f'runpack pieces {iv[bad[0]].tolist()} and {iv[bad[0] + 1].tolist()} overlap or are not in '
'time order')
return iv
[docs]
def unpack_runs(pack, *, preprocessing=None):
"""Rebuild the run table from :func:`pack_runs` arrays (or an accepted v2 runpack).
v2 packs have exactly the keys ``group, duration, offsets, start, stop``: one piece
``[start, stop)`` and 8 bins are used; their record length and bridge1 change counts are
unknown (``record_frames=None``, ``bridge=None``), never assumed. ``preprocessing`` labels a
pack that does not store it and must agree with one that does.
A pack is untrusted input. It is validated before any cast or calculation, and a malformed
pack is refused with ``ValueError`` naming the field, entity, piece or run; it is never
repaired (no merging, rounding, dropping or relabelling) and never modified:
* the layout is the complete :func:`pack_runs` layout or the v2 layout (no unknown keys);
* window, ``intervals`` (the observed pieces: non-empty, inside the window, in time order,
touching allowed), offsets, labels (integers ``>= -1``) and durations have integer dtypes
and consistent shapes; every duration is a positive integer that fits its piece;
* the runs of every entity tile its observed pieces exactly, in time order;
* runs are maximal: two consecutive runs of one entity inside one piece never carry the same
label (rejected ``-1`` included). Equal labels across a piece boundary (a gap or a segment
restart, as stored in ``intervals``), an entity or a record boundary are legitimate;
* bridge1 change counts are consistent with the pieces, and every sum and product (run
coordinates, entry bins) is checked against the int64 range before it is computed; a pack
whose exact values fit is never refused for size.
"""
keys = set(getattr(pack, 'files', None) or pack.keys())
v2 = _pack_is_v2(keys)
f = {k: np.asarray(pack[k]) for k in keys} # each member read once (NpzFile members load lazily)
start = _pack_int(f['start'], 'start', 0)
stop = _pack_int(f['stop'], 'stop', 0)
if stop <= start:
raise ValueError(f'runpack window [{start}, {stop}) is empty')
n_frames = _pack_int(f['record_frames'], 'record_frames', stop) if 'record_frames' in keys else None
n_bins = 8 if v2 else _pack_int(f['n_bins'], 'n_bins', 1)
iv = np.asarray([[start, stop]], np.int64) if v2 else _pack_pieces(f['intervals'], start, stop)
stored = None if v2 else _pack_str(f['preprocessing'], 'preprocessing')
for name, value in (('runpack preprocessing', stored), ('preprocessing', preprocessing)):
if value is not None and value not in PREPROCESSING:
raise ValueError(f'{name} must be one of {PREPROCESSING}, got {value!r}')
if preprocessing is not None and stored is not None and preprocessing != stored:
raise ValueError(f'preprocessing={preprocessing!r} contradicts the runpack ({stored!r})')
prep = preprocessing or stored or 'unknown'
record_id = 'record' if v2 else _pack_str(f['record_id'], 'record_id')
tu = 'frame' if v2 else _pack_str(f['time_unit'], 'time_unit')
fi = 1.0
if not v2:
a = f['frame_interval']
fi = float(a) if a.ndim == 0 and a.dtype.kind in 'iuf' else math.nan
if not (math.isfinite(fi) and fi > 0): # also refuses overflow/underflow of the float conversion
raise ValueError(f'runpack frame_interval must be a real scalar that is a positive finite float; got {a!r}')
bridge = None
if 'bridge_changes' in keys:
if prep != 'bridge1':
raise ValueError(f'a {prep!r} runpack cannot carry bridge change counts')
b = _pack_ints(f['bridge_changes'], 'bridge_changes', 1)
if b.shape != (len(BRIDGE_FIELDS),) or int(b.min()) < 0 or int(b.max()) > _INT64_MAX:
raise ValueError(f'runpack bridge_changes must be {len(BRIDGE_FIELDS)} non-negative integers '
f'{BRIDGE_FIELDS}; got {b.tolist()}')
bridge = BridgeChanges(*(int(v) for v in b.tolist()))
fp = _pack_str(f['record_fingerprint'], 'record_fingerprint') if 'record_fingerprint' in keys else None
source = None
if 'source_id' in keys:
source = SourceSelection(_pack_str(f['source_id'], 'source_id'),
_pack_int(f['source_frame_start'], 'source_frame_start', 0),
_pack_int(f['source_frame_step'], 'source_frame_step', 1))
lengths = iv[:, 1] - iv[:, 0]
observed = int(lengths.sum())
g = _pack_ints(f['group'], 'group', 1)
if len(g) and (int(g.min()) < -1 or int(g.max()) > _INT64_MAX):
raise ValueError('runpack group labels must be integers >= -1 (-1 = rejected) that fit int64')
dur = _pack_ints(f['duration'], 'duration', 1)
off = _pack_ints(f['offsets'], 'offsets', 1)
n = len(g)
if len(dur) != n or len(off) < 1:
raise ValueError(f'runpack lengths are inconsistent: {n} labels, {len(dur)} durations, {len(off)} offsets')
if int(off[0]) != 0 or int(off[-1]) != n or int(off.min()) < 0 or int(off.max()) > n:
raise ValueError(f'runpack offsets must run from 0 to the number of runs ({n}) inside [0, {n}]; got '
f'{int(off[0])} .. {int(off[-1])} within [{int(off.min())}, {int(off.max())}]')
offsets = off.astype(np.int64)
n_e = len(offsets) - 1
if n_e < 1:
raise ValueError('runpack holds no entity; a record has at least one')
if bridge is not None:
interior = n_e * sum(max(0, int(x) - 2) for x in lengths.tolist()) # bridgeable positions
pairs = n_e * sum(max(0, int(x) - 3) for x in lengths.tolist()) # adjacent bridgeable pairs
c, a2a, r2a, ov = (getattr(bridge, k) for k in BRIDGE_FIELDS)
if c != a2a + r2a or c > interior or ov > min(pairs, max(0, c - 1)):
raise ValueError(f'runpack bridge_changes {list((c, a2a, r2a, ov))} are impossible for {n_e} entities '
f'on pieces {iv.tolist()}: need {BRIDGE_FIELDS[0]} = {BRIDGE_FIELDS[1]} + '
f'{BRIDGE_FIELDS[2]} <= {interior} and {BRIDGE_FIELDS[3]} <= min({pairs}, '
f'{BRIDGE_FIELDS[0]} - 1)')
step = np.diff(offsets)
bad = np.flatnonzero(step < 1) if observed else np.flatnonzero(step != 0)
if len(bad):
raise ValueError(f'runpack offsets: entity index {int(bad[0])} owns {int(step[bad[0]])} runs; every entity '
f'must own {"at least one run" if observed else "none (no observed frame)"}')
if v2:
ids = np.arange(n_e, dtype=np.int64)
else:
ids = f['entity_ids']
if ids.shape != (n_e,) or ids.dtype.kind not in 'iuU':
raise ValueError(f'runpack entity_ids must be {n_e} integers or unicode strings; got dtype {ids.dtype}, '
f'shape {ids.shape}')
if len(np.unique(ids)) != n_e:
raise ValueError('runpack entity_ids must be unique: a repeated id would make identity ambiguous')
def where(i):
e = int(np.searchsorted(offsets, i, side='right') - 1)
return f'run {i} (entity index {e}, id {ids[e].item()!r})'
if n:
lo, hi, longest = int(dur.min()), int(dur.max()), int(lengths.max())
if lo < 1:
i = int(np.argmax(dur < 1))
raise ValueError(f'runpack {where(i)}: duration {int(dur[i])} is not a positive integer number of frames')
if hi > longest:
i = int(np.argmax(dur > longest))
raise ValueError(f'runpack {where(i)}: duration {int(dur[i])} exceeds the longest observed piece '
f'({longest})')
if n * hi > _INT64_MAX and sum(dur.tolist()) > _INT64_MAX: # exact total only when the fast bound fails
raise ValueError('runpack durations sum beyond the int64 range')
d = dur.astype(np.int64)
sums = np.add.reduceat(d, offsets[:-1]) if n else np.zeros(n_e, np.int64)
bad = np.flatnonzero(sums != observed)
if len(bad):
e = int(bad[0])
raise ValueError(f'runpack entity index {e} (id {ids[e].item()!r}): runs cover {int(sums[e])} frames, the '
f'observed pieces {observed}')
entry, piece = _entries_from_durations(d, offsets, iv)
bad = np.flatnonzero(d > iv[piece, 1] - entry)
if len(bad):
i = int(bad[0])
raise ValueError(f'runpack {where(i)}: [{int(entry[i])}, {int(entry[i] + d[i])}) crosses the end of its '
f'observed piece {iv[piece[i]].tolist()}')
entity = np.repeat(np.arange(n_e, dtype=np.int64), step)
bad = np.flatnonzero((entity[1:] == entity[:-1]) & (piece[1:] == piece[:-1]) & (g[1:] == g[:-1]))
if len(bad):
i = int(bad[0])
raise ValueError(f'runpack {where(i)} and run {i + 1} carry the same label {int(g[i])} and touch at frame '
f'{int(entry[i + 1])} inside one observed piece {iv[piece[i]].tolist()}: runs must be maximal '
'(the encoding is refused, never merged)')
if n and n_bins * int(entry.max() - start) > _INT64_MAX:
raise ValueError(f'runpack entry bins cannot be computed exactly: n_bins ({n_bins}) x the largest entry '
f'displacement ({int(entry.max() - start)}) exceeds the int64 range')
exit_ = entry + d
src = RunSource(record_id, (start, stop), iv, 0, n_e, fi, tu, n_frames, fp, source)
return Runs(group=g, duration=d, entity=entity, entry=entry, exit=exit_, left_censored=entry == iv[piece, 0],
right_censored=exit_ == iv[piece, 1], entry_bin=(n_bins * (entry - start)) // (stop - start),
source=np.zeros(n, np.int64), sources=(src,), entity_ids=ids, n_bins=n_bins, preprocessing=prep,
bridge=bridge)