| """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: |
| 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: |
| |
| if on_error: |
| on_error('measure-resolve', e) |
| continue |
| out[str(rule['id'])] = sorted(memo[key]) |
| if len(memo) > 200: |
| 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 |
| 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) |
| |
| |
| |
| |
| |
| |
| transient = False |
| try: |
| from harness import datastore as _ds |
| transient = not _ds.ready() |
| except Exception: |
| transient = False |
| if transient: |
| continue |
| 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 |
|
|