loopable / platform /core /measure_resolve.py
fsanyoto's picture
Deploy AIOS web (React glide grid + FastAPI slice)
c14ceee verified
Raw
History Blame Contribute Delete
9.09 kB
"""core/measure_resolve.py — HOST-NEUTRAL measure resolution (EXIT-5, SHELL.md ladder step 3).
The `_cl_measure_offer` / `_cl_resolve_measures` / `_cl_measure_columns` family, OUT of app.py:
one implementation, two hosts. The Streamlit adapter passes its `st.session_state` memos and its
pending-view read; the API passes runtime-level dicts and no pending view (over HTTP the
`view_upsert` lands via `/grid/events` BEFORE the workspace re-read, so the saved views already
carry the freshest rule — there is no mid-run component value to read).
No streamlit import, no app import — harness only, so both adapters can call it.
Memo discipline (unchanged from the app.py originals):
* keys carry (stamp, scope, POOL IDENTITY, question) — the pool's identity, never its size,
because Cohort passes a different pool per cohort and two same-sized cohorts must not be
handed each other's answers.
* `column_values` memoises permanent failures as None but NEVER a transient one (datastore
still warming after a restart) — a memoised transient means blank measure cells that stay
blank for the whole session long after the data landed.
* both memos are bounded by clearing wholesale past a cap — session/runtime-lifetime caches,
not leaks.
"""
import hashlib
def offer(team_id=None, on_error=None):
"""The measures the condition builder / measure columns may use, or [] if the semantic
store is unavailable. The caller's BU narrows the offer (a company-level measure cannot be
answered for one business unit)."""
try:
from harness import measure_filter as _mf
return _mf.measure_fields(team_id)
except Exception as e: # no store / model problem -> offer nothing
if on_error:
on_error('measures', e)
return []
def _pool_key(allowed_pids):
"""The pool's identity, hashed. Part of every ANSWER, not just the question: the zero-group
is reconstructed from it (a customer with no orders in the window has no row for SQL to
group, so `Sales < 100` would miss exactly the lapsed customers the filter is for)."""
return hashlib.blake2b(
repr(sorted(allowed_pids)).encode(), digest_size=8).hexdigest()
def condition_sets(view_configs, pending_cfg, team_id, allowed_pids, today, stamp, memo,
on_error=None):
"""Every measure condition across `view_configs` (+ the optional pending config, which goes
LAST and overwrites — the freshest statement of what the user wants) -> `{ruleId: [pid,…]}`.
A rule that is incomplete (mid-edit), unresolvable, or errors is left OUT of the answer —
the client reads a missing id as PENDING and says "Calculating…" rather than showing a count
it cannot stand behind. Never resolve an incomplete rule: `to_num(None)` is 0 by the
engines' shared contract, so a valueless `Sales > …` would become `Sales > 0` — a wrong set
under a confident count.
"""
from harness import measure_filter as _mf
by_id = {}
for cfg in list(view_configs or []) + [pending_cfg or {}]:
for rule in _mf.collect((cfg or {}).get('filters')):
rid = str(rule.get('id') or '')
if rid:
by_id[rid] = rule
rules = list(by_id.values())
if not rules:
return {}
pk = _pool_key(allowed_pids)
out = {}
for rule in rules:
if not _mf.rule_complete(rule):
continue
key = (stamp, team_id, pk, _mf.signature(rule))
if key not in memo:
try:
memo[key] = _mf.resolve_rule(rule, today, team_id, allowed_pids)
except Exception as e:
# An unresolvable condition must NOT become "everything" or "nothing" silently.
if on_error:
on_error('measure-resolve', e)
continue
out[str(rule['id'])] = sorted(memo[key])
if len(memo) > 200: # a lifetime cache, not a leak
memo.clear()
return out
def series_values(fields, bucket, buckets, team_id, allowed_pids, today, stamp,
memo, on_error=None):
"""`{field key: {'values': {bucket_start: v}, 'agg': kind}}` for every measure column
among `fields`, over `buckets` = [(start_iso, end_iso), …] — the C-TS channel, C-TSWIN
semantics since wave 14: each field's OWN window slides across the buckets
(`today := bucket end`), so two fields on the same measure with different windows are
DIFFERENT questions — the memo key carries the window signature, which the wave-13 key
(measure-only) silently conflated. `custom` fixed-range windows are the CALLER's to drop
(`window_fixed`) before calling; one that slips through memoises as a permanent failure.
Same memo discipline as `column_values`: keys carry (stamp, scope, POOL IDENTITY, the
question); a permanent failure memoises as None (the caller reports that field as
dropped), a TRANSIENT one (datastore still warming) never memoises, so the next call
retries instead of freezing a blank panel for the session's lifetime.
"""
out = {}
mfields = [f for f in fields or [] if isinstance(f.get('measure'), dict)]
if not mfields or not buckets:
return out
from harness import measure_filter as _mf
from harness import windows as _wn
pk = _pool_key(allowed_pids)
span_from, span_to = str(buckets[0][0]), str(buckets[-1][1])
for f in mfields:
spec = f.get('measure') or {}
mkey = spec.get('key')
w = _wn.normalize(spec.get('window'))
wsig = None if w is None else tuple(sorted(w.items()))
key = (stamp, team_id, pk, mkey, bucket, span_from, span_to, wsig)
if key not in memo:
try:
memo[key] = _mf.resolve_series_slid(mkey, spec.get('window'), buckets,
today, team_id, allowed_pids)
except Exception as e:
if on_error:
on_error('measure-series', e)
transient = False
try:
from harness import datastore as _ds
transient = not _ds.ready()
except Exception:
transient = False
if transient:
continue # no memo entry -> the next call tries again
memo[key] = None
ans = memo[key]
if ans is None:
continue
out[f['key']] = ans
if len(memo) > 200:
memo.clear()
return out
def column_values(fields, team_id, allowed_pids, today, stamp, memo, on_error=None):
"""Values for every FORMULA-MEASURE column among `fields` -> `{pid: {field key: value}}`,
handed to `rows_from_pool(derived=…)` beside the cohort cells.
A column that cannot be computed (store down, BU scope on a company-level measure,
unresolvable window) degrades to BLANK — the value dict simply omits that field key, and
the client renders empty cells, never $0: blank is "could not compute", 0 is a real zero.
"""
mfields = [f for f in fields or [] if isinstance(f.get('measure'), dict)]
if not mfields:
return {}
from harness import measure_filter as _mf
from harness import windows as _wn
pk = _pool_key(allowed_pids)
out = {}
for f in mfields:
spec = f['measure']
w = _wn.normalize(spec.get('window'))
key = (stamp, team_id, pk, spec.get('key'),
None if w is None else tuple(sorted(w.items())))
if key not in memo:
try:
memo[key] = _mf.resolve_values(spec.get('key'), spec.get('window'),
today, team_id, allowed_pids)
except Exception as e:
if on_error:
on_error('measure-column', e)
# ⚠ ONLY A PERMANENT FAILURE MAY BE MEMOISED. "The data cache is still warming
# up after a restart" is TRANSIENT and self-healing — memoising it means a user
# who opened the page during that window gets blank measure cells that STAY
# blank for the whole session. A permanent failure (an unadmitted measure, an
# unresolvable window) still memoises, so the retry-storm this guard was
# written for cannot come back.
transient = False
try:
from harness import datastore as _ds
transient = not _ds.ready()
except Exception:
transient = False
if transient:
continue # no memo entry -> the next call tries again
memo[key] = None
vals = memo[key]
if vals is None:
continue
fkey = f['key']
for pid, v in vals.items():
out.setdefault(pid, {})[fkey] = v
if len(memo) > 100:
memo.clear()
return out