Source code for atom_hifi.experimental.temporal.pareto

"""Exact three-objective fronts: minimise K, maximise the finite ratio B/E, maximise acceptance.

Port of the PF ``pareto.py`` front logic and ``exact_verify.compare_f`` comparator.

* a dominates b iff ``K_a <= K_b``, ``ratio_a >= ratio_b`` and ``A_a >= A_b`` with one strict;
* ``K`` and the acceptance ``A`` are exact (int, or :class:`fractions.Fraction` / an integer
  numerator over a common denominator); floats are refused;
* a ratio order is certified only by exact zeros (integer identities), separated Level-1
  intervals of ``B_a*E_b`` vs ``B_b*E_a``, or exact ``Fraction`` cross-products computed on
  demand under a recorded budget; otherwise the relation is ``'unresolved'`` (no relative-gap rule);
* non-finite statuses go to a register and never enter a front; extrema keep ties and are taken
  over the non-dominated members (unresolved members are reported through ``front_status`` and
  the ratio-maximizer ``unresolved`` list, never ranked);
* :func:`build_front` refuses rows that mix namespace, descriptor policy (cutoff, image policy,
  reference-cell policy), PCA policy, preprocessing, window, sample scope, candidate or weighting.
  It does NOT verify that the rows describe one source population: callers pass already
  validated rows (the public route :func:`.assessment.fronts` checks the declared population,
  layout and window identities first).
"""
from __future__ import annotations

from dataclasses import dataclass, field, fields
from fractions import Fraction
from itertools import combinations
import math
import numbers
from typing import Any, Callable, Mapping, Optional

from .separation import STATUS, BudgetExhausted

RELATIONS = ('gt', 'lt', 'eq', 'unresolved')
_FLIP = {'gt': 'lt', 'lt': 'gt'}


[docs] class VerifiedPoint: """A finite-status point: exact-zero flag, Level-1 enclosure and an optional exact provider. ``exact`` is a zero-argument callable returning exact ``(B, E)`` Fractions; it may raise :class:`BudgetExhausted`. ``budget`` is recorded in the comparison evidence. Both outcomes are cached per point: an exhausted budget is not paid again by later comparisons. """ def __init__(self, label, b_zero, enclosure, exact=None, budget=None): self.label = label self.b_zero = bool(b_zero) self.l1 = tuple(float(x) for x in enclosure) # (B_lo, B_hi, E_lo, E_hi) self._exact = exact self.budget = budget self._cache = None self._exhausted = None
[docs] def exact_values(self): if self._exact is None: raise BudgetExhausted(f'{self.label}: no exact provider') if self._exhausted is not None: raise BudgetExhausted(str(self._exhausted)) if self._cache is None: try: B, E = self._exact() except BudgetExhausted as exc: self._exhausted = exc raise self._cache = (Fraction(B), Fraction(E)) return self._cache
[docs] def synthetic_point(label, B, E, budget=None): """VerifiedPoint from exact Fractions (tests/synthetic cases). ``float(Fraction)`` is correctly rounded, so one ulp either side encloses the exact value.""" B = Fraction(B) E = Fraction(E) def l1(x): f = float(x) return max(0.0, math.nextafter(f, -math.inf)), math.nextafter(f, math.inf) return VerifiedPoint(label, B == 0, l1(B) + l1(E), lambda: (B, E), budget)
def _prod_iv(blo, bhi, elo, ehi): return math.nextafter(blo * elo, -math.inf), math.nextafter(bhi * ehi, math.inf)
[docs] def compare_ratio(a, b): """``(relation, level, evidence)`` for ratio_a vs ratio_b of two finite points (E > 0 exactly). relation in {'gt', 'lt', 'eq', 'unresolved'}; level in {'exact_zero', 'level1_binary64_interval', 'level2b_exact_fraction', 'unresolved'}. """ if a.b_zero and b.b_zero: return 'eq', 'exact_zero', 'both B exactly zero with E>0: ratio_a = ratio_b = 0' if a.b_zero: return 'lt', 'exact_zero', 'B_a exactly zero, B_b > 0 exactly' if b.b_zero: return 'gt', 'exact_zero', 'B_b exactly zero, B_a > 0 exactly' ba_lo, ba_hi, ea_lo, ea_hi = a.l1 bb_lo, bb_hi, eb_lo, eb_hi = b.l1 x_lo, x_hi = _prod_iv(max(ba_lo, 0.0), ba_hi, max(eb_lo, 0.0), eb_hi) # B_a * E_b y_lo, y_hi = _prod_iv(max(bb_lo, 0.0), bb_hi, max(ea_lo, 0.0), ea_hi) # B_b * E_a evidence = f'L1 B_a*E_b in [{x_lo!r},{x_hi!r}] vs B_b*E_a in [{y_lo!r},{y_hi!r}]' if x_lo > y_hi: return 'gt', 'level1_binary64_interval', evidence if x_hi < y_lo: return 'lt', 'level1_binary64_interval', evidence budget = {'a': a.budget, 'b': b.budget} try: Ba, Ea = a.exact_values() Bb, Eb = b.exact_values() except BudgetExhausted as exc: return 'unresolved', 'unresolved', evidence + f'; exact escalation not completed (budget {budget}): {exc}' x = Ba * Eb y = Bb * Ea evidence += f'; L2b exact B_a*E_b={x} vs B_b*E_a={y} (budget {budget})' return ('gt' if x > y else 'lt' if x < y else 'eq'), 'level2b_exact_fraction', evidence
def _exact_objective(value, name, integral=False): kind = numbers.Integral if integral else numbers.Rational if isinstance(value, bool) or not isinstance(value, kind): what = 'an exact integer' if integral else 'an exact integer or Fraction' raise TypeError(f'{name} must be {what}, got {type(value).__name__} {value!r}') return value
[docs] def relations(points, compare): """Pairwise ratio relations among finite points: ``rel[(i, j)]`` for ratio_i vs ratio_j.""" rel = {} evidence = {} for i, j in combinations(range(len(points)), 2): r, level, ev = compare(points[i], points[j]) if r not in RELATIONS: raise ValueError(f'unknown relation {r!r}') rel[(i, j)] = r rel[(j, i)] = _FLIP.get(r, r) evidence[(i, j)] = (r, level, ev) return rel, evidence
[docs] def dominance(points, rel): """``dominated_by`` and ``unresolved`` maps for finite points with exact keys K and A.""" dominated_by = {j: [] for j in range(len(points))} unresolved = {j: [] for j in range(len(points))} for i in range(len(points)): for j in range(len(points)): if i == j: continue a, b = points[i], points[j] if not (a['K'] <= b['K'] and a['A'] >= b['A']): continue r = rel[(i, j)] strict_ka = a['K'] < b['K'] or a['A'] > b['A'] if r == 'gt' or (r == 'eq' and strict_ka): dominated_by[j].append(i) elif r == 'unresolved': unresolved[j].append(i) return dominated_by, unresolved
[docs] def front(points, compare): """points: dicts with 'id', exact 'K' and 'A', 'status' (STATUS string) and comparator payload. Returns ``(rows, witnesses, relation_table, register, rel)``; ``rows`` carry ``front_status`` in {'nondominated', 'dominated', 'unresolved'} for finite points. """ for p in points: _exact_objective(p['K'], 'K', integral=True) _exact_objective(p['A'], 'A') if p['status'] not in STATUS: raise ValueError(f'unknown status {p["status"]!r}') if len({isinstance(p['A'], numbers.Integral) for p in points}) > 1: raise ValueError('acceptance must be all integer numerators over one common denominator or all Fractions') finite = [p for p in points if p['status'] == 'finite'] register = [p for p in points if p['status'] != 'finite'] rel, evidence = relations(finite, compare) dom, unres = dominance(finite, rel) rows = [] for j, p in enumerate(finite): status = 'dominated' if dom[j] else ('unresolved' if unres[j] else 'nondominated') rows.append({'id': p['id'], 'front_status': status, 'dominated_by': [finite[i]['id'] for i in dom[j]], 'unresolved_with': [finite[i]['id'] for i in unres[j]]}) witnesses = [] for j, ids in dom.items(): for i in ids: a, b = finite[i], finite[j] r, level, ev = evidence[(min(i, j), max(i, j))] witnesses.append({'dominator': a['id'], 'dominated': b['id'], 'K_dominator': a['K'], 'K_dominated': b['K'], 'A_relation': 'gt' if a['A'] > b['A'] else 'eq', 'K_relation': 'lt' if a['K'] < b['K'] else 'eq', 'ratio_relation': rel[(i, j)], 'ratio_verification_level': level, 'ratio_evidence': ev}) table = [{'a': finite[i]['id'], 'b': finite[j]['id'], 'ratio_relation': r, 'level': level, 'evidence': ev} for (i, j), (r, level, ev) in sorted(evidence.items())] return rows, witnesses, table, register, rel
[docs] def ratio_maximizers(points, compare=None, rel=None): """Finite-ratio maximizers with ties kept; returns ``(maximizers, unresolved_candidates)``.""" finite = [p for p in points if p['status'] == 'finite'] if not finite: return [], [] if rel is None: rel, _ = relations(finite, compare) maxim, unresolved = [], [] for j in range(len(finite)): others = [i for i in range(len(finite)) if i != j] if any(rel[(i, j)] == 'gt' for i in others): continue (unresolved if any(rel[(i, j)] == 'unresolved' for i in others) else maxim).append(finite[j]['id']) return maxim, unresolved
[docs] def extrema(points, rows, maximizers, unresolved): """Observed property extrema of one front (ties kept): ratio maximum (+unresolved), and the minimum K / maximum A among non-dominated members. Descriptive, never an operating point.""" status = {r['id']: r['front_status'] for r in rows} members = [p for p in points if status.get(p['id']) == 'nondominated'] min_k = min((p['K'] for p in members), default=None) max_a = max((p['A'] for p in members), default=None) return {'finite_ratio_maximum': list(maximizers), 'finite_ratio_maximum_unresolved': list(unresolved), 'front_minimum_K': [p['id'] for p in members if p['K'] == min_k], 'front_maximum_A': [p['id'] for p in members if p['A'] == max_a]}
# ------------------------------------------------------------------------------- candidate table
[docs] @dataclass(frozen=True) class DescriptorPolicy: """Descriptor identity of a candidate: cutoff, periodic-image policy, reference-cell policy.""" cutoff: Any image_policy: str reference_cell_policy: Any
CONTEXT_FIELDS = ('namespace', 'descriptor_policy.cutoff', 'descriptor_policy.image_policy', 'descriptor_policy.reference_cell_policy', 'pca_policy', 'preprocessing', 'window', 'sample_scope', 'candidate', 'weighting')
[docs] @dataclass(frozen=True) class CandidateRow: """One candidate-table row (one K, one candidate, one weighting, one context). ``enclosures`` are the Level-1 ARITHMETIC enclosures ``(B_lo, B_hi, E_lo, E_hi)``; ``acceptance`` is exact (``Fraction(accepted, all entity-frames)`` or an int numerator over a denominator shared by every row of the front); ``exact`` optionally returns exact (B, E). """ candidate_id: str namespace: str descriptor_policy: DescriptorPolicy pca_policy: str K: int preprocessing: str window: str sample_scope: str candidate: str weighting: str B: float E: float ratio: float status_code: int enclosures: Optional[tuple] coverage: Mapping = field(default_factory=dict, compare=False) acceptance: Any = None B_exact_zero: Optional[bool] = None exact: Optional[Callable] = field(default=None, compare=False, repr=False) exact_budget: Optional[Mapping] = field(default=None, compare=False) def __post_init__(self): _exact_objective(self.K, 'K', integral=True) if self.acceptance is not None: _exact_objective(self.acceptance, 'acceptance') if self.status_code not in range(len(STATUS)): raise ValueError(f'status_code must be one of 0..{len(STATUS) - 1}') if not isinstance(self.descriptor_policy, DescriptorPolicy): raise TypeError('descriptor_policy must be a DescriptorPolicy') @property def status(self): return STATUS[self.status_code]
[docs] def context(self): d = self.descriptor_policy return (self.namespace, d.cutoff, d.image_policy, d.reference_cell_policy, self.pca_policy, self.preprocessing, self.window, self.sample_scope, self.candidate, self.weighting)
[docs] def verified_point(self): if self.status_code != 0: return None if self.enclosures is None or self.B_exact_zero is None: raise ValueError(f'{self.candidate_id}: a finite point needs Level-1 enclosures and B_exact_zero') return VerifiedPoint(self.candidate_id, self.B_exact_zero, self.enclosures, self.exact, self.exact_budget)
[docs] def check_single_context(rows): """Raise ValueError naming every context field that differs between rows.""" contexts = [r.context() for r in rows] mixed = [name for k, name in enumerate(CONTEXT_FIELDS) if len({c[k] for c in contexts}) > 1] if mixed: raise ValueError(f'refusing to build a front that mixes {", ".join(mixed)}; build one front per context')
[docs] def build_front(rows): """Front, register, witnesses, relations and extrema of candidate rows sharing one context.""" rows = list(rows) for r in rows: if not isinstance(r, CandidateRow): raise TypeError('build_front takes CandidateRow objects') if r.acceptance is None: raise ValueError(f'{r.candidate_id}: acceptance is required for a front') check_single_context(rows) if len({r.candidate_id for r in rows}) != len(rows): raise ValueError('candidate_id must be unique within a front') points = [{'id': r.candidate_id, 'K': r.K, 'A': r.acceptance, 'status': r.status, 'vp': r.verified_point(), 'row': r} for r in rows] compare = lambda p, q: compare_ratio(p['vp'], q['vp']) front_rows, witnesses, table, register, rel = front(points, compare) finite = [p for p in points if p['status'] == 'finite'] maxim, unresolved = ratio_maximizers(finite, rel=rel) return {'rows': front_rows, 'witnesses': witnesses, 'relations': table, 'register': [p['row'] for p in register], 'maximizers': maxim, 'maximizers_unresolved': unresolved, 'extrema': extrema(finite, front_rows, maxim, unresolved)}
CANDIDATE_FIELDS = tuple(f.name for f in fields(CandidateRow) if f.name not in ('exact', 'exact_budget'))