diff --git a/oonipipeline/pyproject.toml b/oonipipeline/pyproject.toml index 4feef080..8fde5e99 100644 --- a/oonipipeline/pyproject.toml +++ b/oonipipeline/pyproject.toml @@ -42,6 +42,7 @@ analysis = [ "bokeh ~= 3.5.2", "streamlit", "pyarrow < 19", # fixes segfault with streamlit + "vl-convert-python>=1.9.0.post1", ] [tool.hatch.build.targets.sdist] @@ -81,6 +82,11 @@ test-cov = "pytest --cov=./ --cov-report=xml --cov-report=html --cov-report=term cov-report = ["coverage report"] cov = ["test-cov", "cov-report"] +[tool.coverage.run] +omit = [ + "src/oonipipeline/events_panel/panel.py", +] + [tool.hatch.envs.analysis] template = "default" features = ["analysis"] diff --git a/oonipipeline/src/oonipipeline/analysis/detector.py b/oonipipeline/src/oonipipeline/analysis/detector.py index 52746761..81389905 100644 --- a/oonipipeline/src/oonipipeline/analysis/detector.py +++ b/oonipipeline/src/oonipipeline/analysis/detector.py @@ -511,6 +511,7 @@ def run_detector( trace: bool = False, slack_webhook: str | None = None, explorer_base_url: str = "https://explorer.ooni.org/", + detector_panel_base_url: str = "https://detector-panel.prod.ooni.io/", ) -> Tuple[List[Changepoint], List[LastCusum], List[CusumStep]]: db = ClickhouseClient.from_url(clickhouse_url) domains = get_domain_list(db) @@ -536,7 +537,15 @@ def run_detector( update_tables(db, updated_cusums, changepoints) if slack_webhook is not None: - notify_slack(changepoints, slack_webhook, explorer_base_url) + notify_slack( + changepoints, + slack_webhook, + explorer_base_url, + detector_panel_base_url, + edd=edd, + gap_halflife=gap_halflife, + warmup=warmup, + ) return changepoints, updated_cusums, steps @@ -580,11 +589,44 @@ def run_detector_for( def plot(steps: List[CusumStep], block_type: str): make_cusums_chart(steps, block_type).show() -def make_cusums_chart(steps: List[CusumStep], block_type: str): +def make_cusums_chart( + steps: List[CusumStep], + block_type: str, + zoom=None, + nearest=None, + show_legend: bool = True, + show_x_axis: bool = True, +): + """ + zoom: an optional shared alt.selection_interval(bind="scales") to reuse + across multiple calls (see make_cusums_chart_grid) so pan/zoom on one + chart's time axis applies to all of them. When omitted, a private one is + created so this function still works as a standalone chart. + + nearest: an optional shared alt.selection_single(nearest=True, ...) to + reuse across multiple calls (see make_cusums_chart_grid), so a single + mouseover listener drives the hover crosshair/highlights on every row. + When omitted, a private one is created + + show_legend: when False, suppresses this chart's own State/Value/CUSUM + legends. Used when stacking several of these charts so only one copy + of each legend is shown for the whole stack. + + show_x_axis: when False, hides the time-axis tick labels/ticks (grid + lines stay). Used when stacking several of these charts so only the + bottom-most one shows the shared time axis. + """ import altair as alt import pandas as pd STATE_COLORS = {"ok": "#d4edda", "blk": "#f8d7da", "unk": "#fff3cd"} + x_axis = alt.Axis( + format="%b %d, %H:%M", + title=None, + labels=show_x_axis, + ticks=show_x_axis, + ) + ts_tooltip_fmt = "%b %d, %H:%M" df_steps = pd.DataFrame([s for s in steps if s["block_type"] == block_type]) df_last = df_steps.loc[[df_steps["ts"].idxmax()]] @@ -608,7 +650,7 @@ def make_cusums_chart(steps: List[CusumStep], block_type: str): alt.Chart(df_bands) .mark_rect(opacity=0.25) .encode( - x=alt.X("state_start:T"), + x=alt.X("state_start:T", axis=x_axis), x2=alt.X2("state_end:T"), color=alt.Color( "current_state:N", @@ -620,18 +662,30 @@ def make_cusums_chart(steps: List[CusumStep], block_type: str): STATE_COLORS["unk"], ], ), - legend=alt.Legend(title="State"), + legend=alt.Legend(title="State") if show_legend else None, ), - tooltip=["current_state:N", "state_start:T", "state_end:T"], + tooltip=[ + "current_state:N", + alt.Tooltip("state_start:T", format=ts_tooltip_fmt), + alt.Tooltip("state_end:T", format=ts_tooltip_fmt), + ], ) ) - base = alt.Chart(df_steps).encode(x="ts:T") + base = alt.Chart(df_steps).encode(x=alt.X("ts:T", axis=x_axis)) + + # One tooltip covering all three series at once (rather than trying to + # guess which line the cursor is closer to), using a plain x-only + # "nearest" selection — cheap, unlike a 2D Voronoi nearest selection. + if nearest is None: + nearest = alt.selection_single( + nearest=True, on="mouseover", fields=["ts"], empty="none", name="nearest" + ) def make_label(df, field, label, color): return ( alt.Chart(df) - .mark_text(align="left", dx=5, fontSize=11, color=color) + .mark_text(align="right", dx=-5, fontSize=11, color=color) .encode( x="ts:T", y=alt.Y(f"{field}:Q"), @@ -639,20 +693,34 @@ def make_label(df, field, label, color): ) ) + def make_highlight(field, color): + # transform_filter (rather than an opacity condition over the full + # dataset) means only the single matched point is ever rendered, so + # this stays O(1) instead of O(n) marks per row. + return ( + base.transform_filter(nearest) + .mark_point(color=color, filled=True, size=80) + .encode(y=alt.Y(f"{field}:Q")) + ) + + def make_legend_swatch(labels, colors, title): + """A zero-opacity mark whose sole purpose is to render a fixed-domain + color legend, since the actual lines it labels use static (not + data-driven) mark colors and so don't produce one on their own.""" + return ( + alt.Chart(pd.DataFrame({"label": labels})) + .mark_point(filled=True, opacity=0) + .encode( + color=alt.Color( + "label:N", + scale=alt.Scale(domain=labels, range=colors), + legend=alt.Legend(title=title), + ) + ) + ) + obs_line = base.mark_line(color="steelblue", opacity=0.5).encode( y=alt.Y("obs_value:Q", title="value"), - tooltip=["ts:T", "obs_value:Q", "weight:Q", "current_state:N"], - ) - s_pos_line = base.mark_line(color="red").encode( - y=alt.Y("s_pos:Q"), - tooltip=["ts:T", "s_pos:Q"], - ) - s_neg_line = base.mark_line(color="orange").encode( - y=alt.Y("s_neg:Q"), - tooltip=["ts:T", "s_neg:Q"], - ) - threshold = ( - alt.Chart(df_steps).mark_rule(color="green", strokeDash=[4, 4]).encode(y="h:Q") ) cp_points = ( alt.Chart(df_steps[df_steps["is_changepoint"]]) @@ -660,11 +728,56 @@ def make_label(df, field, label, color): .encode( x="ts:T", y=alt.Y("obs_value:Q"), - tooltip=["ts:T", "obs_value:Q", "s_pos:Q", "s_neg:Q", "current_state:N"], + tooltip=[ + alt.Tooltip("ts:T", format=ts_tooltip_fmt), + "obs_value:Q", + "s_pos:Q", + "s_neg:Q", + "current_state:N", + ], + ) + ) + obs_last_value = df_last["obs_value"].iloc[0] + obs_label_text = ( + f"observed: {obs_last_value:.2f}" if pd.notna(obs_last_value) else "observed: n/a" + ) + obs_label = make_label(df_last, "obs_value", obs_label_text, "steelblue") + obs_legend = make_legend_swatch(["observed"], ["steelblue"], "Value") + + s_pos_line = base.mark_line(color="red").encode( + y=alt.Y("s_pos:Q", title="CUSUM statistic") + ) + s_neg_line = base.mark_line(color="orange").encode(y=alt.Y("s_neg:Q")) + + selectors = ( + base.mark_point() + .encode( + opacity=alt.value(0), + tooltip=[ + alt.Tooltip("ts:T", format=ts_tooltip_fmt), + alt.Tooltip("obs_value:Q", title="šŸ”µ observed"), + alt.Tooltip("s_pos:Q", title="šŸ”“ S+"), + alt.Tooltip("s_neg:Q", title="🟠 Sāˆ’"), + ], ) + .add_selection(nearest) ) + hover_rule = ( + base.transform_filter(nearest).mark_rule(color="gray", opacity=0.3) + ) + obs_highlight = make_highlight("obs_value", "steelblue") + s_pos_highlight = make_highlight("s_pos", "red") + s_neg_highlight = make_highlight("s_neg", "orange") + hover = hover_rule + selectors + + value_group_layers = [obs_line, obs_highlight, cp_points, obs_label] + if show_legend: + value_group_layers.append(obs_legend) + value_group = alt.layer(*value_group_layers) - obs_label = make_label(df_last, "obs_value", "observed", "steelblue") + threshold = ( + alt.Chart(df_steps).mark_rule(color="green", strokeDash=[4, 4]).encode(y="h:Q") + ) s_pos_label = make_label(df_last, "s_pos", "S+", "red") s_neg_label = make_label(df_last, "s_neg", "Sāˆ’", "orange") df_h = df_last[["ts", "h"]].copy() @@ -673,33 +786,97 @@ def make_label(df, field, label, color): .mark_text(align="left", dx=5, fontSize=11, color="green") .encode(x="ts:T", y="h:Q", text=alt.value("threshold (h)")) ) + cusum_group_layers = [ + s_pos_line, + s_neg_line, + s_pos_highlight, + s_neg_highlight, + threshold, + s_pos_label, + s_neg_label, + threshold_label, + ] + if show_legend: + cusum_group_layers.append( + make_legend_swatch( + ["S+", "Sāˆ’", "threshold (h)"], ["red", "orange", "green"], "CUSUM" + ) + ) + cusum_group = alt.layer(*cusum_group_layers) + + if zoom is None: + zoom = alt.selection_interval(bind="scales", encodings=["x"]) chart = ( - ( - state_bands - + obs_line - + obs_label - + s_pos_line - + s_pos_label - + s_neg_line - + s_neg_label - + threshold - + threshold_label - + cp_points - ) + alt.layer(state_bands, value_group, cusum_group, hover) + .resolve_scale(y="independent", color="independent") + .resolve_legend(color="independent") .properties( width=900, - height=400, - title=f"CUSUM Detector: {block_type}", + height=100, + title=alt.TitleParams( + text=block_type, + fontSize=11, + fontWeight="normal", + color="gray", + anchor="start", + dy=-4, + offset=2, + ), ) - .interactive() + .add_selection(zoom) ) return chart +def make_cusums_chart_grid(steps: List[CusumStep], block_types: List[str]): + """ + One row per block type, stacked vertically in a single chart, so all of + an ASN's CUSUM series can be compared at a glance. The State/Value/CUSUM + legends (identical across rows) are only shown once, on the last row, + and every row shares one time-axis pan/zoom selection. + """ + import altair as alt + + present_block_types = [ + block_type + for block_type in block_types + if any(s["block_type"] == block_type for s in steps) + ] + if not present_block_types: + return None + + zoom = alt.selection_interval(bind="scales", encodings=["x"]) + nearest = alt.selection_single( + nearest=True, on="mouseover", fields=["ts"], empty="none", name="nearest" + ) + last_idx = len(present_block_types) - 1 + charts = [ + make_cusums_chart( + steps, + block_type, + zoom=zoom, + nearest=nearest, + show_legend=(i == last_idx), + show_x_axis=(i == last_idx), + ) + for i, block_type in enumerate(present_block_types) + ] + return ( + alt.vconcat(*charts, spacing=0) + .resolve_scale(x="shared") + .properties(title="CUSUM Detector", padding=0) + .configure_view(strokeWidth=0) + ) + + def notify_slack( changepoints: list[Changepoint], slack_webhook: str, explorer_base_url: str = "https://explorer.ooni.org/", + detector_panel_base_url: str = "https://detector-panel.prod.ooni.io/", + edd: int = 10, + gap_halflife: float = 48.0, + warmup: bool = False, ): """ Sends a message to slack with a list of all changepoints that were detected @@ -725,10 +902,13 @@ def dir_to_str(dir: int) -> str: explorer = get_explorer_url(cp, explorer_base_url) # Alerts panel not yet deployed to prod, we use the test one for now alerts = get_alert_page_url(cp, "https://explorer.test.ooni.org/") + panel = get_detector_panel_url( + cp, detector_panel_base_url, edd=edd, gap_halflife=gap_halflife, warmup=warmup + ) message += ( f"• :flag-{cp['probe_cc'].lower()}: [{cp['probe_cc']}/AS{cp['probe_asn']}] " f"*{cp['domain']}* {dir_to_str(cp['change_dir'])} - `{cp['block_type']}` " - f"| <{explorer}|explorer> | <{alerts}|alerts>\n" + f"| <{explorer}|explorer> | <{alerts}|alerts> | <{panel}|detector panel>\n" ) # Send messages in 10 entries batches to avoid max message size limit @@ -786,3 +966,35 @@ def to_s(dt: datetime): } url = f"{base_url}/chart/alerts?{urlencode(params)}" return url + + +def get_detector_panel_url( + changepoint: Changepoint, + base_url: str = "https://detector-panel.prod.ooni.io/", + edd: int = 10, + gap_halflife: float = 48.0, + warmup: bool = False, +) -> str: + """ + Builds a link to the events-panel app (see + oonipipeline.events_panel.panel), prefilled via query params so the app + auto-runs the detector for this changepoint's metadata + """ + start_time = changepoint["ts"] - timedelta(days=13) + end_time = changepoint["ts"] + timedelta(days=2) + + def to_iso(dt: datetime) -> str: + return dt.strftime("%Y-%m-%dT%H:%M:%S") + + params = { + "probe_cc": changepoint["probe_cc"], + "domain": changepoint["domain"], + "probe_asn": changepoint["probe_asn"], + "start_time": to_iso(start_time), + "end_time": to_iso(end_time), + "edd": edd, + "gap_halflife": gap_halflife, + "warmup": str(warmup).lower(), + } + url = f"{base_url}?{urlencode(params)}" + return url diff --git a/oonipipeline/src/oonipipeline/events_panel/panel.py b/oonipipeline/src/oonipipeline/events_panel/panel.py index cfe28988..83c7f6f1 100644 --- a/oonipipeline/src/oonipipeline/events_panel/panel.py +++ b/oonipipeline/src/oonipipeline/events_panel/panel.py @@ -1,15 +1,17 @@ from collections import defaultdict import streamlit as st -from oonipipeline.analysis.detector import make_cusums_chart, run_detector_for, ANALYSIS_COLS -from datetime import datetime, timezone, timedelta, date +from oonipipeline.analysis.detector import ( + make_cusums_chart_grid, + run_detector_for, + ANALYSIS_COLS, +) +from datetime import datetime, timezone, timedelta import logging import pandas as pd +import vl_convert as vlc log = logging.getLogger(__name__) -def to_datetime(d: date): - return datetime(year=d.year, month=d.month, day=d.day, hour=0, tzinfo=timezone.utc) - @st.cache_data(ttl=300) def run_detector_cached(*args, **kwargs): @@ -17,6 +19,34 @@ def run_detector_cached(*args, **kwargs): st.set_page_config(layout="wide") +# When every one of these is present the form auto-runs +# on first load +QUERY_FIELD_TO_WIDGET_KEY = { + "probe_cc": "probe_cc_input", + "domain": "domain_input", + "start_time": "start_time_input", + "end_time": "end_time_input", + "edd": "edd_input", + "gap_halflife": "gap_halflife_input", + "warmup": "warmup_input", +} + + +def _parse_query_value(field: str, raw: str): + if field in ("start_time", "end_time"): + dt = datetime.fromisoformat(raw) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + if field in ("edd", "probe_asn"): + return int(raw) + if field == "gap_halflife": + return float(raw) + if field == "warmup": + return raw.strip().lower() in ("1", "true", "yes") + return raw + + def detector_panel(): st.write( """ @@ -29,6 +59,31 @@ def detector_panel(): now = datetime.now(timezone.utc) + # Prefill the form from URL query params on first load + if "query_params_applied" not in st.session_state: + st.session_state["query_params_applied"] = True + parsed_ok = set() + for field, widget_key in QUERY_FIELD_TO_WIDGET_KEY.items(): + raw = st.query_params.get(field) + if raw is not None: + try: + st.session_state[widget_key] = _parse_query_value(field, raw) + parsed_ok.add(field) + except (ValueError, TypeError): + pass + st.session_state["auto_submit_pending"] = parsed_ok == set( + QUERY_FIELD_TO_WIDGET_KEY + ) + + raw_probe_asn = st.query_params.get("probe_asn") + if raw_probe_asn is not None: + try: + st.session_state["query_probe_asn"] = _parse_query_value( + "probe_asn", raw_probe_asn + ) + except (ValueError, TypeError): + pass + clickhouse_url = st.sidebar.text_input( "**Clickhouse url**", "clickhouse://localhost:9000/ooni" ) @@ -37,25 +92,44 @@ def detector_panel(): c1, c2 = st.columns(2) # column 1 - start_time = c1.date_input("**Start time**", now - timedelta(days=30)) - probe_cc = c1.text_input("**Country code (two chars)**", "VE") - edd = c1.number_input("**Estimated Detection Delay (EDD)**", value=10) + start_time = c1.datetime_input( + "**Start date**", now - timedelta(days=30), key="start_time_input" + ) + probe_cc = c1.text_input( + "**Country code (two chars)**", "VE", key="probe_cc_input" + ) + edd = c1.number_input( + "**Estimated Detection Delay (EDD)**", value=10, key="edd_input" + ) # column2 - end_time = c2.date_input("**End time**", now) - domain = c2.text_input("**domain**", "x.com") - gap_halflife = c2.number_input("**Gap half life**", value=48.0) + end_time = c2.datetime_input("**End date**", now, key="end_time_input") + domain = c2.text_input("**domain**", "x.com", key="domain_input") + gap_halflife = c2.number_input( + "**Gap half life**", value=48.0, key="gap_halflife_input" + ) - warmup = st.checkbox("**Warmup**", True) + warmup = st.checkbox( + "**Warmup**", + False, + help="When enabled, no changepoints will be returned.", + key="warmup_input", + ) submitted = st.form_submit_button("Run detector") - # Only recompute on submit; store in session_state so results survive - # reruns triggered by the selectboxes below. - if submitted: + auto_submit = st.session_state.pop("auto_submit_pending", False) + + # store in session_state so results survive reruns triggered by + # the selectboxes below. + if submitted or auto_submit: + # Drop any cusum-related state left over + for key in ("changepoints", "cusum_steps", "block_type_select", "asn_select"): + st.session_state.pop(key, None) + changepoints, _, cusum_steps = run_detector_cached( clickhouse_url, - to_datetime(start_time), - to_datetime(end_time), + start_time, + end_time, probe_cc, [domain], edd, @@ -81,16 +155,40 @@ def detector_panel(): for step in cusum_steps: asns[step["probe_asn"]] += 1 + asns_with_changepoints = {cp["probe_asn"] for cp in changepoints} + c1, c2 = st.columns(2) c1.write(f"Changepoints: **{len(changepoints)}**") c2.write(f"Cusum steps: **{len(cusum_steps)}**") - c1, c2 = st.columns(2) - block_type = c1.selectbox("Block type", [c[0] for c in ANALYSIS_COLS]) + if changepoints: + st.write("**Changepoints**") + cp_df = pd.DataFrame(changepoints) + cp_df = cp_df.sort_values("ts", ascending=False).reset_index(drop=True) + display_cols = [ + c + for c in ["ts", "probe_asn", "domain", "block_type", "change_dir", "s_pos", "s_neg", "h"] + if c in cp_df.columns + ] + st.dataframe(cp_df[display_cols], hide_index=True) asn_list = list(asns.keys()) asn_list.sort(key=lambda k: asns[k], reverse=True) - selected_asn = c2.selectbox("ASN", asn_list, format_func=lambda k: f"{k} ({asns[k]})") + + query_probe_asn = st.session_state.get("query_probe_asn") + if ( + query_probe_asn is not None + and "asn_select" not in st.session_state + and query_probe_asn in asn_list + ): + st.session_state["asn_select"] = query_probe_asn + + selected_asn = st.selectbox( + "ASN", + asn_list, + format_func=lambda k: f"{'ā—ļø' if k in asns_with_changepoints else ''}{k} ({asns[k]})", + key="asn_select", + ) chart_steps = [s for s in cusum_steps if s["probe_asn"] == selected_asn] @@ -98,7 +196,34 @@ def detector_panel(): st.warning(f"No cusum steps found for ASN {selected_asn}") return - st.altair_chart(make_cusums_chart(chart_steps, block_type)) + block_types = [c[0] for c in ANALYSIS_COLS] + for block_type in block_types: + block_steps = [s for s in chart_steps if s["block_type"] == block_type] + if block_steps and all(pd.isna(s["obs_value"]) for s in block_steps): + st.info( + f"No **{block_type}** measurements were observed for this ASN in the " + "selected window (the underlying weight/count was zero every hour), so " + "the observed-value line has nothing to plot for that row. The CUSUM " + "statistics still reflect the detector's last known state." + ) + + chart = make_cusums_chart_grid(chart_steps, block_types) + if chart is None: + st.warning(f"No cusum steps found for ASN {selected_asn}") + return + st.altair_chart(chart) + + with st.expander("šŸ”§ Debug"): + if st.checkbox("Render chart as PNG", key="debug_render_png"): + spec = chart.to_dict() + png_bytes = vlc.vegalite_to_png(spec, scale=2) + st.image(png_bytes, caption="Chart rendered to PNG via vl-convert") + + if st.checkbox("Show Vega-Lite spec", key="debug_show_spec"): + st.json(chart.to_dict()) + + if st.checkbox("Show cusum data as dataframe", key="debug_show_cusum_df"): + st.dataframe(pd.DataFrame(chart_steps)) if asns: df = pd.DataFrame({"ASN": list(asns.keys()), "total": list(asns.values())})