"""Per-group complete-visit descriptions (v2 definitions), returned as lists of row dicts.
Definitions are inherited unchanged from the accepted v2 code: mean, median, linear-interpolated
quantiles, sample SD (``ddof=1``), CV = SD/mean, unscaled MAD, ``D_MAD = 1.4826*MAD/median``,
one-frame fraction, entity support/concentration, the mean-matched geometric comparator
``q = 1/mu``, ``CV_geom = sqrt(1 - 1/mu)``, and ``Q_raw``/``Q_log`` on all complete visits.
Means/frequencies need one complete visit; dispersion needs two (otherwise NaN with a reason).
Every occupied accepted group keeps a row. The 20 visits / 5 entities / 4 bins support view is a
supplementary flag set, never a filter. The numpy call order of v2 is kept (complete visits in
run order per group) so values are bitwise comparable with archived v2 tables. Rows use
generic ``entity`` names (archived ``oxygen`` names: :mod:`.archived_names`, regression only).
"""
from __future__ import annotations
import numpy as np
from .runs import _metadata
D_MAD_SCALE = 1.4826 # frozen v2 scale convention, not a physical calibration
[docs]
def q_stat(duration, group, log=False):
"""In-sample mean-explained variation ``1 - SS_W/SS_T`` over all complete visits (v2).
``y = log1p(d)`` when ``log``. NaN when fewer than two visits or zero total variance.
"""
duration = np.asarray(duration)
if len(duration) < 2:
return np.nan
y = np.log1p(duration.astype(float)) if log else duration.astype(float)
grand = np.mean(y)
total = np.sum((y - grand) ** 2)
if total <= 0:
return np.nan
ids, inv = np.unique(np.asarray(group), return_inverse=True)
n = np.bincount(inv)
mean = np.bincount(inv, weights=y) / n
between = np.sum(n * (mean - grand) ** 2)
return float(between / total)
def _group_index(g):
"""Accepted run indices grouped by label: (ids ascending, bounds, run indices ascending within group)."""
acc_idx = np.flatnonzero(g >= 0)
run_index = acc_idx[np.argsort(g[acc_idx], kind='stable')]
k = g[run_index]
bounds = np.r_[0, np.flatnonzero(k[1:] != k[:-1]) + 1, len(k)]
return (k[bounds[:-1]] if len(k) else k), bounds, run_index
[docs]
def group_tables(runs, *, resolution, window_name, sample_scope=None):
"""All occupied accepted groups: per-group rows, exact duration frequencies/ECDF, entity and bin counts.
Returns ``(groups, frequencies, entity_counts, bin_counts)`` as lists of dicts.
``sample_scope`` follows :func:`runs.accounting` (derived from the windows, never the name).
"""
meta = _metadata(runs, resolution, window_name, sample_scope)
mask = runs.complete()
g = np.asarray(runs.group)
d = runs.duration
o = runs.entity
bins = runs.entry_bin
occupied, bounds, run_index = _group_index(g)
rows, freq, o_counts, bin_counts = [], [], [], []
for q, gid in enumerate(occupied.tolist()):
all_idx = run_index[bounds[q]:bounds[q + 1]] # ascending run index, as np.flatnonzero(g == gid)
ok_idx = all_idx[mask[all_idx]]
x = d[ok_idx].astype(float)
ox = o[ok_idx]
bn = bins[ok_idx]
n = len(x)
all_n = len(all_idx)
cen = all_idx[~mask[all_idx]]
left = runs.left_censored[all_idx]
right = runs.right_censored[all_idx]
med = float(np.median(x)) if n else np.nan
mean = float(np.mean(x)) if n else np.nan
sd = float(np.std(x, ddof=1)) if n >= 2 else np.nan
mad = float(np.median(np.abs(x - med))) if n >= 2 else np.nan
ox_ids, ox_ct = np.unique(ox, return_counts=True)
by_entity = np.split(x[np.argsort(ox, kind='stable')], np.cumsum(ox_ct)[:-1]) if n else []
entity_medians = np.array([np.median(v) for v in by_entity])
b_ids, b_ct = np.unique(bn, return_counts=True)
dur_ids, dur_ct = np.unique(x.astype(np.int64), return_counts=True)
cumulative = 0
for v, c in zip(dur_ids, dur_ct):
cumulative += int(c)
freq.append({**meta, 'group': int(gid), 'visit_duration_frames': int(v), 'complete_visits': int(c),
'cumulative_visits': cumulative, 'empirical_cdf': cumulative / n})
for oid, c in zip(ox_ids, ox_ct):
o_counts.append({'resolution': meta['resolution'], 'preprocessing': meta['preprocessing'],
'window': meta['window'], 'group': int(gid), 'entity_local_index': int(oid),
'complete_visits': int(c)})
for b, c in zip(b_ids, b_ct):
bin_counts.append({'resolution': meta['resolution'], 'preprocessing': meta['preprocessing'],
'window': meta['window'], 'group': int(gid), 'entry_bin': int(b),
'complete_visits': int(c)})
accepted_occ = int(d[all_idx].sum())
complete_occ = int(x.sum())
mu = mean
q_geo = 1 / mu if n else np.nan
geom_cv = np.sqrt(1 - 1 / mu) if n else np.nan
rows.append({
**meta, 'group': int(gid), 'nominal_group_count': resolution,
'realized_occupied_group_count': len(occupied),
'total_visits': all_n, 'complete_visits': n,
'left_only_censored_visits': int(np.count_nonzero(left & ~right)),
'right_only_censored_visits': int(np.count_nonzero(right & ~left)),
'both_censored_visits': int(np.count_nonzero(left & right)),
'censored_union_visits': len(cen), 'censored_fraction': len(cen) / all_n,
'complete_event_coverage': n / all_n, 'complete_occupancy_frames': complete_occ,
'censored_occupancy_frames': int(d[cen].sum()), 'accepted_occupancy_frames': accepted_occ,
'complete_time_coverage': complete_occ / accepted_occ if accepted_occ else np.nan,
'distinct_entities_complete': len(ox_ids), 'occupied_entry_bins': len(b_ids),
'max_entity_episode_share': ox_ct.max() / n if n else np.nan,
'effective_entity_concentration': n * n / np.sum(ox_ct.astype(float) ** 2) if n else np.nan,
'mean_duration_frames': mean, 'median_duration_frames': med,
'q25_duration_frames': np.quantile(x, .25) if n else np.nan,
'q75_duration_frames': np.quantile(x, .75) if n else np.nan,
'min_duration_frames': np.min(x) if n else np.nan,
'max_duration_frames': np.max(x) if n else np.nan,
'sd_duration_frames': sd, 'cv': sd / mean if n >= 2 else np.nan,
'mad_unscaled_frames': mad, 'd_mad': D_MAD_SCALE * mad / med if n >= 2 else np.nan,
'one_frame_fraction': np.mean(x == 1) if n else np.nan,
'entity_balanced_median_frames': np.median(entity_medians) if n else np.nan,
'entity_balanced_q25_frames': np.quantile(entity_medians, .25) if n else np.nan,
'entity_balanced_q75_frames': np.quantile(entity_medians, .75) if n else np.nan,
'geometric_q': q_geo, 'geometric_mean_frames': mu,
'geometric_variance_frames2': mu * (mu - 1) if n else np.nan,
'geometric_cv': geom_cv, 'geometric_one_frame_probability': q_geo,
'cv_minus_geometric_cv': sd / mean - geom_cv if n >= 2 else np.nan,
'sampling_floor_status': ('degenerate_mu_one' if n and mu == 1 else
('nondegenerate' if n else 'undefined_no_complete')),
'duration_undefined_reason': 'zero_complete_visits' if n == 0 else '',
'dispersion_undefined_reason': 'fewer_than_two_complete_visits' if n < 2 else ''})
return rows, freq, o_counts, bin_counts
def _column(rows, key, dtype):
return np.asarray([r[key] for r in rows], dtype=dtype)
[docs]
def aggregate(runs, groups, account):
"""Window summary over all complete visits plus group-equal/episode-weighted D_MAD and CV (v2)."""
mask = runs.complete()
d = runs.duration[mask]
g = runs.group[mask]
out = {key: account[key] for key in ('resolution', 'preprocessing', 'window', 'sample_scope')}
out.update({'occupied_groups': len(groups), 'complete_visits': len(d),
'mean_duration_frames': np.mean(d) if len(d) else np.nan,
'median_duration_frames': np.median(d) if len(d) else np.nan,
'one_frame_fraction': np.mean(d == 1) if len(d) else np.nan,
'q_log': q_stat(d, g, True), 'q_raw': q_stat(d, g, False),
'complete_event_coverage': account['complete_event_coverage'],
'complete_time_coverage': account['complete_time_coverage'],
'rejected_fraction': account['rejected_fraction']})
for tag, col in (('d', 'd_mad'), ('cv', 'cv')):
values = _column(groups, col, np.float64)
defined = ~np.isnan(values)
x = values[defined]
n = _column(groups, 'complete_visits', np.int64)[defined]
occ = _column(groups, 'complete_occupancy_frames', np.int64)[defined]
out.update({tag + '_group_median': np.median(x) if len(x) else np.nan,
tag + '_group_q25': np.quantile(x, .25) if len(x) else np.nan,
tag + '_group_q75': np.quantile(x, .75) if len(x) else np.nan,
tag + '_episode_weighted': np.average(x, weights=n) if len(x) else np.nan,
tag + '_defined_groups': int(defined.sum()),
tag + '_defined_group_fraction': float(defined.mean()) if len(groups) else np.nan,
tag + '_defined_episodes': int(n.sum()),
tag + '_defined_episode_fraction': float(n.sum() / len(d)) if len(d) else np.nan,
tag + '_defined_occupancy_frames': int(occ.sum()),
tag + '_defined_occupancy_fraction': (float(occ.sum() / account['accepted_occupancy_frames'])
if account['accepted_occupancy_frames'] else np.nan)})
return out
def _skipna_median(x):
x = x[~np.isnan(x)]
return np.median(x) if len(x) else np.nan
[docs]
def support_diagnostic(groups, summary, *, min_visits=20, min_entities=5, min_bins=4):
"""Supplementary support flags per group (never a filter) and the qualified-subset summary.
Defaults are the inherited v2 thresholds (20 complete visits, 5 distinct entities, entries in
4 of the run table's ``n_bins`` bins); every group keeps its row with its condition flags.
"""
keep = ('resolution', 'preprocessing', 'window', 'group', 'complete_visits', 'distinct_entities_complete',
'occupied_entry_bins', 'complete_occupancy_frames', 'accepted_occupancy_frames', 'd_mad', 'cv')
tab = []
for r in groups:
row = {key: r[key] for key in keep}
row['condition_visits'] = r['complete_visits'] >= min_visits
row['condition_entities'] = r['distinct_entities_complete'] >= min_entities
row['condition_bins'] = r['occupied_entry_bins'] >= min_bins
row['qualified'] = row['condition_visits'] and row['condition_entities'] and row['condition_bins']
tab.append(row)
qualified = [r for r in tab if r['qualified']]
q_visits = _column(qualified, 'complete_visits', np.int64)
q_cocc = _column(qualified, 'complete_occupancy_frames', np.int64)
q_aocc = _column(qualified, 'accepted_occupancy_frames', np.int64)
all_cocc = _column(groups, 'complete_occupancy_frames', np.int64).sum()
all_aocc = _column(groups, 'accepted_occupancy_frames', np.int64).sum()
record = {key: summary[key] for key in ('resolution', 'preprocessing', 'window', 'sample_scope')}
record.update({
'qualified_group_ids': [int(r['group']) for r in qualified],
'qualified_groups': len(qualified), 'occupied_groups': len(groups),
'qualified_group_fraction': len(qualified) / len(groups) if len(groups) else np.nan,
'qualified_complete_visits': int(q_visits.sum()),
'qualified_complete_visit_fraction': (q_visits.sum() / summary['complete_visits']
if summary['complete_visits'] else np.nan),
'qualified_complete_occupancy_frames': int(q_cocc.sum()),
'qualified_complete_occupancy_fraction': q_cocc.sum() / all_cocc if all_cocc else np.nan,
'qualified_accepted_occupancy_frames': int(q_aocc.sum()),
'qualified_accepted_occupancy_fraction': q_aocc.sum() / all_aocc if all_aocc else np.nan,
'qualified_d_group_median': _skipna_median(_column(qualified, 'd_mad', np.float64)),
'qualified_cv_group_median': _skipna_median(_column(qualified, 'cv', np.float64)),
'full_d_group_median': summary['d_group_median'],
'full_cv_group_median': summary['cv_group_median'],
'support_thresholds': {'min_visits': min_visits, 'min_entities': min_entities, 'min_bins': min_bins}})
return tab, record