Source code for atom_hifi.experimental.temporal.example

"""Small deterministic synthetic example of the EXPERIMENTAL temporal assessment (installed).

One synthetic record (``synthetic:latent_record_1``, 48 frames x 6 entities) is labelled at
three resolutions K2/K3/K4 -- deterministic partitions of the same latent fine states, with
candidate-specific abstention (-1) -- so all three describe ONE observed population and may share
a front.  ``K5_padded`` reuses K3's labels at a larger nominal K (dominated by K3 through an exact
Level-2b equality), K3 also appears under a second PCA-policy identity (its own fronts), and a
separate record (``synthetic:repeat_record``, own namespace) has identical early/late halves
(E = 0 < B: the non-finite register).  The latent chain changes its stay probabilities at the
split, so the early/late comparison is informative (E > 0).

The arrays are generated by ``examples/experimental_temporal/make_labels.py`` in the repository
(seed and partitions documented there) and embedded here as text rows: one string per frame,
one character per entity, ``-`` = rejected.  Descriptor/PCA metadata are placeholders for the
context identity only; nothing here is a PIT setting or a recommendation.

    python -m atom_hifi.experimental.temporal --write-example DIR
    python -m atom_hifi.experimental.temporal --config DIR/config.yaml
"""
from __future__ import annotations

from pathlib import Path

LABELS = {
    'labels_k2.npy': (
        '-11000', '110000', '010000', '010001', '010001', '010001', '010011', '010011',
        '000000', '00010-', '011100', '0101-0', '000101', '000-00', '000000', '00-100',
        '100110', '0-1110', '000110', '-01000', '101010', '101010', '101010', '101011',
        '101011', '111010', '010000', '101001', '11110-', '101100', '1011-0', '110101',
        '110-00', '110000', '11-100', '110100', '1-0000', '110000', '-10100', '010111',
        '010111', '010010', '010010', '000000', '000010', '001010', '000100', '00010-',
    ),
    'labels_k3.npy': (
        '211000', '210000', '010010', '010011', '010011', '010011', '010021', '010021',
        '010001', '011102', '022200', '020200', '000201', '000200', '000000', '000200',
        '200220', '001220', '001220', '002000', '102010', '102020', '101020', '101012',
        '101012', '211010', '010000', '112011', '222211', '202210', '202210', '121112',
        '121011', '121011', '221211', '221211', '221011', '121011', '121111', '121122',
        '121122', '121121', '111121', '111101', '111021', '112021', '111201', '111201',
    ),
    'labels_k4.npy': (
        '-22000', '3200-0', '020010', '020012', '020012', '020012', '0-0032', '02003-',
        '010001', '011203', '033300', '030300', '00-302', '000300', '000000', '000300',
        '300330', '002330', '001-30', '003000', '203020', '203030', '202030', '-02023',
        '2020-3', '322020', '020000', '213012', '333312', '3-3310', '30331-', '231213',
        '231011', '231011', '331311', '33-311', '331011', '231011', '131211', '131233',
        '131233', '131-31', '121131', '111101', '111031', '113031', '-11301', '1113-1',
    ),
    'labels_k2_repeat.npy': (
        '011110', '011110', '010110', '010110', '011110', '011110', '010100', '010100',
        '010100', '010100', '010100', '010100', '010000', '010000', '000000', '000000',
        '001000', '000000', '000000', '000010', '000010', '000010', '000010', '000010',
        '011110', '011110', '010110', '010110', '011110', '011110', '010100', '010100',
        '010100', '010100', '010100', '010100', '010000', '010000', '000000', '000000',
        '001000', '000000', '000000', '000010', '000010', '000010', '000010', '000010',
    ),
}

CONFIG_YAML = """\
# Synthetic example for the EXPERIMENTAL labels-first temporal assessment (opt-in).
#   python -m atom_hifi.experimental.temporal --config config.yaml
# One synthetic record (48 frames x 6 entities) labelled at K2/K3/K4: partitions of the same latent
# states (-1 = rejected), so the three share one declared source population and one front.  A
# separate record with identical halves lives in its own namespace.  Relative paths resolve
# against this file's directory.  B/E is a temporal separation diagnostic, not HiFi's fidelity L/R.
# Descriptor/PCA fields are placeholders for the context identity of synthetic labels.
output_dir: results
frame_interval: 1
time_unit: frame
windows:
  split: 24                      # explicit early/late split frame (no default)
preprocessing: [raw, bridge1]
candidates: [tau, P]
weightings: [visit, equal]
exact_budget: {max_seconds: 30}  # optional Level-2b escalation; omit -> unordered pairs stay 'unresolved'
exports: {descriptions: true, runpacks: true}

inputs:
  - path: labels_k2.npy
    candidate_id: K2
    K: 2
    namespace: synthetic
    descriptor_policy: {cutoff: not_applicable, image_policy: not_applicable, reference_cell_policy: not_applicable}
    pca_policy: {kind: fixed_k, k: 2}
    source: {id: 'synthetic:latent_record_1'}
  - path: labels_k3.npy
    candidate_id: K3
    K: 3
    namespace: synthetic
    descriptor_policy: {cutoff: not_applicable, image_policy: not_applicable, reference_cell_policy: not_applicable}
    pca_policy: {kind: fixed_k, k: 2}
    source: {id: 'synthetic:latent_record_1'}
  - path: labels_k4.npy
    candidate_id: K4
    K: 4
    namespace: synthetic
    descriptor_policy: {cutoff: not_applicable, image_policy: not_applicable, reference_cell_policy: not_applicable}
    pca_policy: {kind: fixed_k, k: 2}
    source: {id: 'synthetic:latent_record_1'}
  - path: labels_k3.npy          # K3's labels at a larger nominal K: equal ratio and acceptance -> dominated by K3
    candidate_id: K5_padded
    K: 5
    namespace: synthetic
    descriptor_policy: {cutoff: not_applicable, image_policy: not_applicable, reference_cell_policy: not_applicable}
    pca_policy: {kind: fixed_k, k: 2}
    source: {id: 'synthetic:latent_record_1'}
  - path: labels_k3.npy          # same labels under another PCA-policy identity: its own fronts
    candidate_id: K3
    K: 3
    namespace: synthetic
    descriptor_policy: {cutoff: not_applicable, image_policy: not_applicable, reference_cell_policy: not_applicable}
    pca_policy: {kind: native_auto}
    source: {id: 'synthetic:latent_record_1'}
  - path: labels_k2_repeat.npy   # a different record with identical halves: E = 0 < B (register)
    candidate_id: K2_repeat
    K: 2
    namespace: synthetic_repeat
    descriptor_policy: {cutoff: not_applicable, image_policy: not_applicable, reference_cell_policy: not_applicable}
    pca_policy: {kind: fixed_k, k: 2}
    source: {id: 'synthetic:repeat_record'}
"""


[docs] def arrays(): """The example label arrays (int8, frames x entities, -1 = rejected).""" import numpy as np return {name: np.asarray([[-1 if ch == '-' else int(ch) for ch in row] for row in rows], dtype=np.int8) for name, rows in LABELS.items()}
[docs] def config(): """The example configuration as a mapping (equal to :data:`CONFIG_YAML`); paths are relative.""" na = {'cutoff': 'not_applicable', 'image_policy': 'not_applicable', 'reference_cell_policy': 'not_applicable'} fixed = {'kind': 'fixed_k', 'k': 2} one = {'id': 'synthetic:latent_record_1'} def item(path, cid, K, pca=fixed, namespace='synthetic', source=one): return {'path': path, 'candidate_id': cid, 'K': K, 'namespace': namespace, 'descriptor_policy': dict(na), 'pca_policy': dict(pca), 'source': dict(source)} return {'output_dir': 'results', 'frame_interval': 1, 'time_unit': 'frame', 'windows': {'split': 24}, 'preprocessing': ['raw', 'bridge1'], 'candidates': ['tau', 'P'], 'weightings': ['visit', 'equal'], 'exact_budget': {'max_seconds': 30}, 'exports': {'descriptions': True, 'runpacks': True}, 'inputs': [item('labels_k2.npy', 'K2', 2), item('labels_k3.npy', 'K3', 3), item('labels_k4.npy', 'K4', 4), item('labels_k3.npy', 'K5_padded', 5), item('labels_k3.npy', 'K3', 3, {'kind': 'native_auto'}), item('labels_k2_repeat.npy', 'K2_repeat', 2, namespace='synthetic_repeat', source={'id': 'synthetic:repeat_record'})]}
[docs] def write_example(directory, *, overwrite=False): """Write ``config.yaml`` and the label arrays into ``directory`` (created if needed). Existing files are refused unless ``overwrite``. Returns ``{'config': path, 'arrays': [paths]}``. """ import numpy as np d = Path(directory) targets = [d / 'config.yaml'] + [d / name for name in LABELS] present = [p.name for p in targets if p.exists()] if present and not overwrite: raise FileExistsError(f'{d}: example files already present {present}; pass --overwrite to replace them') d.mkdir(parents=True, exist_ok=True) for name, a in arrays().items(): np.save(d / name, a) (d / 'config.yaml').write_text(CONFIG_YAML) return {'config': d / 'config.yaml', 'arrays': [d / name for name in LABELS]}