"""Canonical labels-first temporal assessment API (experimental, opt-in), shared by Python and CLI.
Flow per input: integer labels (frames x entities, ``-1`` = rejected) ->
:class:`~.record.LabelRecord` -> early/late runs at an explicit split, per preprocessing ->
:func:`~.separation.assess` (B, E, ratio, status, Level-1 enclosures, coverage) -> one
:class:`~.pareto.CandidateRow` per (preprocessing, candidate, weighting) -> :func:`fronts`, one
front per context (namespace, descriptor policy, PCA policy, preprocessing, window, sample scope,
candidate, weighting). The ratio B/E is a temporal separation diagnostic, NOT HiFi's fidelity
L/R, and nothing here selects a K, cutoff or winner.
Python::
from atom_hifi.experimental.temporal import assessment, record
rec = record.LabelRecord.from_labels(labels, entity_ids=ids, n_groups=K,
source=record.SourceSelection('traj.xyz#sha256=...'))
a = assessment.assess_record(rec, split=s, candidate_id='K12', K=12, namespace='mine',
descriptor_policy={...}, pca_policy='fixed_k:8')
front_rows, relation_rows, summaries = assessment.fronts(a.pairs + b.pairs)
or ``assessment.run(config)`` with the same mapping the CLI reads from YAML/JSON (see
:func:`resolve_config`), then ``assessment.read_results(output_dir)``.
Identity contract. A front compares candidates of ONE observed population at one sampling and
one pair of windows. :func:`check_front_identity` (called by :func:`fronts`, hence by :func:`run`
and the CLI) requires every member of a context with two or more members to declare the same
source population (:class:`~.record.SourceSelection`), the same layout and the same windows.
Shapes, labels, record ids, candidate ids and paths never establish a population.
Exact escalation is optional: without ``exact_budget`` every pair that the exact-zero identities
and Level-1 enclosures cannot order stays ``'unresolved'`` (a valid result). With a budget each
(input, preprocessing) computes exact components at most once; success or exhaustion is cached
for that assessment only.
"""
from __future__ import annotations
import csv
from dataclasses import dataclass, field
from fractions import Fraction
import hashlib
import io
import json
import math
import os
from pathlib import Path
import re
import sys
import time
SCHEMA_VERSION = 'atom_hifi.experimental.temporal/1'
DOCS_PAGE = 'docs/sphinx/experimental_temporal.md ("Experimental: temporal assessment")'
TOP_REQUIRED = ('inputs', 'output_dir', 'windows', 'preprocessing', 'candidates', 'weightings')
TOP_OPTIONAL = ('frame_interval', 'time_unit', 'n_bins', 'exact_budget', 'exports')
INPUT_REQUIRED = ('path', 'candidate_id', 'K', 'namespace', 'descriptor_policy', 'pca_policy')
INPUT_OPTIONAL = ('n_groups', 'entity_ids', 'record_id', 'source')
DESCRIPTOR_KEYS = ('cutoff', 'image_policy', 'reference_cell_policy')
EXPORT_KEYS = ('descriptions', 'runpacks')
WINDOW = 'early+late'
SAMPLE_SCOPE = 'joint_halves'
ACCEPTANCE_DEFINITION = ('native acceptance: accepted raw entity-frames / observed entity-frames over the full '
'record window (same value for every preprocessing)')
RATIO_NOTE = 'ratio = B/E is a temporal separation diagnostic, not HiFi fidelity F = L/R'
OUTPUTS = ('candidates.csv', 'candidates.json', 'fronts.csv', 'front_relations.csv', 'group_contributions.csv',
'run_summary.json')
DESCRIPTIONS_DIR = 'descriptions'
CANDIDATE_COLUMNS = (
'candidate_id', 'namespace', 'cutoff', 'image_policy', 'reference_cell_policy', 'pca_policy', 'K', 'n_groups',
'preprocessing', 'window', 'sample_scope', 'candidate', 'weighting', 'status_code', 'status', 'B', 'E', 'ratio',
'separation_early', 'separation_late', 'B_exact_zero', 'E_exact_zero', 'B_enclosure_lo', 'B_enclosure_hi',
'E_enclosure_lo', 'E_enclosure_hi', 'production_inside_enclosure', 'arithmetic_path', 'visit_mass_total',
'acceptance_numerator', 'acceptance_denominator', 'record_id', 'record_fingerprint', 'label_path')
CONTEXT_COLUMNS = ('namespace', 'cutoff', 'image_policy', 'reference_cell_policy', 'pca_policy', 'preprocessing',
'window', 'sample_scope', 'candidate', 'weighting')
FRONT_COLUMNS = CONTEXT_COLUMNS + (
'candidate_id', 'K', 'acceptance_numerator', 'acceptance_denominator', 'status_code', 'status', 'front_status',
'dominated_by', 'unresolved_with', 'is_ratio_maximizer', 'ratio_maximizer_unresolved', 'front_minimum_K',
'front_maximum_A')
RELATION_COLUMNS = CONTEXT_COLUMNS + ('a', 'b', 'ratio_relation', 'level', 'evidence')
CONTRIBUTION_COLUMNS = (
'input_index', 'candidate_id', 'namespace', 'cutoff', 'image_policy', 'reference_cell_policy', 'pca_policy', 'K',
'n_groups', 'preprocessing', 'arithmetic_path', 'group', 'n_early', 'n_late', 's_early', 's_late', 'visit_mass',
'ntau', 'nw1', 'den', 'mu_early', 'mu_late', 'dmu', 'w1', 'weight_visit', 'weight_equal',
'E_contribution_tau_visit', 'E_contribution_tau_equal', 'E_contribution_P_visit', 'E_contribution_P_equal')
STATUS_EXPLANATIONS = {
'finite': 'E > 0 exactly: the ratio B/E is defined (B may be 0 exactly).',
'undefined_no_between_separation_and_no_window_difference':
'B = 0 and E = 0 exactly (0/0): no ratio; listed in the non-finite register.',
'zero_denominator_positive_separation':
'E = 0 exactly while B > 0: the ratio diverges; listed in the non-finite register, never ranked.',
'unavailable_empty_joint_set':
'no group has a complete visit in both windows: B and E are unavailable.',
'numerical_indeterminate':
'production floats are non-finite or contradict the exact zero identities; listed in the register.',
}
FRONT_STATUS_EXPLANATIONS = {
'nondominated': 'no member dominates it (K <= , ratio >= , acceptance >= with one strict), certified exactly.',
'dominated': 'dominated_by lists certified dominators (exact zero, Level-1 interval or Level-2b exact).',
'unresolved': 'not dominated by a certified relation, but a ratio comparison with a potential dominator could '
'not be ordered (no exact budget, or the budget was exhausted); never ranked by floats.',
'register': 'non-finite status: excluded from the front.',
}
EXCLUDED_FROM_PARITY = ('wall_seconds',)
[docs]
class ConfigError(ValueError):
"""Invalid configuration or input (actionable message; CLI exit status 2)."""
[docs]
class IdentityError(ValueError):
"""Candidates that may not share a front: undeclared or different source population, layout or windows."""
# ------------------------------------------------------------------------------------- config
def _load_mapping(path):
path = Path(path)
suffix = path.suffix.lower()
if suffix not in ('.json', '.yaml', '.yml'):
raise ConfigError(f'{path}: config must be .yaml, .yml or .json')
text = path.read_text()
try:
if suffix == '.json':
data = json.loads(text)
else:
import yaml
data = yaml.safe_load(text)
except ValueError as exc:
raise ConfigError(f'{path}: invalid JSON: {exc}') from exc
except Exception as exc:
if type(exc).__module__.startswith('yaml'):
raise ConfigError(f'{path}: invalid YAML: {exc}') from exc
raise
if not isinstance(data, dict):
raise ConfigError(f'{path}: config must be a mapping')
return data
def _keys(mapping, required, optional, where):
if not isinstance(mapping, dict):
raise ConfigError(f'{where} must be a mapping')
unknown = sorted(set(mapping) - set(required) - set(optional))
missing = [k for k in required if k not in mapping]
if unknown or missing:
raise ConfigError(f'{where}: unknown keys {unknown}, missing keys {missing}')
def _choices(value, allowed, where):
if not isinstance(value, list) or not value or len(set(value)) != len(value) or not set(value) <= set(allowed):
raise ConfigError(f'{where} must be a non-empty list of distinct values from {list(allowed)}, got {value!r}')
return tuple(value)
def _int(value, where, minimum=None):
if isinstance(value, bool) or not isinstance(value, int) or (minimum is not None and value < minimum):
raise ConfigError(f'{where} must be an int' + (f' >= {minimum}' if minimum is not None else '') + f', got {value!r}')
return value
def _scalar(value, where):
if isinstance(value, bool) or not isinstance(value, (str, int, float)) or (
isinstance(value, float) and not math.isfinite(value)):
raise ConfigError(f'{where} must be a string or a finite number, got {value!r}')
return value
def _pca_identity(value, where):
if isinstance(value, str) and value:
return value
from atom_hifi.experimental.judges import PCAPolicy
if isinstance(value, dict):
_keys(value, ('kind',), ('k',), where)
try:
return PCAPolicy(value['kind'], value.get('k')).identity
except (TypeError, ValueError) as exc:
raise ConfigError(f'{where}: {exc}') from exc
if isinstance(value, PCAPolicy):
return value.identity
raise ConfigError(f'{where} must be a non-empty string or a {{kind, k}} mapping, got {value!r}')
def _descriptor_policy(value, where):
from .pareto import DescriptorPolicy
if isinstance(value, DescriptorPolicy):
return value
_keys(value, DESCRIPTOR_KEYS, (), where)
return DescriptorPolicy(**{k: _scalar(value[k], f'{where}.{k}') for k in DESCRIPTOR_KEYS})
def _source(value, where):
from .record import SourceSelection
if value is None or isinstance(value, SourceSelection):
return value
_keys(value, ('id',), ('frame_start', 'frame_step'), where)
if not isinstance(value['id'], str) or not value['id']:
raise ConfigError(f'{where}.id must be a non-empty string (an identity of the observed record, e.g. a '
f'trajectory name with its file hash), got {value["id"]!r}')
return SourceSelection(value['id'], _int(value.get('frame_start', 0), f'{where}.frame_start', 0),
_int(value.get('frame_step', 1), f'{where}.frame_step', 1))
def _budget(value, where='exact_budget'):
if value is None:
return None
_keys(value, (), ('max_seconds', 'max_work'), where)
if not value:
raise ConfigError(f'{where} needs max_seconds and/or max_work; omit {where} entirely to run without exact '
'escalation (unordered pairs then stay unresolved)')
for key, v in value.items():
if isinstance(v, bool) or not isinstance(v, (int, float)) or not (math.isfinite(v) and v > 0):
raise ConfigError(f'{where}.{key} must be a positive finite number, got {v!r}')
return dict(value)
def _path_text(value, where, suffixes=('.npy',)):
if not isinstance(value, (str, os.PathLike)) or not str(value):
raise ConfigError(f'{where} must be a path string, got {value!r}')
if Path(value).suffix.lower() not in suffixes:
raise ConfigError(f'{where}: only {"/".join(suffixes)} files are supported, got {str(value)!r}')
return str(value)
def _canonical_sha256(mapping):
text = json.dumps(mapping, sort_keys=True, separators=(',', ':'), default=str)
return hashlib.sha256(text.encode()).hexdigest()
[docs]
def resolve_config(config, *, base_dir=None):
"""Validated, fully resolved configuration from a YAML/JSON path or a mapping.
Relative paths resolve against ``base_dir`` (default: the config file's directory, or the
current directory for a mapping). Unknown keys are refused. Keys::
output_dir: out # required
windows: {split: S} # required explicit early/late split; optional segments
preprocessing: [raw, bridge1] # required, from raw / bridge1
candidates: [tau, P] # required
weightings: [visit, equal] # required
frame_interval: 1 # optional metadata; durations are in frames
time_unit: frame # optional
n_bins: 8 # optional entry bins (descriptions only)
exact_budget: {max_seconds: 60} # optional; enables Level-2b escalation
exports: {descriptions: false, runpacks: false} # optional per-input exports
inputs: # required, one per candidate labeling
- path: labels.npy # 2-D integer .npy, -1 = rejected
candidate_id: K12
K: 12
namespace: mine
descriptor_policy: {cutoff: ..., image_policy: ..., reference_cell_policy: ...}
pca_policy: {kind: fixed_k, k: 8} # or {kind: native_auto}, or an identity string
source: {id: 'traj.xyz#sha256=...', frame_start: 0, frame_step: 1} # needed to share a front
n_groups: 12 # optional (default K)
entity_ids: [..] or ids.npy # optional
record_id: K12 # optional (default candidate_id)
"""
from .runs import PREPROCESSING
from .separation import CANDIDATES, WEIGHTINGS
if isinstance(config, dict):
raw = config
base = Path(base_dir if base_dir is not None else Path.cwd()).resolve()
cfg = {'config_path': None, 'config_name': None, 'config_sha256': _canonical_sha256(raw),
'config_sha256_definition': 'sha256 of the canonical JSON of the mapping (sorted keys)'}
elif isinstance(config, (str, os.PathLike)):
path = Path(config).resolve()
if not path.is_file():
raise FileNotFoundError(f'config file not found: {path}')
raw = _load_mapping(path)
base = Path(base_dir).resolve() if base_dir is not None else path.parent
from atom_hifi.experimental.judges import file_sha256
cfg = {'config_path': str(path), 'config_name': path.name, 'config_sha256': file_sha256(path),
'config_sha256_definition': 'sha256 of the config file bytes'}
else:
raise ConfigError(f'config must be a path or a mapping, got {type(config).__name__}')
_keys(raw, TOP_REQUIRED, TOP_OPTIONAL, 'config')
if not isinstance(raw['output_dir'], (str, os.PathLike)) or not str(raw['output_dir']):
raise ConfigError('output_dir must be a path string')
cfg.update({'base_dir': str(base), 'output_dir_as_given': str(raw['output_dir']),
'output_dir': str((base / str(raw['output_dir'])).resolve())})
fi = raw.get('frame_interval', 1)
if isinstance(fi, bool) or not isinstance(fi, (int, float)) or not (math.isfinite(fi) and fi > 0):
raise ConfigError(f'frame_interval must be a positive finite number, got {fi!r}')
cfg['frame_interval'] = fi
tu = raw.get('time_unit', 'frame')
if not isinstance(tu, str) or not tu:
raise ConfigError('time_unit must be a non-empty string')
cfg['time_unit'] = tu
_keys(raw['windows'], ('split',), ('segments',), 'windows')
cfg['split'] = _int(raw['windows']['split'], 'windows.split', 1)
seg = raw['windows'].get('segments')
if seg is not None:
if not isinstance(seg, list) or not all(isinstance(s, list) and len(s) == 2 for s in seg):
raise ConfigError('windows.segments must be a list of [start, stop] pairs')
seg = [[_int(a, 'windows.segments', 0), _int(b, 'windows.segments', 0)] for a, b in seg]
cfg['segments'] = seg
cfg['preprocessing'] = _choices(raw['preprocessing'], PREPROCESSING, 'preprocessing')
cfg['candidates'] = _choices(raw['candidates'], CANDIDATES, 'candidates')
cfg['weightings'] = _choices(raw['weightings'], WEIGHTINGS, 'weightings')
cfg['n_bins'] = _int(raw.get('n_bins', 8), 'n_bins', 1)
cfg['exact_budget'] = _budget(raw.get('exact_budget'))
exports = raw.get('exports') or {}
_keys(exports, (), EXPORT_KEYS, 'exports')
for key, v in exports.items():
if not isinstance(v, bool):
raise ConfigError(f'exports.{key} must be true or false, got {v!r}')
cfg['exports'] = {k: bool(exports.get(k, False)) for k in EXPORT_KEYS}
if not isinstance(raw['inputs'], list) or not raw['inputs']:
raise ConfigError('inputs must be a non-empty list')
inputs = []
for i, item in enumerate(raw['inputs']):
where = f'inputs[{i}]'
_keys(item, INPUT_REQUIRED, INPUT_OPTIONAL, where)
cid = item['candidate_id']
if not isinstance(cid, str) or not cid:
raise ConfigError(f'{where}.candidate_id must be a non-empty string')
where = f'inputs[{i}] ({cid})'
ns = item['namespace']
if not isinstance(ns, str) or not ns:
raise ConfigError(f'{where}.namespace must be a non-empty string')
dp = _descriptor_policy(item['descriptor_policy'], f'{where}.descriptor_policy')
K = _int(item['K'], f'{where}.K', 1)
path = _path_text(item['path'], f'{where}.path')
ids = item.get('entity_ids')
ids_as_given = ids
if isinstance(ids, (str, os.PathLike)):
ids = str((base / _path_text(ids, f'{where}.entity_ids')).resolve())
ids_as_given = str(ids_as_given)
elif ids is not None and not (isinstance(ids, list) and (all(isinstance(x, int) and not isinstance(x, bool)
for x in ids)
or all(isinstance(x, str) for x in ids))):
raise ConfigError(f'{where}.entity_ids must be a list of ints, a list of strings or a path to a .npy')
rid = item.get('record_id', cid)
if not isinstance(rid, str) or not rid:
raise ConfigError(f'{where}.record_id must be a non-empty string')
inputs.append({'path': str((base / path).resolve()), 'path_as_given': path, 'candidate_id': cid, 'K': K,
'n_groups': _int(item.get('n_groups', K), f'{where}.n_groups', 1), 'namespace': ns,
'descriptor_policy': {k: getattr(dp, k) for k in DESCRIPTOR_KEYS},
'pca_policy': _pca_identity(item['pca_policy'], f'{where}.pca_policy'),
'entity_ids': ids, 'entity_ids_as_given': ids_as_given, 'record_id': rid,
'source': _source(item.get('source'), f'{where}.source')})
keys = [(x['namespace'], tuple(x['descriptor_policy'].values()), x['pca_policy'], x['candidate_id']) for x in inputs]
if len(set(keys)) != len(keys):
raise ConfigError('candidate_id must be unique within one (namespace, descriptor policy, PCA policy)')
cfg['inputs'] = inputs
return cfg
# ------------------------------------------------------------------------------------- compute
class _ExactProvider:
"""Exact (Level-2b) components of one immutable assessment: one (record, windows,
preprocessing, n_groups, budget). Success or exhaustion is computed at most once."""
def __init__(self, n_groups, early, late, budget):
self._args = (n_groups, early, late)
self._budget = dict(budget)
self._value = None
self._exhausted = None
self.calls = 0
def components(self):
from . import separation
if self._exhausted is not None:
raise separation.BudgetExhausted(str(self._exhausted))
if self._value is None:
self.calls += 1
try:
self._value = separation.exact_components(*self._args, **self._budget)
except separation.BudgetExhausted as exc:
self._exhausted = exc
raise
return self._value
def combo(self, combo):
return lambda: self.components()[combo]
def _identity(record, early_w, late_w):
return {'population_key': record.population_key, 'layout_key': record.layout_key,
'source': record.source.as_dict() if record.source is not None else None,
'n_frames': record.n_frames, 'n_entities': record.n_entities,
'early_window': list(early_w), 'late_window': list(late_w)}
def _contributions(res, prep, meta):
import numpy as np
r = res['result']
rows = []
early, late = res['early'], res['late']
for j, g in enumerate(np.asarray(r['joint_ids']).tolist()):
row = dict(meta, preprocessing=prep, arithmetic_path=r['path'], group=int(g),
n_early=int(early['n'][g]), n_late=int(late['n'][g]), s_early=int(early['s'][g]),
s_late=int(late['s'][g]), visit_mass=int(r['visit_mass'][j]), ntau=int(r['ntau'][j]),
nw1=int(r['nw1'][j]), den=int(r['den'][j]), mu_early=float(r['mu_early'][j]),
mu_late=float(r['mu_late'][j]), dmu=float(r['dmu'][j]), w1=float(r['w1'][j]),
weight_visit=float(r['weights']['visit'][j]), weight_equal=float(r['weights']['equal'][j]))
for (cand, wt), v in r['E_contributions'].items():
row[f'E_contribution_{cand}_{wt}'] = float(v[j])
rows.append(row)
return rows
[docs]
def assess_record(record, *, split, candidate_id, K, namespace, descriptor_policy, pca_policy,
preprocessing=('raw',), candidates=('tau', 'P'), weightings=('visit', 'equal'), n_groups=None,
exact_budget=None, label_path=None, input_index=0):
"""Assess one :class:`~.record.LabelRecord` at the explicit early/late ``split``.
``descriptor_policy`` is a :class:`~.pareto.DescriptorPolicy` or a ``{cutoff, image_policy,
reference_cell_policy}`` mapping (metadata only); ``pca_policy`` an identity string, a
``{kind, k}`` mapping or a :class:`~atom_hifi.experimental.judges.PCAPolicy`.
``exact_budget`` (optional ``{max_seconds, max_work}``) enables Level-2b escalation; it is
validated as in the configuration. Returns an :class:`InputAssessment`.
"""
from fractions import Fraction
from .pareto import CandidateRow
from .runs import extract_runs, native_acceptance
from .separation import assess
budget = _budget(exact_budget)
policy = _descriptor_policy(descriptor_policy, 'descriptor_policy')
pca = _pca_identity(pca_policy, 'pca_policy')
ng = int(K if n_groups is None else n_groups)
early_w, late_w = record.half_windows(split)
a_num, a_den = native_acceptance(record)
acceptance = Fraction(a_num, a_den)
identity = _identity(record, early_w, late_w)
meta = {'input_index': input_index, 'candidate_id': candidate_id, 'namespace': namespace,
'cutoff': policy.cutoff, 'image_policy': policy.image_policy,
'reference_cell_policy': policy.reference_cell_policy, 'pca_policy': pca, 'K': K, 'n_groups': ng}
pairs, contributions, results, providers = [], [], {}, {}
for prep in preprocessing:
early = extract_runs(record, early_w, preprocessing=prep)
late = extract_runs(record, late_w, preprocessing=prep)
res = assess(early, late, ng)
results[prep] = res
contributions.extend(_contributions(res, prep, meta))
provider = _ExactProvider(ng, res['early'], res['late'], budget) if budget else None
providers[prep] = provider
for cand in candidates:
for wt in weightings:
combo = (cand, wt)
c = res['result']['combos'][combo]
enc = res['level1']['enclosures'][combo] if res['level1'] else None
row = CandidateRow(
candidate_id=candidate_id, namespace=namespace, descriptor_policy=policy, pca_policy=pca, K=K,
preprocessing=prep, window=WINDOW, sample_scope=SAMPLE_SCOPE, candidate=cand, weighting=wt,
B=c['B'], E=c['E'], ratio=c['ratio'], status_code=c['status_code'], enclosures=enc,
coverage=res['coverage'], acceptance=acceptance, B_exact_zero=c['B_exact_zero'],
exact=provider.combo(combo) if provider else None, exact_budget=budget)
extras = {'n_groups': ng, 'separation_early': c['separation_early'],
'separation_late': c['separation_late'], 'E_exact_zero': c['E_exact_zero'],
'production_inside_enclosure': res['reliability'][combo]['production_inside_enclosure'],
'arithmetic_path': res['result']['path'],
'visit_mass_total': int(res['result']['visit_mass_total']),
'acceptance_numerator': a_num, 'acceptance_denominator': a_den,
'record_id': record.record_id, 'record_fingerprint': record.fingerprint,
'label_path': label_path, 'identity': identity}
pairs.append((row, extras))
provenance = {'input_index': input_index, 'candidate_id': candidate_id, 'record_id': record.record_id,
'record_fingerprint': record.fingerprint, 'K': K, 'n_groups': ng,
'acceptance_numerator': a_num, 'acceptance_denominator': a_den,
'early_window': list(early_w), 'late_window': list(late_w), 'identity': identity}
return InputAssessment(pairs, provenance, contributions, results, record, providers)
def _load_array(path, what):
import numpy as np
try:
out = np.load(path, allow_pickle=False)
except (EOFError, ValueError) as exc:
raise ValueError(f'cannot read {what} {path} as a .npy array: {exc}') from exc
if not isinstance(out, np.ndarray):
close = getattr(out, 'close', None)
if close:
close()
raise ValueError(f'{what} {path} is not a single .npy array (got {type(out).__name__}, e.g. an .npz archive)')
return out
def _load_record(item, cfg):
from atom_hifi.experimental.judges import array_sha256, file_sha256
from .record import LabelRecord
labels = _load_array(item['path'], 'labels')
if labels.ndim != 2 or labels.dtype.kind not in 'iu':
raise ValueError(f'labels must be a 2-D integer array (frames x entities); got shape {labels.shape}, '
f'dtype {labels.dtype}')
ids = item['entity_ids']
if isinstance(ids, str):
ids = _load_array(ids, 'entity_ids')
rec = LabelRecord.from_labels(labels, entity_ids=ids, segments=cfg['segments'],
frame_interval=cfg['frame_interval'], time_unit=cfg['time_unit'],
record_id=item['record_id'], n_groups=item['n_groups'], source=item['source'])
files = {'path': item['path_as_given'], 'file_sha256': file_sha256(item['path']),
'array_sha256': array_sha256(labels), 'shape': list(labels.shape), 'dtype': labels.dtype.str}
return rec, files
def _input_error(i, item, exc):
return ConfigError(f'inputs[{i}] ({item["candidate_id"]}, {item["path_as_given"]}): {exc}')
[docs]
def describe_record(record, *, split, preprocessing=('raw',), resolution=None, n_bins=8, runpacks=True):
"""Per-window descriptions (full, early, late) x preprocessing of one record.
Returns ``{'accounting': [...], 'aggregates': [...], 'tables': {(prep, window): {'groups',
'frequencies'}}, 'runpacks': {'<prep>_<window>': pack}}`` using :func:`~.runs.accounting`,
:func:`~.describe.group_tables`, :func:`~.describe.aggregate` and :func:`~.runs.pack_runs`.
Descriptive only: no front, threshold or filter.
"""
from .describe import aggregate, group_tables
from .runs import accounting, extract_runs, pack_runs
early_w, late_w = record.half_windows(split)
windows = (('full', record.full_window()), ('early', early_w), ('late', late_w))
out = {'accounting': [], 'aggregates': [], 'tables': {}, 'runpacks': {}}
for prep in preprocessing:
for name, window in windows:
runs = extract_runs(record, window, preprocessing=prep, n_bins=n_bins)
acc = accounting(runs, resolution=resolution, window_name=name)
groups, freq, _, _ = group_tables(runs, resolution=resolution, window_name=name)
out['accounting'].append(acc)
out['aggregates'].append(aggregate(runs, groups, acc))
out['tables'][(prep, name)] = {'groups': groups, 'frequencies': freq}
if runpacks:
out['runpacks'][f'{prep}_{name}'] = pack_runs(runs)
return out
[docs]
def check_front_identity(members):
"""Validate that the ``(CandidateRow, extras)`` members of ONE context may share a front.
With two or more members, every member must carry a declared source population and all must
share ``population_key``, ``layout_key`` and the early/late windows; otherwise
:class:`IdentityError` names the difference. A single member is always allowed. Returns
the identity record of the context (``population_status`` 'declared' or 'undeclared').
"""
idents = [(row.candidate_id, (extras or {}).get('identity') or {}) for row, extras in members]
first = idents[0][1]
record = {'population_status': 'declared' if first.get('population_key') else 'undeclared',
'population_key': first.get('population_key'), 'layout_key': first.get('layout_key'),
'source': first.get('source'), 'early_window': first.get('early_window'),
'late_window': first.get('late_window'), 'members': len(members)}
if len(members) < 2:
return record
row = members[0][0]
where = (f'context (namespace={row.namespace!r}, pca_policy={row.pca_policy!r}, '
f'preprocessing={row.preprocessing!r}, candidate={row.candidate!r}, weighting={row.weighting!r})')
undeclared = [cid for cid, ident in idents if not ident.get('population_key')]
if undeclared:
raise IdentityError(
f'{where}: candidates {undeclared} declare no source population, so a front cannot verify that its '
'members describe the same observations. Declare it for every input (CLI: inputs[i].source: {id, '
'frame_start, frame_step}; Python: LabelRecord.from_labels(..., source=SourceSelection(...))). '
'Shapes, labels and paths are never used to infer it.')
def differ(key, what):
values = {}
for cid, ident in idents:
values.setdefault(json.dumps(ident.get(key), sort_keys=True, default=str), []).append(cid)
if len(values) > 1:
groups = '; '.join(f'{ids} -> {k}' for k, ids in values.items())
raise IdentityError(f'{where}: members differ in {what}: {groups}. Candidates on one front must '
'share the observed population, its sampling and the windows; assess different '
'populations in separate namespaces.')
differ('source', 'declared source (source_id, frame_start, frame_step)')
differ('population_key', 'source population (frame count or entity ids under the same declared source)')
differ('layout_key', 'sampling/layout (frame count, entity order, segments, frame interval, time unit, '
'timestamps)')
differ('early_window', 'early window')
differ('late_window', 'late window')
return record
def _context_record(row):
d = row.descriptor_policy
return dict(zip(CONTEXT_COLUMNS, (row.namespace, d.cutoff, d.image_policy, d.reference_cell_policy, row.pca_policy,
row.preprocessing, row.window, row.sample_scope, row.candidate, row.weighting)))
[docs]
def fronts(pairs):
"""Fronts per context of ``(CandidateRow, extras)`` pairs (first-seen order; rows in input
order), after :func:`check_front_identity`. Returns (front rows, relation rows, summaries)."""
from .pareto import build_front
groups = {}
for row, extras in pairs:
groups.setdefault(row.context(), []).append((row, extras))
for group in groups.values():
check_front_identity(group)
front_rows, relation_rows, summaries = [], [], []
for group in groups.values():
members = [row for row, _ in group]
identity = check_front_identity(group)
out = build_front(members)
ctx = _context_record(members[0])
status = {r['id']: r for r in out['rows']}
ext = out['extrema']
for row, extras in group:
r = status.get(row.candidate_id)
front_rows.append({**ctx, 'candidate_id': row.candidate_id, 'K': row.K,
'acceptance_numerator': extras['acceptance_numerator'],
'acceptance_denominator': extras['acceptance_denominator'],
'status_code': row.status_code, 'status': row.status,
'front_status': r['front_status'] if r else 'register',
'dominated_by': ';'.join(r['dominated_by']) if r else '',
'unresolved_with': ';'.join(r['unresolved_with']) if r else '',
'is_ratio_maximizer': row.candidate_id in out['maximizers'],
'ratio_maximizer_unresolved': row.candidate_id in out['maximizers_unresolved'],
'front_minimum_K': row.candidate_id in ext['front_minimum_K'],
'front_maximum_A': row.candidate_id in ext['front_maximum_A']})
for rel in out['relations']:
relation_rows.append({**ctx, 'a': rel['a'], 'b': rel['b'], 'ratio_relation': rel['ratio_relation'],
'level': rel['level'], 'evidence': rel['evidence']})
summaries.append({'context': ctx, 'members': len(members), 'identity': identity,
'nondominated': [r['id'] for r in out['rows'] if r['front_status'] == 'nondominated'],
'dominated': [r['id'] for r in out['rows'] if r['front_status'] == 'dominated'],
'unresolved': [r['id'] for r in out['rows'] if r['front_status'] == 'unresolved'],
'register': [{'candidate_id': r.candidate_id, 'status': r.status} for r in out['register']],
'ratio_maximizers': out['maximizers'],
'ratio_maximizers_unresolved': out['maximizers_unresolved'], 'extrema': ext,
'escalations_beyond_level1': sum(1 for r in out['relations']
if r['level'] != 'level1_binary64_interval')})
return front_rows, relation_rows, summaries
[docs]
def candidate_record(row, extras):
"""One ``candidates.csv`` row (plus ``coverage_*`` columns) of a ``(CandidateRow, extras)`` pair."""
d = row.descriptor_policy
enc = row.enclosures or (math.nan,) * 4
out = {'candidate_id': row.candidate_id, 'namespace': row.namespace, 'cutoff': d.cutoff,
'image_policy': d.image_policy, 'reference_cell_policy': d.reference_cell_policy,
'pca_policy': row.pca_policy, 'K': row.K, 'n_groups': extras['n_groups'], 'preprocessing': row.preprocessing,
'window': row.window, 'sample_scope': row.sample_scope, 'candidate': row.candidate,
'weighting': row.weighting, 'status_code': row.status_code, 'status': row.status, 'B': row.B, 'E': row.E,
'ratio': row.ratio, 'separation_early': extras['separation_early'],
'separation_late': extras['separation_late'], 'B_exact_zero': row.B_exact_zero,
'E_exact_zero': extras['E_exact_zero'], 'B_enclosure_lo': enc[0], 'B_enclosure_hi': enc[1],
'E_enclosure_lo': enc[2], 'E_enclosure_hi': enc[3],
'production_inside_enclosure': extras['production_inside_enclosure'],
'arithmetic_path': extras['arithmetic_path'], 'visit_mass_total': extras['visit_mass_total'],
'acceptance_numerator': extras['acceptance_numerator'],
'acceptance_denominator': extras['acceptance_denominator'], 'record_id': extras['record_id'],
'record_fingerprint': extras['record_fingerprint'], 'label_path': extras['label_path']}
out.update({f'coverage_{k}': v for k, v in row.coverage.items()})
return out
# ------------------------------------------------------------------------------------- output
def _cell(value):
"""CSV text: floats round-trip exact (repr), NaN 'nan', None ''."""
import numpy as np
if value is None:
return ''
if isinstance(value, (bool, np.bool_)):
return 'True' if value else 'False'
if isinstance(value, np.integer):
return str(int(value))
if isinstance(value, np.floating):
value = float(value)
if isinstance(value, float):
return repr(value)
return str(value)
def _json_value(value):
import numpy as np
if isinstance(value, dict):
return {str(k): _json_value(v) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [_json_value(v) for v in value]
if isinstance(value, Fraction):
return {'numerator': value.numerator, 'denominator': value.denominator}
if isinstance(value, (bool, np.bool_)):
return bool(value)
if isinstance(value, np.integer):
return int(value)
if isinstance(value, (float, np.floating)):
value = float(value)
return value if math.isfinite(value) else None
if isinstance(value, np.ndarray):
return _json_value(value.tolist())
return value
def _json_text(value):
return json.dumps(_json_value(value), indent=1) + '\n'
def _csv_text(rows, columns=None):
if columns is None:
if not rows:
return ''
columns = list(rows[0])
buf = io.StringIO()
w = csv.writer(buf, lineterminator='\n')
w.writerow(columns)
for r in rows:
w.writerow([_cell(r.get(c)) for c in columns])
return buf.getvalue()
def _write(path, text):
tmp = path.with_name(path.name + '.part')
tmp.write_text(text)
tmp.replace(path)
def _sanitize(text):
return re.sub(r'[^A-Za-z0-9._-]+', '_', str(text)).strip('._') or 'input'
def _description_dirs(cfg):
return [f'{i:02d}_{_sanitize(item["candidate_id"])}' for i, item in enumerate(cfg['inputs'])]
def _description_files(cfg):
"""Exactly the files one description directory of this run will hold."""
names = {'DESCRIPTION.json'}
if cfg['exports']['descriptions']:
names |= {'accounting.json', 'aggregates.json'}
names |= {f'{kind}_{prep}_{w}.csv' for kind in ('groups', 'frequencies') for prep in cfg['preprocessing']
for w in ('full', 'early', 'late')}
if cfg['exports']['runpacks']:
names.add('runpacks.npz')
return names
def _check_collisions(out, cfg, overwrite):
present = [name for name in OUTPUTS if (out / name).exists()]
desc = out / DESCRIPTIONS_DIR
planned = set(_description_dirs(cfg)) if any(cfg['exports'].values()) else set()
if desc.exists():
present.append(DESCRIPTIONS_DIR + '/')
if present and not overwrite:
raise FileExistsError(f'{out}: outputs already present {present}; pass --overwrite (overwrite=True) to '
'replace them')
if desc.exists():
stale = sorted(p.name for p in desc.iterdir() if p.name not in planned)
files = _description_files(cfg)
for name in sorted(planned):
if (desc / name).is_dir():
stale += sorted(f'{name}/{p.name}' for p in (desc / name).iterdir() if p.name not in files)
elif (desc / name).exists():
stale.append(name)
if stale:
raise FileExistsError(f'{desc}: entries {stale} would not be rewritten by this run and would remain '
'next to new results; remove them yourself (they are never deleted automatically)')
for name in OUTPUTS:
if (out / name).is_dir():
raise FileExistsError(f'{out / name} is a directory; refusing to replace it')
def _manifest(out, names):
from atom_hifi.experimental.judges import file_sha256
return {name: {'sha256': file_sha256(out / name), 'bytes': (out / name).stat().st_size} for name in names}
def _write_descriptions(out, cfg, rel_dir, item, rec, files):
import numpy as np
d = out / DESCRIPTIONS_DIR / rel_dir
d.mkdir(parents=True, exist_ok=True)
desc = describe_record(rec, split=cfg['split'], preprocessing=cfg['preprocessing'], resolution=item['K'],
n_bins=cfg['n_bins'], runpacks=cfg['exports']['runpacks'])
written = []
if cfg['exports']['descriptions']:
_write(d / 'accounting.json', _json_text(desc['accounting']))
_write(d / 'aggregates.json', _json_text(desc['aggregates']))
written += ['accounting.json', 'aggregates.json']
for (prep, name), tab in desc['tables'].items():
for kind in ('groups', 'frequencies'):
fname = f'{kind}_{prep}_{name}.csv'
_write(d / fname, _csv_text(tab[kind]))
written.append(fname)
if cfg['exports']['runpacks']:
flat = {f'{name}__{k}': np.asarray(v) for name, pk in desc['runpacks'].items() for k, v in pk.items()}
tmp = d / 'runpacks.part.npz'
np.savez_compressed(tmp, **flat)
tmp.replace(d / 'runpacks.npz')
written.append('runpacks.npz')
info = {'schema_version': SCHEMA_VERSION, 'candidate_id': item['candidate_id'], 'record_id': rec.record_id,
'record_fingerprint': rec.fingerprint, 'label_array_sha256': files['array_sha256'],
'resolution': item['K'], 'windows': {'full': list(rec.full_window()),
'early': list(rec.half_windows(cfg['split'])[0]),
'late': list(rec.half_windows(cfg['split'])[1])},
'preprocessing': list(cfg['preprocessing']), 'n_bins': cfg['n_bins'],
'files': _manifest(d, written),
'notes': 'floats are JSON numbers (NaN -> null) and repr() in CSV; runpack keys are '
'<preprocessing>_<window>__<field> (see runs.pack_runs); an empty CSV means no row'}
_write(d / 'DESCRIPTION.json', _json_text(info))
return {'directory': f'{DESCRIPTIONS_DIR}/{rel_dir}', 'files': info['files']}
def _echo(cfg):
"""Resolved configuration for the summary, without absolute paths."""
from atom_hifi.experimental.judges import array_sha256
import numpy as np
keep = ('frame_interval', 'time_unit', 'split', 'segments', 'preprocessing', 'candidates', 'weightings', 'n_bins',
'exact_budget', 'exports')
out = {k: cfg[k] for k in keep}
out['output_dir'] = cfg['output_dir_as_given']
out['inputs'] = []
for item in cfg['inputs']:
ids = item['entity_ids_as_given']
if isinstance(ids, list):
arr = np.asarray(ids)
ids = {'inline_count': len(ids), 'first': ids[0] if ids else None, 'last': ids[-1] if ids else None,
'array_sha256': array_sha256(arr)}
src = item['source']
out['inputs'].append({'path': item['path_as_given'], 'candidate_id': item['candidate_id'], 'K': item['K'],
'n_groups': item['n_groups'], 'namespace': item['namespace'],
'descriptor_policy': item['descriptor_policy'], 'pca_policy': item['pca_policy'],
'entity_ids': ids, 'record_id': item['record_id'],
'source': src.as_dict() if src is not None else None})
return out
[docs]
def run(config, *, overwrite=False, base_dir=None):
"""Run one configuration (path or mapping, see :func:`resolve_config`) and write its outputs.
Existing outputs are refused unless ``overwrite``; collisions are checked before anything is
written. Returns the summary written to ``run_summary.json``.
"""
import numpy as np
import atom_hifi
from .separation import STATUS
t0 = time.monotonic()
cfg = resolve_config(config, base_dir=base_dir)
out = Path(cfg['output_dir'])
_check_collisions(out, cfg, overwrite)
assessed = []
for i, item in enumerate(cfg['inputs']):
try:
a = assess_input(item, cfg, input_index=i)
except (ValueError, TypeError, OSError) as exc:
if isinstance(exc, (ConfigError, FileExistsError)):
raise
raise _input_error(i, item, exc) from exc
a.record = a.results = None # one input's labels in memory at a time (reloaded for exports)
assessed.append(a)
pairs = [p for a in assessed for p in a.pairs]
records = [candidate_record(row, extras) for row, extras in pairs]
front_rows, relation_rows, summaries = fronts(pairs)
contributions = [c for a in assessed for c in a.contributions]
columns = list(CANDIDATE_COLUMNS) + [c for c in records[0] if c.startswith('coverage_')]
out.mkdir(parents=True, exist_ok=True)
_write(out / 'candidates.csv', _csv_text(records, columns))
_write(out / 'candidates.json', _json_text(records))
_write(out / 'fronts.csv', _csv_text(front_rows, FRONT_COLUMNS))
_write(out / 'front_relations.csv', _csv_text(relation_rows, RELATION_COLUMNS))
_write(out / 'group_contributions.csv', _csv_text(contributions, CONTRIBUTION_COLUMNS))
descriptions = []
if any(cfg['exports'].values()):
for item, a, rel_dir in zip(cfg['inputs'], assessed, _description_dirs(cfg)):
rec, files = _load_record(item, cfg)
if files['array_sha256'] != a.provenance['array_sha256'] or rec.fingerprint != a.provenance['record_fingerprint']:
raise RuntimeError(f'{item["path"]}: the label file changed during the run')
descriptions.append(_write_descriptions(out, cfg, rel_dir, item, rec, files))
del rec
exact_calls = sum(p.calls for a in assessed for p in a.providers.values() if p is not None)
summary = {
'schema_version': SCHEMA_VERSION, 'tool': 'atom_hifi.experimental.temporal', 'experimental': True,
'atom_hifi_version': atom_hifi.__version__, 'module': __name__,
'config_name': cfg['config_name'], 'config_sha256': cfg['config_sha256'],
'config_sha256_definition': cfg['config_sha256_definition'],
'resolved_config': _echo(cfg), 'inputs': [a.provenance for a in assessed],
'window': WINDOW, 'sample_scope': SAMPLE_SCOPE, 'acceptance_definition': ACCEPTANCE_DEFINITION,
'status_codes': list(STATUS), 'status_explanations': STATUS_EXPLANATIONS,
'front_status_explanations': FRONT_STATUS_EXPLANATIONS, 'ratio_note': RATIO_NOTE,
'production_inside_enclosure_note': ('False is not an error: the canonical production float can round just '
'outside the enclosure of the exact value; fronts use the enclosures'),
'identity_contract': ('a context with two or more members requires a declared and equal source population, '
'an equal layout and equal windows (check_front_identity)'),
'exact_escalation': {'enabled': bool(cfg['exact_budget']), 'budget': cfg['exact_budget'],
'exact_component_evaluations': exact_calls,
'note': 'without a budget, pairs not ordered by exact zeros or Level-1 stay unresolved'},
'fronts': summaries, 'candidate_rows': len(records), 'front_rows': len(front_rows),
'relation_rows': len(relation_rows), 'contribution_rows': len(contributions), 'descriptions': descriptions,
'environment': {'python': sys.version.split()[0], 'numpy': np.__version__},
'outputs': _manifest(out, OUTPUTS[:-1]), 'wall_seconds': time.monotonic() - t0}
_write(out / 'run_summary.json', _json_text(summary))
return summary
# ------------------------------------------------------------------------------------- reading
_INT_COLUMNS = {'K', 'n_groups', 'status_code', 'acceptance_numerator', 'acceptance_denominator', 'input_index',
'group', 'n_early', 'n_late', 's_early', 's_late', 'visit_mass', 'ntau', 'nw1', 'den'}
_BOOL_COLUMNS = {'is_ratio_maximizer', 'ratio_maximizer_unresolved', 'front_minimum_K', 'front_maximum_A'}
_FLOAT_COLUMNS = {'mu_early', 'mu_late', 'dmu', 'w1', 'weight_visit', 'weight_equal', 'E_contribution_tau_visit',
'E_contribution_tau_equal', 'E_contribution_P_visit', 'E_contribution_P_equal'}
def _typed(rows):
out = []
for r in rows:
t = {}
for k, v in r.items():
if k in _INT_COLUMNS and v != '':
v = int(v)
elif k in _BOOL_COLUMNS:
v = v == 'True'
elif k in _FLOAT_COLUMNS and v != '':
v = float(v)
t[k] = v
out.append(t)
return out
[docs]
def read_results(output_dir, *, verify=True):
"""Read a finished run's outputs (independent of where they were written).
Returns ``{'summary', 'candidates', 'fronts', 'relations', 'contributions'}``; candidates
come from ``candidates.json`` (typed, NaN -> None), the CSV tables with integer/boolean
columns typed. ``verify`` checks ``schema_version`` and every file's sha256 against the
manifest in ``run_summary.json``. Fronts are read as persisted, never rebuilt.
"""
out = Path(output_dir)
summary = json.loads((out / 'run_summary.json').read_text())
if verify:
if summary.get('schema_version') != SCHEMA_VERSION:
raise ValueError(f'{out}: schema_version {summary.get("schema_version")!r} != {SCHEMA_VERSION!r}')
from atom_hifi.experimental.judges import file_sha256
bad = [name for name, m in summary['outputs'].items() if file_sha256(out / name) != m['sha256']]
for entry in summary.get('descriptions', []):
d = out / entry['directory']
bad += [f'{entry["directory"]}/{name}' for name, m in entry['files'].items()
if file_sha256(d / name) != m['sha256']]
if bad:
raise ValueError(f'{out}: files changed since the run (sha256 mismatch): {bad}')
def table(name):
with open(out / name, newline='') as f:
return _typed(list(csv.DictReader(f)))
return {'summary': summary, 'candidates': json.loads((out / 'candidates.json').read_text()),
'fronts': table('fronts.csv'), 'relations': table('front_relations.csv'),
'contributions': table('group_contributions.csv')}