dashobserve

18@dataclass
19class MonitorConfig:
20    table: str
21    freshness_column: str = None
22    max_staleness_minutes: float = None
23    min_rows: int = None
24    max_rows: int = None
25    volume_tolerance_pct: float = None
26    track_schema: bool = False

29class MonitorReport:
30    def __init__(self, results: list):
31        self.results = results
32
33    def to_dict(self) -> list:
34        return [r.to_dict() for r in self.results]
35
36    def to_pandas(self):
37        import pandas as pd
38        return pd.DataFrame(self.to_dict())
39
40    def summary(self) -> dict:
41        total = len(self.results)
42        passed = sum(1 for r in self.results if r.status == "PASS")
43        return {
44            "total_monitors": total,
45            "passed": passed,
46            "failed": total - passed,
47            "pass_rate_pct": round(passed / total * 100, 1) if total else 0,
48        }
49
50    def display(self):
51        for r in self.results:
52            print(f"[{r.status}] [{r.monitor_type}] {r.table_name} — {r.message}")

13@dataclass
14class MonitorResult:
15    table_name: str
16    monitor_type: str      # freshness | volume | schema
17    monitor_name: str
18    status: str             # PASS | FAIL | ERROR
19    message: str
20    run_timestamp: str = ""
21    details: str = "{}"     # json-encoded dict of monitor-specific details
22
23    def to_dict(self) -> dict:
24        return dict(self.__dict__)

157class ForecastReport:
158    def __init__(self, table: str, next_update: dict, volume_projections: list):
159        self.table = table
160        self.next_update = next_update
161        self.volume_projections = volume_projections
162
163    def summary(self) -> dict:
164        return {
165            "table": self.table,
166            "next_update_pattern": self.next_update["pattern"],
167            "predicted_next_update": self.next_update["predicted_next_update"],
168            "volume_periods_forecast": len(self.volume_projections),
169        }
170
171    def display(self):
172        print(f"📅 Next update prediction for {self.table}:")
173        if self.next_update["predicted_next_update"]:
174            print(f"   Pattern:         {self.next_update['pattern']}")
175            print(f"   Avg interval:    {self.next_update['avg_interval_hours']} hours")
176            print(f"   Last update:     {self.next_update['last_known_update']}")
177            print(f"   Predicted next:  {self.next_update['predicted_next_update']}")
178            print(f"   Based on:        {self.next_update['history_points']} observations")
179        else:
180            print("   Insufficient history — run monitors a few more times first")
181
182        if self.volume_projections:
183            print("\nVolume forecast:")
184            print(f"   {'Period':<14} {'Date':<14} {'Projected rows':>14}   Range")
185            print(f"   {'-'*14} {'-'*14} {'-'*14}   {'-'*24}")
186            for p in self.volume_projections:
187                rng = f"{p['lower_bound']:,} – {p['upper_bound']:,}"
188                print(f"   {p['period_label']:<14} {p['predicted_date']:<14} {p['predicted_row_count']:>14,}   {rng}")
189        else:
190            print("\nVolume forecast: insufficient history")

 55def run_monitors(config: MonitorConfig, history_table: str = None, spark=None) -> MonitorReport:
 56    """Run every monitor configured on config.table and return a MonitorReport."""
 57    if spark is None:
 58        from pyspark.sql import SparkSession
 59        spark = SparkSession.getActiveSession()
 60    from pyspark.sql import functions as F
 61
 62    df = spark.table(config.table)
 63    run_ts = datetime.now().isoformat(timespec="seconds")
 64    results = []
 65
 66    if config.freshness_column and config.max_staleness_minutes is not None:
 67        latest = df.agg(F.max(config.freshness_column)).collect()[0][0]
 68        ok, msg = check_freshness(latest, config.max_staleness_minutes)
 69        results.append(MonitorResult(
 70            config.table, "freshness", "data_freshness",
 71            "PASS" if ok else "FAIL", msg, run_ts,
 72            details=json.dumps({"latest_timestamp": latest.isoformat() if latest else None}),
 73        ))
 74
 75    if config.min_rows is not None or config.max_rows is not None or config.volume_tolerance_pct is not None:
 76        row_count = df.count()
 77        baseline_avg = None
 78        if config.volume_tolerance_pct is not None and history_table:
 79            baseline_avg = _read_volume_baseline(spark, history_table, config.table)
 80        ok, msg = check_volume(row_count, config.min_rows, config.max_rows, baseline_avg, config.volume_tolerance_pct)
 81        results.append(MonitorResult(
 82            config.table, "volume", "row_count",
 83            "PASS" if ok else "FAIL", msg, run_ts,
 84            details=json.dumps({"row_count": row_count}),
 85        ))
 86
 87    if config.track_schema:
 88        new_schema = {f.name: str(f.dataType) for f in df.schema.fields}
 89        old_schema = _read_last_schema(spark, history_table, config.table) if history_table else None
 90        ok, msg, diff = check_schema(old_schema, new_schema)
 91        results.append(MonitorResult(
 92            config.table, "schema", "schema_change",
 93            "PASS" if ok else "FAIL", msg, run_ts,
 94            details=json.dumps({"schema": new_schema, "diff": diff}),
 95        ))
 96
 97    if history_table and results:
 98        _save_results(spark, history_table, results)
 99
100    return MonitorReport(results)

193def run_forecast(
194    table: str,
195    history_table: str,
196    n_periods: int = 4,
197    period: str = "weeks",
198    spark=None,
199) -> ForecastReport:
200    """
201    Read historical monitor results and produce next-update and volume forecasts.
202
203    Requires that run_monitors() has been called at least a few times with the
204    same history_table so there is enough data to fit a trend.
205    """
206    if spark is None:
207        from pyspark.sql import SparkSession
208        spark = SparkSession.getActiveSession()
209    from pyspark.sql import functions as F
210    from datetime import datetime
211
212    update_timestamps = []
213    try:
214        rows = (
215            spark.table(history_table)
216                 .filter((F.col("table_name") == table) & (F.col("monitor_type") == "freshness"))
217                 .orderBy("run_timestamp")
218                 .select("details")
219                 .collect()
220        )
221        for r in rows:
222            try:
223                ts_str = json.loads(r["details"]).get("latest_timestamp")
224                if ts_str:
225                    update_timestamps.append(datetime.fromisoformat(ts_str))
226            except Exception:
227                continue
228    except Exception:
229        pass
230
231    volume_history = []
232    try:
233        rows = (
234            spark.table(history_table)
235                 .filter((F.col("table_name") == table) & (F.col("monitor_type") == "volume"))
236                 .orderBy("run_timestamp")
237                 .select("run_timestamp", "details")
238                 .collect()
239        )
240        for r in rows:
241            try:
242                count = json.loads(r["details"]).get("row_count")
243                if count is not None:
244                    volume_history.append((r["run_timestamp"], count))
245            except Exception:
246                continue
247    except Exception:
248        pass
249
250    return ForecastReport(
251        table=table,
252        next_update=predict_next_update(update_timestamps),
253        volume_projections=predict_volume(volume_history, n_periods=n_periods, period=period),
254    )

27def check_freshness(latest_timestamp, max_staleness_minutes: float, now: datetime = None) -> tuple[bool, str]:
28    """Most recent value in the freshness column must be within max_staleness_minutes of now."""
29    if latest_timestamp is None:
30        return False, "No timestamp values found — table may be empty"
31    now = now or datetime.now(timezone.utc).replace(tzinfo=None)
32    if getattr(latest_timestamp, "tzinfo", None) is not None:
33        latest_timestamp = latest_timestamp.replace(tzinfo=None)
34    staleness_minutes = (now - latest_timestamp).total_seconds() / 60
35    ok = staleness_minutes <= max_staleness_minutes
36    msg = f"Data is {staleness_minutes:.1f} min old (threshold {max_staleness_minutes} min)"
37    return ok, msg

40def check_volume(row_count: int, min_rows: int = None, max_rows: int = None,
41                  baseline_avg: float = None, tolerance_pct: float = None) -> tuple[bool, str]:
42    """Row count must satisfy absolute bounds and/or stay within tolerance_pct of a historical baseline."""
43    if min_rows is not None and row_count < min_rows:
44        return False, f"Row count {row_count} below min_rows {min_rows}"
45    if max_rows is not None and row_count > max_rows:
46        return False, f"Row count {row_count} above max_rows {max_rows}"
47    if baseline_avg is not None and tolerance_pct is not None and baseline_avg > 0:
48        deviation_pct = abs(row_count - baseline_avg) / baseline_avg * 100
49        if deviation_pct > tolerance_pct:
50            return False, (
51                f"Row count {row_count} deviates {deviation_pct:.1f}% from baseline "
52                f"{baseline_avg:.0f} (tolerance {tolerance_pct}%)"
53            )
54    return True, f"Row count {row_count} within expected bounds"

174def check_schema(old_schema: dict, new_schema: dict) -> tuple[bool, str, dict]:
175    """old_schema=None means no prior snapshot exists yet — treated as a baseline, not a failure."""
176    if old_schema is None:
177        return True, "Initial schema snapshot recorded", {"added": [], "removed": [], "type_changed": []}
178
179    diff = diff_schema(old_schema, new_schema)
180    changed = diff["added"] or diff["removed"] or diff["type_changed"]
181    if not changed:
182        return True, "No schema changes detected", diff
183
184    parts = []
185    if diff["added"]:
186        parts.append(f"added: {', '.join(diff['added'])}")
187    if diff["removed"]:
188        parts.append(f"removed: {', '.join(diff['removed'])}")
189    if diff["type_changed"]:
190        parts.append(f"type changed: {', '.join(diff['type_changed'])}")
191    return False, "Schema changed — " + "; ".join(parts), diff

57def diff_schema(old_schema: dict, new_schema: dict) -> dict:
58    """Compare two {column_name: dtype_string} schemas. Pure set/dict logic, no I/O."""
59    old_cols, new_cols = set(old_schema), set(new_schema)
60    added = sorted(new_cols - old_cols)
61    removed = sorted(old_cols - new_cols)
62    type_changed = sorted(c for c in (old_cols & new_cols) if old_schema[c] != new_schema[c])
63    return {"added": added, "removed": removed, "type_changed": type_changed}

 66def predict_next_update(update_timestamps: list) -> dict:
 67    """
 68    Predict when a table will next be updated based on historical update timestamps.
 69
 70    update_timestamps: list of datetime objects (when the freshness column last changed).
 71    Returns dict: predicted_next_update (ISO str), avg_interval_hours, pattern, history_points.
 72    """
 73    from datetime import timedelta
 74
 75    if len(update_timestamps) < 2:
 76        return {
 77            "predicted_next_update": None,
 78            "avg_interval_hours": None,
 79            "pattern": "insufficient_data",
 80            "last_known_update": update_timestamps[0].isoformat(timespec="seconds") if update_timestamps else None,
 81            "history_points": len(update_timestamps),
 82        }
 83
 84    sorted_ts = sorted(update_timestamps)
 85    intervals_h = [
 86        (sorted_ts[i + 1] - sorted_ts[i]).total_seconds() / 3600
 87        for i in range(len(sorted_ts) - 1)
 88    ]
 89    avg_h = sum(intervals_h) / len(intervals_h)
 90
 91    pattern = "irregular"
 92    for threshold_h, label in [
 93        (1.4,  "hourly"),
 94        (5.0,  "every few hours"),
 95        (18.0, "twice daily"),
 96        (30.0, "daily"),
 97        (55.0, "every few days"),
 98        (200.0, "weekly"),
 99        (400.0, "bi-weekly"),
100        (800.0, "monthly"),
101    ]:
102        if avg_h <= threshold_h:
103            pattern = label
104            break
105
106    predicted_next = sorted_ts[-1] + timedelta(hours=avg_h)
107    return {
108        "predicted_next_update": predicted_next.isoformat(timespec="seconds"),
109        "avg_interval_hours": round(avg_h, 2),
110        "pattern": pattern,
111        "last_known_update": sorted_ts[-1].isoformat(timespec="seconds"),
112        "history_points": len(sorted_ts),
113    }

116def predict_volume(history: list, n_periods: int = 4, period: str = "weeks") -> list[dict]:
117    """
118    Forecast future row counts using a linear trend fitted to historical observations.
119
120    history: list of (datetime_or_isostr, row_count) pairs.
121    period: 'days' | 'weeks' | 'months'
122    Returns list of dicts: period_label, predicted_date, predicted_row_count,
123    lower_bound, upper_bound (95% prediction interval).
124    """
125    from datetime import datetime, timedelta
126
127    PERIOD_DAYS = {"days": 1, "weeks": 7, "months": 30}
128    if period not in PERIOD_DAYS:
129        raise ValueError(f"period must be one of {list(PERIOD_DAYS)}, got {period!r}")
130    if len(history) < 2:
131        return []
132
133    parsed = []
134    for ts, count in history:
135        if isinstance(ts, str):
136            ts = datetime.fromisoformat(ts)
137        parsed.append((ts, int(count)))
138    parsed.sort(key=lambda x: x[0])
139
140    t0 = parsed[0][0]
141    xs = [(p[0] - t0).total_seconds() / 86400 for p in parsed]
142    ys = [p[1] for p in parsed]
143    n = len(xs)
144    mean_x = sum(xs) / n
145    mean_y = sum(ys) / n
146    denom = sum((x - mean_x) ** 2 for x in xs)
147    slope = sum((xs[i] - mean_x) * (ys[i] - mean_y) for i in range(n)) / denom if denom else 0
148    intercept = mean_y - slope * mean_x
149
150    residuals = [ys[i] - (intercept + slope * xs[i]) for i in range(n)]
151    std_err = (sum(r ** 2 for r in residuals) / max(n - 2, 1)) ** 0.5
152
153    step_days = PERIOD_DAYS[period]
154    last_x = xs[-1]
155    last_ts = parsed[-1][0]
156
157    projections = []
158    for i in range(1, n_periods + 1):
159        future_x = last_x + i * step_days
160        future_ts = last_ts + timedelta(days=i * step_days)
161        predicted = max(0, int(intercept + slope * future_x))
162        ci = int(1.96 * std_err)
163        projections.append({
164            "period": i,
165            "period_label": f"+{i} {period}",
166            "predicted_date": future_ts.strftime("%Y-%m-%d"),
167            "predicted_row_count": predicted,
168            "lower_bound": max(0, predicted - ci),
169            "upper_bound": predicted + ci,
170        })
171    return projections

 8def env_setup() -> None:
 9    """Open the environment setup panel — where should dashobserve read/write
10    its configs? Defaults to the notebook's current working directory if
11    never called."""
12    try:
13        import dashui
14        from IPython.display import display
15    except ImportError:
16        raise RuntimeError("ipywidgets required. Run: %pip install ipywidgets") from None
17
18    display(dashui.card([
19        dashui.header("DashObserve — Environment Setup", library=_LIBRARY),
20        dashui.env_setup_panel(_LIBRARY).widget,
21    ]))

 24def launch():
 25    try:
 26        import ipywidgets as w
 27        from IPython.display import display
 28    except ImportError:
 29        raise RuntimeError("ipywidgets required. Run: %pip install ipywidgets")
 30
 31    import dashui
 32
 33    saved = dashui.load_config(_LIBRARY, defaults={"monitors": [], "history_table": ""})
 34    monitors: list[dict] = list(saved["monitors"])
 35
 36    # ── Add monitor ───────────────────────────────────────────────────────
 37    m_table = w.Text(description="UC Table:", placeholder="catalog.schema.table")
 38
 39    m_freshness_col = w.Text(description="Freshness col:", placeholder="updated_at (optional)")
 40    m_max_staleness = w.IntText(description="Max staleness (min):", value=1440)
 41
 42    m_min_rows = w.Text(description="Min rows:", placeholder="optional")
 43    m_max_rows = w.Text(description="Max rows:", placeholder="optional")
 44    m_tolerance = w.Text(description="Baseline tolerance %:", placeholder="optional, e.g. 20")
 45
 46    m_track_schema = w.Checkbox(value=True, description="Track schema changes")
 47
 48    add_btn = dashui.action_button("Add Monitor", style="info")
 49    monitors_output, render_monitors = dashui.running_list(
 50        lambda i, m: (
 51            f"  {i}. {m['table']} — "
 52            f"freshness:{'on' if m['freshness_column'] else 'off'}, "
 53            f"volume:{'on' if (m['min_rows'] or m['max_rows'] or m['volume_tolerance_pct']) else 'off'}, "
 54            f"schema:{'on' if m['track_schema'] else 'off'}"
 55        )
 56    )
 57
 58    def _parse_int(text):
 59        text = text.strip()
 60        return int(text) if text else None
 61
 62    def _parse_float(text):
 63        text = text.strip()
 64        return float(text) if text else None
 65
 66    def _save_state() -> None:
 67        try:
 68            dashui.save_config(_LIBRARY, {"monitors": monitors, "history_table": history_table.value.strip()})
 69        except Exception:
 70            pass  # persistence is a convenience, never block the actual operation on it
 71
 72    def on_add(b):
 73        table = m_table.value.strip()
 74        if not table:
 75            return
 76        monitors.append({
 77            "table": table,
 78            "freshness_column": m_freshness_col.value.strip() or None,
 79            "max_staleness_minutes": m_max_staleness.value,
 80            "min_rows": _parse_int(m_min_rows.value),
 81            "max_rows": _parse_int(m_max_rows.value),
 82            "volume_tolerance_pct": _parse_float(m_tolerance.value),
 83            "track_schema": m_track_schema.value,
 84        })
 85        render_monitors(monitors)
 86        m_table.value = m_freshness_col.value = m_min_rows.value = m_max_rows.value = m_tolerance.value = ""
 87        _save_state()
 88
 89    add_btn.on_click(on_add)
 90    render_monitors(monitors)  # show any monitors restored from a previous session
 91
 92    # ── Run ──────────────────────────────────────────────────────────────
 93    history_table = w.Text(description="History table:", placeholder="catalog.schema.observe_history (optional)", value=saved["history_table"])
 94    run_btn = dashui.action_button("Run All Monitors", style="success")
 95    output = dashui.output_panel()
 96
 97    def on_run(b):
 98        with output:
 99            output.clear_output()
100            if not monitors:
101                print("No monitors configured — add at least one above")
102                return
103            _save_state()
104            try:
105                from dashobserve.runner import MonitorConfig, run_monitors
106                hist = history_table.value.strip() or None
107                for m in monitors:
108                    cfg = MonitorConfig(
109                        table=m["table"],
110                        freshness_column=m["freshness_column"],
111                        max_staleness_minutes=m["max_staleness_minutes"] if m["freshness_column"] else None,
112                        min_rows=m["min_rows"],
113                        max_rows=m["max_rows"],
114                        volume_tolerance_pct=m["volume_tolerance_pct"],
115                        track_schema=m["track_schema"],
116                    )
117                    report = run_monitors(cfg, history_table=hist)
118                    report.display()
119                    s = report.summary()
120                    print(f"   → {s['passed']}/{s['total_monitors']} passed\n")
121            except Exception as e:
122                print(f"Error: {e}")
123
124    run_btn.on_click(on_run)
125
126    # ── Forecast ─────────────────────────────────────────────────────────
127    f_table = w.Text(description="UC Table:", placeholder="catalog.schema.table")
128    f_history = w.Text(description="History table:", placeholder="catalog.schema.observe_history")
129    f_n_periods = w.IntText(description="Periods ahead:", value=4, min=1, max=52)
130    f_period = w.ToggleButtons(options=["days", "weeks", "months"], description="Period:")
131    forecast_btn = dashui.action_button("Run Forecast", style="info")
132    forecast_output = dashui.output_panel()
133
134    def on_forecast(b):
135        with forecast_output:
136            forecast_output.clear_output()
137            table = f_table.value.strip()
138            hist = f_history.value.strip()
139            if not table or not hist:
140                print("Specify both a UC table and the history table")
141                return
142            try:
143                from dashobserve.runner import run_forecast
144                report = run_forecast(
145                    table=table,
146                    history_table=hist,
147                    n_periods=f_n_periods.value,
148                    period=f_period.value,
149                )
150                report.display()
151            except Exception as e:
152                print(f"Error: {e}")
153
154    forecast_btn.on_click(on_forecast)
155
156    env_accordion = w.Accordion(children=[dashui.env_setup_panel(_LIBRARY).widget])
157    env_accordion.set_title(0, "Environment setup")
158    env_accordion.selected_index = None
159
160    ui = dashui.card([
161        dashui.header("DashObserve — Data Observability", library="dashobserve"),
162        env_accordion,
163        dashui.section("Step 1: Configure a monitor"),
164        m_table,
165        w.HBox([m_freshness_col, m_max_staleness]),
166        w.HBox([m_min_rows, m_max_rows, m_tolerance]),
167        m_track_schema,
168        add_btn, monitors_output,
169        dashui.section("Step 2: Run monitors"),
170        history_table,
171        run_btn,
172        output,
173        dashui.section("Step 3: Forecast"),
174        dashui.html(
175            "<div style='font-size:12px;color:#666;margin-bottom:4px'>"
176            "Predicts next table update and projects volume trend. "
177            "Requires history built from prior monitor runs.</div>"
178        ),
179        w.HBox([f_table, f_history]),
180        w.HBox([f_n_periods, f_period]),
181        forecast_btn,
182        forecast_output,
183    ])
184    display(ui)