Add labeling API for building a labeled corpus

ooniapi/services/oonimeasurements/src/oonimeasurements/main.py

@@ -18,6 +18,7 @@
     list_observations,
 )
 from .routers.v1 import aggregation, measurements
+from .routers import labeling
 
 pkg_name = "oonimeasurements"
 
@@ -37,7 +38,8 @@ def create_app() -> FastAPI:
     app.add_middleware(
         CORSMiddleware,
         # allow from observable notebooks
-        allow_origin_regex=r"^https://[-A-Za-z0-9]+(\.(test|dev))?\.ooni\.(org|io)$|^https://.*\.observableusercontent\.com$",
+        #allow_origin_regex=r"^https://[-A-Za-z0-9]+(\.(test|dev))?\.ooni\.(org|io)$|^https://.*\.observableusercontent\.com$",
+        allow_origins=["*"],
         # allow_origin_regex="^https://[-A-Za-z0-9]+(\.test)?\.ooni\.(org|io)$",
         allow_credentials=True,
         allow_methods=["*"],
@@ -60,6 +62,7 @@ class HealthStatus(BaseModel):
     app.include_router(list_observations.router, prefix="/api")
     app.include_router(aggregate_observations.router, prefix="/api")
     app.include_router(aggregate_analysis.router, prefix="/api")
+    app.include_router(labeling.router)
 
     instrumentor = Instrumentator().instrument(
         app, metric_namespace="ooniapi", metric_subsystem="oonimeasurements"

ooniapi/services/oonimeasurements/src/oonimeasurements/routers/data/aggregate_analysis.py

@@ -12,6 +12,12 @@
 
 from ...common.clickhouse_utils import async_query_click
 from ...dependencies import ClickhouseDep, get_clickhouse_session
+from ...scoring import (
+    BLOCKING_THRESHOLD,
+    SCORING_VERSION,
+    Calibration,
+    blocked_probability_sql,
+)
 from ...utils.api import ProbeASNOrNone, ProbeCCOrNone
 from .list_analysis import (
     SinceUntil,
@@ -50,6 +56,9 @@ class Loni(BaseModel):
     tls_down: Optional[float]
     tls_ok: Optional[float]
 
+    # Which layers cleared BLOCKING_THRESHOLD. Reading this needs to know the
+    # threshold that produced it, so the response carries both — see
+    # scoring_version / blocking_threshold on the response body.
     likely_blocked_protocols: List[Tuple[str, float]]
     blocked_max_outcome: Optional[str]
     blocked_max: Optional[float]
@@ -58,6 +67,13 @@ class Loni(BaseModel):
     tcp_blocked_outcome: Optional[str]
     tls_blocked_outcome: Optional[str]
 
+    # Mean calibrated P(blocked) over the measurements in this cell, i.e. the
+    # estimated share of them that are blocked. Deliberately the mean of the
+    # per-measurement probabilities and not the probability of the aggregated
+    # score: the latter would answer "is the 99th-percentile measurement
+    # blocked", which is not a question anyone asks.
+    blocked_probability_mean: Optional[float]
+
 
 class AggregationEntry(BaseModel):
     count: float
@@ -79,6 +95,15 @@ class AggregationResponse(BaseModel):
     db_stats: DBStats
     dimension_count: int
     results: List[AggregationEntry]
+    # The scoring regime behind likely_blocked_protocols. Two responses with
+    # different scoring_version are not comparable, so it travels with the
+    # verdicts rather than living only in a deploy log.
+    blocking_threshold: float = BLOCKING_THRESHOLD
+    scoring_version: str = SCORING_VERSION
+    # The corpus the probabilities were calibrated against, and the range over
+    # which that corpus actually constrains them.
+    calibration_corpus: str = Calibration.CORPUS
+    calibration_trustworthy_range: Tuple[float, float] = Calibration.TRUSTWORTHY_RANGE
 
 
 # editable chart link: https://excalidraw.com/#json=mnoOrMXdSDLVirr8Albuu,xRyHC8-8JlsTTEovwNxOdQ
@@ -183,6 +208,7 @@ def format_aggregate_query(extra_cols: Dict[str, str], where: str):
     {",".join(extra_cols.keys())},
     probe_analysis,
     count,
+    blocked_probability_mean,
 
     dns_blocked_q99 as dns_blocked,
     dns_down_q99 as dns_down,
@@ -215,7 +241,10 @@ def format_aggregate_query(extra_cols: Dict[str, str], where: str):
         arraySort(
             x -> -x.1,
             arrayFilter(
-                x -> x.1 > 0.5,
+                -- Threshold from scoring.BLOCKING_THRESHOLD. Was a literal
+                -- `> 0.5` here and `>= 0.5` in the labeling router; they
+                -- disagreed at exactly 0.5, which dns.failure_no_ctrl hits.
+                x -> x.1 >= {BLOCKING_THRESHOLD},
                 [
                     (dns_blocked, 'dns'),
                     (tcp_blocked, 'tcp'),
@@ -243,6 +272,8 @@ def format_aggregate_query(extra_cols: Dict[str, str], where: str):
             {",".join(extra_cols.values())},
             COUNT() as count,
 
+            avg({blocked_probability_sql()}) as blocked_probability_mean,
+
             anyHeavy(top_probe_analysis) as probe_analysis,
 
             topKWeighted(10, 3, 'counts')(
@@ -415,6 +446,7 @@ async def get_aggregation_analysis(
             dns_blocked_outcome=d["dns_blocked_outcome"],
             tcp_blocked_outcome=d["tcp_blocked_outcome"],
             tls_blocked_outcome=d["tls_blocked_outcome"],
+            blocked_probability_mean=nan_to_none(d.get("blocked_probability_mean")),
         )
 
         entry = AggregationEntry(
@@ -438,6 +470,8 @@ async def get_aggregation_analysis(
         ),
         dimension_count=dimension_count,
         results=results,
+        blocking_threshold=BLOCKING_THRESHOLD,
+        scoring_version=SCORING_VERSION,
     )
 
 

ooniapi/services/oonimeasurements/src/oonimeasurements/routers/data/list_analysis.py

@@ -11,6 +11,7 @@
 from ...common.clickhouse_utils import async_query_click
 from ...common.dependencies import get_settings
 from ...dependencies import get_clickhouse_session
+from ...scoring import SCORING_VERSION, blocked_probability_sql
 from .utils import (
     SinceUntil,
     parse_probe_asn_to_int,
@@ -62,11 +63,18 @@ class AnalysisEntry(BaseModel):
     tls_blocked: float
     tls_down: float
     tls_ok: float
+    # Calibrated P(blocked) for this measurement, from scoring.Calibration.
+    # Computed in the query rather than stored, so a recalibration is a deploy
+    # and not a reprocess.
+    blocked_probability: float
 
 
 class ListAnalysisResponse(BaseModel):
     metadata: ResponseMetadata
     results: List[AnalysisEntry]
+    # Which calibration produced blocked_probability. A probability is only
+    # interpretable next to the fit behind it.
+    scoring_version: str = SCORING_VERSION
 
 
 @router.get("/v1/analysis", tags=["analysis", "list_data"])
@@ -126,7 +134,13 @@ async def list_measurements(
         and_clauses.append("measurement_start_time <= %(until)s")
         q_args["until"] = until
 
-    cols = list(AnalysisEntry.model_json_schema()["properties"].keys())
+    # Every field is a column except the calibrated probability, which is
+    # derived. Keep the alias so the row still maps onto AnalysisEntry.
+    derived = {"blocked_probability": blocked_probability_sql()}
+    cols = [
+        f"{derived[c]} AS {c}" if c in derived else c
+        for c in AnalysisEntry.model_json_schema()["properties"].keys()
+    ]
     q = f"SELECT {','.join(cols)} FROM analysis_web_measurement"
     if len(and_clauses) > 0:
         q += " WHERE "