File size: 9,087 Bytes
c14ceee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | """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
|