Rifqi Hafizuddin
Claude Opus 5.5
[NOTICKET] feat(query): PA over several periods from summed hours
307bac0 | """Measures computed from summed inputs (2026-11-09, docs/plans/RATIO_MEASURES_REQUIREMENTS.md). | |
| Over several periods or units the plant measures PA as ONE ratio of summed hours, | |
| `100 * (Σ total hours - Σ breakdown) / Σ total hours` (STD_2026_006_MNO p. 4, EMA's | |
| `PAPerformanceMonthly.yml`), not as the average of the per-month `pa` values. The | |
| query IR can compute that (`AggSelect.computed`); this module decides WHEN, from data: | |
| 1. `bind_computed_measures` (read time, M3): an approved measure whose formula entry | |
| carries a `compute` block, e.g. | |
| {"column": "pa", | |
| "numerator": [["+", "total_inpr_hour"], ["-", "breakdown_duration_total"]], | |
| "denominator": [["+", "total_inpr_hour"]], | |
| "scale": 100, "label": {"id": "PA dari total jam", "en": "PA from summed hours"}} | |
| marks every catalog table holding all those columns (by NAME, numeric terms only) | |
| with `Column.measure_method = {"kind": "ratio", ...}` in column ids. An approved | |
| measure WITH a formula but no usable block marks a same-named column | |
| `{"kind": "plain"}`, so the answer can say it is a plain average (U9). | |
| 2. `mark_computed` (plan validation, M4): every `avg` of a ratio-marked column gets the | |
| `computed` spec. The planner keeps writing `avg(pa)`; it is never taught the form. | |
| 3. `wants_plain_average` (M8): the user explicitly asked for a plain average, so the | |
| rewrite is skipped for the turn (U10). | |
| The definition lives once, on the approved formula entry (one home, §5.22); no prompt | |
| text. Every function here is fail-open: a bad block or a slip leaves the plain `avg`, | |
| which is today's behaviour. | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from typing import Any | |
| from src.middlewares.logging import get_logger | |
| logger = get_logger("knowledge_computed") | |
| _NUMERIC = frozenset({"int", "decimal"}) | |
| _SIGNS = frozenset({"+", "-"}) | |
| _SCALES = frozenset({1, 100}) | |
| _MAX_TERMS = 4 | |
| def _terms(raw: Any) -> list[tuple[str, str]] | None: | |
| """[[sign, column], ...] → [(sign, column_name)], or None when malformed.""" | |
| if not isinstance(raw, list) or not 1 <= len(raw) <= _MAX_TERMS: | |
| return None | |
| out: list[tuple[str, str]] = [] | |
| for item in raw: | |
| if not (isinstance(item, list | tuple) and len(item) == 2): | |
| return None | |
| sign, name = item | |
| if sign not in _SIGNS or not isinstance(name, str) or not name.strip(): | |
| return None | |
| out.append((sign, name.strip())) | |
| return out | |
| def _parse(spec: Any) -> dict | None: | |
| if not isinstance(spec, dict): | |
| return None | |
| column = spec.get("column") | |
| numerator, denominator = _terms(spec.get("numerator")), _terms(spec.get("denominator")) | |
| try: | |
| scale = int(spec.get("scale", 1)) | |
| except (TypeError, ValueError): | |
| return None | |
| label = spec.get("label") if isinstance(spec.get("label"), dict) else {} | |
| if not isinstance(column, str) or not numerator or not denominator or scale not in _SCALES: | |
| return None | |
| return {"column": column.strip(), "numerator": numerator, "denominator": denominator, | |
| "scale": scale, "label": {k: str(v) for k, v in label.items() if v}} | |
| def bind_computed_measures(measures: list[Any], catalog: Any) -> int: | |
| """Mark catalog columns with how their measure is computed. Returns tables bound. | |
| In place on the catalog's Column objects (catalogs are rebuilt per read, so this | |
| never leaks across requests). Idempotent. Never raises. | |
| """ | |
| try: | |
| return _bind(measures, catalog) | |
| except Exception as e: # an enrichment must never cost the answer | |
| logger.warning( | |
| "computed_measures_bind_failed", | |
| extra={"error": repr(e), "degraded_seam": "computed_measures"}, | |
| ) | |
| return 0 | |
| def _bind(measures: list[Any], catalog: Any) -> int: | |
| ratios: list[tuple[str, dict]] = [] | |
| plain_names: dict[str, str] = {} | |
| for m in measures or []: | |
| if not getattr(m, "approved", False): | |
| continue | |
| spec = _parse(getattr(m, "compute", None)) | |
| if spec is not None: | |
| ratios.append((m.name, spec)) | |
| elif getattr(m, "compute", None) is not None: | |
| logger.warning("computed_measure_spec_invalid", extra={"measure": m.name}) | |
| if getattr(m, "formula_latex", None): | |
| for form in [m.name, *getattr(m, "surface_forms", [])]: | |
| plain_names.setdefault(form.strip().casefold(), m.name) | |
| bound = 0 | |
| for source in getattr(catalog, "sources", []) or []: | |
| for table in getattr(source, "tables", []) or []: | |
| by_name = {c.name.strip().casefold(): c for c in table.columns} | |
| ratio_cols: set[str] = set() | |
| for measure, spec in ratios: | |
| target = by_name.get(spec["column"].casefold()) | |
| terms = [*spec["numerator"], *spec["denominator"]] | |
| cols = [by_name.get(name.casefold()) for _, name in terms] | |
| if target is None or any(c is None or c.data_type not in _NUMERIC for c in cols): | |
| continue | |
| ids = iter(c.column_id for c in cols) | |
| num = [{"column_id": next(ids), "sign": s} for s, _ in spec["numerator"]] | |
| den = [{"column_id": next(ids), "sign": s} for s, _ in spec["denominator"]] | |
| target.measure_method = { | |
| "kind": "ratio", | |
| "measure": measure, | |
| "numerator": num, | |
| "denominator": den, | |
| "scale": spec["scale"], | |
| "label": spec["label"], | |
| } | |
| ratio_cols.add(target.column_id) | |
| bound += 1 | |
| for name, column in by_name.items(): | |
| measure = plain_names.get(name) | |
| if measure and column.column_id not in ratio_cols: | |
| column.measure_method = {"kind": "plain", "measure": measure} | |
| logger.info("computed_measures_bound", extra={"n_bound": bound, "n_defined": len(ratios)}) | |
| return bound | |
| def mark_computed(ir: Any, catalog: Any) -> list[str]: | |
| """Give every `avg` of a ratio-marked column its `computed` spec. In place. | |
| Returns the aliases (or column ids) it marked. Leaves sum/min/max/count, raw | |
| columns and an already-computed select alone. | |
| """ | |
| from src.query.ir.models import AggSelect, ComputedRatio | |
| columns = { | |
| c.column_id: c | |
| for s in getattr(catalog, "sources", []) or [] | |
| if s.source_id == ir.source_id | |
| for t in s.tables | |
| for c in t.columns | |
| } | |
| marked: list[str] = [] | |
| for item in ir.select: | |
| if not isinstance(item, AggSelect) or item.fn != "avg" or item.computed is not None: | |
| continue | |
| method = getattr(columns.get(item.column_id), "measure_method", None) or {} | |
| if method.get("kind") != "ratio": | |
| continue | |
| item.computed = ComputedRatio( | |
| numerator=method["numerator"], | |
| denominator=method["denominator"], | |
| scale=method["scale"], | |
| label=method.get("label") or {}, | |
| ) | |
| marked.append(item.alias or item.column_id) | |
| return marked | |
| # Explicit requests for the plain average (U10). Matched on the user's words, EN + ID. | |
| _PLAIN_AVERAGE = re.compile( | |
| r"rata[\s-]?rata\s+(sederhana|biasa|aritmatika|aritmetika)" | |
| r"|bukan\s+(dari\s+)?(total|jumlah)\s+jam" | |
| r"|tanpa\s+bobot|tidak\s+(di)?bobot" | |
| r"|simple\s+(average|mean)|plain\s+(average|mean)|arithmetic\s+(average|mean)" | |
| r"|unweighted|not\s+weighted|not\s+from\s+(summed|total)\s+hours", | |
| re.IGNORECASE, | |
| ) | |
| def wants_plain_average(*texts: str | None) -> bool: | |
| """True when the user's own words ask for a plain (unweighted) average.""" | |
| return any(t and _PLAIN_AVERAGE.search(t) for t in texts) | |