Add map filtering controls with status, RAT, and search capabilities including auto-fit bounds, city/site search, dominant status classification, and interactive center override with fit-to-points functionality

panel_app/kpi_health_check_drilldown_plots.py

@@ -2,6 +2,7 @@ import pandas as pd
 import plotly.graph_objects as go
 from plotly.subplots import make_subplots
 
+
 def build_drilldown_plot(
     df: pd.DataFrame,
     kpis: list[str],
@@ -25,7 +26,7 @@ def build_drilldown_plot(
     # Sort by date
     # Sort by date
     plot_df = df.sort_values("date_only").copy()
-    
+
     title_text = f"{rat} - Site {site_code}"
     # If single KPI, be explicit in title
     if len(valid_kpis) == 1:
@@ -34,18 +35,19 @@ def build_drilldown_plot(
     # Subplot for Timeline (Streak) - Row 2
     # Row 1: Main Trend
     fig = make_subplots(
-        rows=2, cols=1, 
-        shared_xaxes=True, 
+        rows=2,
+        cols=1,
+        shared_xaxes=True,
         vertical_spacing=0.05,
         row_heights=[0.85, 0.15],
-        subplot_titles=(title_text, "Status Check")
+        subplot_titles=(title_text, "Status Check"),
     )
 
     for kpi in valid_kpis:
         # Data preparation
         x_data = plot_df["date_only"]
         y_data = pd.to_numeric(plot_df[kpi], errors="coerce")
-        
+
         # Add Trace
         fig.add_trace(
             go.Scatter(
@@ -55,7 +57,8 @@ def build_drilldown_plot(
                 name=kpi,
                 legendgroup=kpi,  # Allows grouping logic if needed
             ),
-            row=1, col=1
+            row=1,
+            col=1,
         )
 
         # Add SLA line if available
@@ -65,15 +68,23 @@ def build_drilldown_plot(
                 # Note: This implies rules_df is filtered for the correct RAT
                 rule = rules_df[rules_df["KPI"] == kpi]
                 if not rule.empty:
+                    pol = (
+                        str(rule.iloc[0].get("policy", "enforce") or "enforce")
+                        .strip()
+                        .lower()
+                    )
+                    if pol == "notify":
+                        continue
                     sla_val = pd.to_numeric(rule.iloc[0].get("sla"), errors="coerce")
                     if pd.notna(sla_val):
                         fig.add_hline(
-                            y=sla_val, 
-                            line_dash="dot", 
-                            line_color="red", 
+                            y=sla_val,
+                            line_dash="dot",
+                            line_color="red",
                             annotation_text=f"SLA {kpi}",
                             annotation_position="bottom right",
-                            row=1, col=1
+                            row=1,
+                            col=1,
                         )
             except Exception:
                 pass
@@ -90,16 +101,17 @@ def build_drilldown_plot(
                 mode="markers",
                 opacity=0,
                 showlegend=False,
-                hoverinfo="skip"
+                hoverinfo="skip",
             ),
-            row=2, col=1
+            row=2,
+            col=1,
         )
-    
+
     fig.update_layout(
         template="plotly_white",
         height=500,
         margin=dict(l=50, r=50, t=50, b=50),
-        legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
+        legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
     )
-    
+
     return fig

panel_app/kpi_health_check_panel.py

@@ -46,7 +46,11 @@ from process_kpi.kpi_health_check.profiles import (
     load_profile,
     save_profile,
 )
-from process_kpi.kpi_health_check.rules import infer_kpi_direction, infer_kpi_sla
+from process_kpi.kpi_health_check.rules import (
+    infer_kpi_direction,
+    infer_kpi_policy,
+    infer_kpi_sla,
+)
 
 pn.extension("plotly", "tabulator")
 
@@ -472,6 +476,33 @@ delta_table = pn.widgets.Tabulator(
 map_show_site_codes = pn.widgets.Checkbox(name="Show site codes", value=True)
 map_max_labels = pn.widgets.IntInput(name="Max labels", value=200, step=50)
 
+map_top_n = pn.widgets.IntInput(name="Top N sites (0=All)", value=0, step=100)
+map_impacted_rat = pn.widgets.Select(
+    name="RAT impacted", options=["Any", "2G", "3G", "LTE"], value="Any"
+)
+map_status_filter = pn.widgets.Select(
+    name="Status",
+    options=[
+        "Any",
+        "PERSISTENT_DEGRADED",
+        "DEGRADED",
+        "NOTIFY",
+        "RESOLVED",
+        "NOTIFY_RESOLVED",
+        "OK",
+        "NO_DATA",
+    ],
+    value="Any",
+)
+map_auto_fit = pn.widgets.Checkbox(name="Auto fit", value=True)
+map_fit_button = pn.widgets.Button(name="Fit to points", button_type="default")
+
+map_search_input = pn.widgets.TextInput(name="Search (site_code or City)", value="")
+map_search_go = pn.widgets.Button(name="Go", button_type="primary")
+map_search_clear = pn.widgets.Button(name="Clear", button_type="default")
+map_search_results = pn.widgets.Select(name="Matches", options={}, value=None)
+map_search_results.visible = False
+
 site_select = pn.widgets.AutocompleteInput(
     name="Select a site (Type to search)",
     options={},
@@ -536,7 +567,9 @@ function format(cell, formatterParams, onRendered){
   var bg = '#455a64';
   if(s === 'PERSISTENT_DEGRADED') bg = '#b71c1c';
   else if(s === 'DEGRADED') bg = '#e53935';
+  else if(s === 'NOTIFY') bg = '#f9a825';
   else if(s === 'RESOLVED') bg = '#2e7d32';
+  else if(s === 'NOTIFY_RESOLVED') bg = '#2e7d32';
   else if(s === 'NO_DATA') bg = '#616161';
   else if(s === 'OK') bg = '#1565c0';
 
@@ -552,9 +585,15 @@ function format(cell, formatterParams, onRendered){
         } else if(s === 'DEGRADED'){
           rowEl.style.borderLeft = '6px solid #e53935';
           rowEl.style.backgroundColor = '#fff5f5';
+        } else if(s === 'NOTIFY'){
+          rowEl.style.borderLeft = '6px solid #f9a825';
+          rowEl.style.backgroundColor = '#fff8e1';
         } else if(s === 'RESOLVED'){
           rowEl.style.borderLeft = '6px solid #2e7d32';
           rowEl.style.backgroundColor = '#f1f8e9';
+        } else if(s === 'NOTIFY_RESOLVED'){
+          rowEl.style.borderLeft = '6px solid #2e7d32';
+          rowEl.style.backgroundColor = '#f1f8e9';
         } else if(s === 'NO_DATA'){
           rowEl.style.borderLeft = '6px solid #616161';
           rowEl.style.backgroundColor = '#f5f5f5';
@@ -597,6 +636,10 @@ hist_plot_pane = pn.pane.Plotly(sizing_mode="stretch_both", config=PLOTLY_CONFIG
 map_pane = pn.pane.Plotly(sizing_mode="stretch_both", config=PLOTLY_CONFIG)
 map_message = pn.pane.Alert("", alert_type="info", visible=False)
 
+_map_search_city: str = ""
+_map_center_override: dict | None = None
+_map_force_fit: bool = False
+
 corr_plot_pane = pn.pane.Plotly(sizing_mode="stretch_both", config=PLOTLY_CONFIG)
 corr_message = pn.pane.Alert("", alert_type="info", visible=False)
 
@@ -624,6 +667,45 @@ def _coords_by_site() -> pd.DataFrame:
     return out
 
 
+def _dominant_status_by_site() -> pd.DataFrame:
+    s = (
+        current_status_df
+        if isinstance(current_status_df, pd.DataFrame)
+        else pd.DataFrame()
+    )
+    if (
+        s is None
+        or s.empty
+        or "site_code" not in s.columns
+        or "status" not in s.columns
+    ):
+        return pd.DataFrame(columns=["site_code", "dominant_status"])
+
+    tmp = s[["site_code", "status"]].copy()
+    tmp["site_code"] = pd.to_numeric(tmp["site_code"], errors="coerce")
+    tmp = tmp.dropna(subset=["site_code"]).copy()
+    if tmp.empty:
+        return pd.DataFrame(columns=["site_code", "dominant_status"])
+    tmp["site_code"] = tmp["site_code"].astype(int)
+    tmp["status"] = tmp["status"].astype(str).str.strip().str.upper()
+
+    prio = {
+        "PERSISTENT_DEGRADED": 6,
+        "DEGRADED": 5,
+        "NOTIFY": 4,
+        "RESOLVED": 3,
+        "NOTIFY_RESOLVED": 2,
+        "OK": 1,
+        "NO_DATA": 0,
+    }
+    tmp["_prio"] = tmp["status"].map(prio).fillna(2).astype(int)
+    idx = tmp.groupby("site_code")["_prio"].idxmax()
+    out = tmp.loc[idx, ["site_code", "status"]].rename(
+        columns={"status": "dominant_status"}
+    )
+    return out
+
+
 def _map_df() -> pd.DataFrame:
     base = (
         current_multirat_df
@@ -653,6 +735,77 @@ def _map_df() -> pd.DataFrame:
         out["City"] = out["City_coord"]
     if "City" in out.columns and "City_coord" in out.columns:
         out["City"] = out["City"].where(out["City"].notna(), out["City_coord"])
+
+    dom = _dominant_status_by_site()
+    if dom is not None and not dom.empty and "site_code" in dom.columns:
+        out = pd.merge(out, dom, on="site_code", how="left")
+    if "dominant_status" not in out.columns:
+        out["dominant_status"] = "NO_DATA"
+    out["dominant_status"] = (
+        out["dominant_status"]
+        .astype(str)
+        .str.strip()
+        .str.upper()
+        .replace({"": "NO_DATA"})
+    )
+    out["dominant_status"] = out["dominant_status"].where(
+        out["dominant_status"].notna(), "NO_DATA"
+    )
+    return out
+
+
+def _apply_map_filters(df_map: pd.DataFrame) -> pd.DataFrame:
+    if df_map is None or df_map.empty:
+        return pd.DataFrame()
+    out = df_map.copy()
+
+    if _map_search_city:
+        if "City" in out.columns:
+            q = str(_map_search_city).strip().lower()
+            out = out[
+                out["City"].astype(str).str.lower().str.contains(q, na=False)
+            ].copy()
+        else:
+            out = out.iloc[0:0].copy()
+
+    rat_imp = str(map_impacted_rat.value or "Any")
+    if rat_imp and rat_imp != "Any":
+        pcol = f"persistent_{rat_imp}"
+        dcol = f"degraded_{rat_imp}"
+        if pcol in out.columns or dcol in out.columns:
+            p = (
+                pd.to_numeric(out[pcol], errors="coerce").fillna(0)
+                if pcol in out.columns
+                else 0
+            )
+            d = (
+                pd.to_numeric(out[dcol], errors="coerce").fillna(0)
+                if dcol in out.columns
+                else 0
+            )
+            out = out[(p.astype(float) > 0) | (d.astype(float) > 0)].copy()
+        else:
+            out = out.iloc[0:0].copy()
+
+    st = str(map_status_filter.value or "Any")
+    if st and st != "Any" and "dominant_status" in out.columns:
+        out = out[out["dominant_status"].astype(str).str.upper() == st].copy()
+
+    score_col = (
+        "criticality_score_weighted"
+        if "criticality_score_weighted" in out.columns
+        else "criticality_score"
+    )
+    if score_col in out.columns:
+        out["_score"] = (
+            pd.to_numeric(out[score_col], errors="coerce").fillna(0).astype(float)
+        )
+    else:
+        out["_score"] = 0.0
+
+    topn = _coerce_int(map_top_n.value)
+    if topn is not None and int(topn) > 0:
+        out = out.sort_values(by=["_score"], ascending=False).head(int(topn)).copy()
     return out
 
 
@@ -687,11 +840,31 @@ def _build_map_fig(df_map: pd.DataFrame) -> go.Figure | None:
     size = (tmp["_score"].clip(lower=0) + 1.0).pow(0.5) * 6.0
     tmp["_size"] = size.clip(lower=6, upper=28)
 
+    dom_order = [
+        "PERSISTENT_DEGRADED",
+        "DEGRADED",
+        "NOTIFY",
+        "RESOLVED",
+        "NOTIFY_RESOLVED",
+        "OK",
+        "NO_DATA",
+    ]
+    color_map = {
+        "PERSISTENT_DEGRADED": "#8b0000",
+        "DEGRADED": "#ff6f00",
+        "NOTIFY": "#f9a825",
+        "RESOLVED": "#2e7d32",
+        "NOTIFY_RESOLVED": "#2e7d32",
+        "OK": "#1565c0",
+        "NO_DATA": "#616161",
+    }
+
     hover_cols = [
         c
         for c in [
             "site_code",
             "City",
+            "dominant_status",
             score_col,
             "impacted_rats",
             "persistent_kpis_total",
@@ -699,16 +872,19 @@ def _build_map_fig(df_map: pd.DataFrame) -> go.Figure | None:
         ]
         if c and c in tmp.columns
     ]
+
     fig = px.scatter_mapbox(
         tmp,
         lat="Latitude",
         lon="Longitude",
-        color="_score" if score_col is not None else None,
+        color="dominant_status" if "dominant_status" in tmp.columns else None,
         size="_size",
         size_max=28,
         zoom=4,
         hover_data=hover_cols,
         custom_data=["site_code"],
+        category_orders={"dominant_status": dom_order},
+        color_discrete_map=color_map,
     )
 
     try:
@@ -740,8 +916,49 @@ def _build_map_fig(df_map: pd.DataFrame) -> go.Figure | None:
         margin=dict(l=10, r=10, t=10, b=10),
         height=700,
     )
-    if score_col is not None:
-        fig.update_layout(coloraxis_colorbar=dict(title="Score"))
+
+    global _map_center_override, _map_force_fit
+    do_fit = bool(map_auto_fit.value) or bool(_map_force_fit)
+    if (
+        _map_center_override
+        and "lat" in _map_center_override
+        and "lon" in _map_center_override
+    ):
+        try:
+            fig.update_layout(
+                mapbox_center=dict(
+                    lat=float(_map_center_override["lat"]),
+                    lon=float(_map_center_override["lon"]),
+                ),
+                mapbox_zoom=float(_map_center_override.get("zoom", 9.0)),
+            )
+            do_fit = False
+        except Exception:  # noqa: BLE001
+            pass
+
+    if do_fit:
+        try:
+            lat_min = float(tmp["Latitude"].min())
+            lat_max = float(tmp["Latitude"].max())
+            lon_min = float(tmp["Longitude"].min())
+            lon_max = float(tmp["Longitude"].max())
+
+            lat_pad = max(0.01, abs(lat_max - lat_min) * 0.1)
+            lon_pad = max(0.01, abs(lon_max - lon_min) * 0.1)
+            fig.update_layout(
+                mapbox=dict(
+                    bounds=dict(
+                        west=lon_min - lon_pad,
+                        east=lon_max + lon_pad,
+                        south=lat_min - lat_pad,
+                        north=lat_max + lat_pad,
+                    )
+                )
+            )
+        except Exception:  # noqa: BLE001
+            pass
+
+    _map_force_fit = False
     return fig
 
 
@@ -755,6 +972,14 @@ def _refresh_map_view(event=None) -> None:
         map_message.visible = True
         return
 
+    df_map = _apply_map_filters(df_map)
+    if df_map is None or df_map.empty:
+        map_pane.object = None
+        map_message.alert_type = "warning"
+        map_message.object = "No sites to display after Map filters/search."
+        map_message.visible = True
+        return
+
     fig = _build_map_fig(df_map)
     if fig is None:
         map_pane.object = None
@@ -769,6 +994,125 @@ def _refresh_map_view(event=None) -> None:
     map_pane.object = fig
 
 
+def _on_map_fit_click(event=None) -> None:
+    global _map_force_fit, _map_center_override
+    _map_center_override = None
+    _map_force_fit = True
+    _refresh_map_view()
+
+
+def _on_map_search(event=None) -> None:
+    global _map_search_city, _map_center_override
+    q = str(map_search_input.value or "").strip()
+
+    _map_search_city = ""
+    _map_center_override = None
+    map_search_results.options = {}
+    map_search_results.value = None
+    map_search_results.visible = False
+
+    if not q:
+        _refresh_map_view()
+        return
+
+    code = _coerce_int(q)
+    if code is not None:
+        try:
+            df_map = _map_df()
+            if isinstance(df_map, pd.DataFrame) and not df_map.empty:
+                m = df_map[df_map["site_code"] == int(code)]
+                if not m.empty and "Latitude" in m.columns and "Longitude" in m.columns:
+                    lat = pd.to_numeric(m.iloc[0].get("Latitude"), errors="coerce")
+                    lon = pd.to_numeric(m.iloc[0].get("Longitude"), errors="coerce")
+                    if pd.notna(lat) and pd.notna(lon):
+                        _map_center_override = {
+                            "lat": float(lat),
+                            "lon": float(lon),
+                            "zoom": 10.5,
+                        }
+        except Exception:  # noqa: BLE001
+            _map_center_override = None
+
+        _apply_drilldown_selection(site_code=int(code))
+        _refresh_map_view()
+        return
+
+    _map_search_city = q
+    try:
+        df_map = _map_df()
+        df_map = _apply_map_filters(df_map)
+        if (
+            isinstance(df_map, pd.DataFrame)
+            and not df_map.empty
+            and "City" in df_map.columns
+        ):
+            m = df_map[
+                df_map["City"].astype(str).str.lower().str.contains(q.lower(), na=False)
+            ].copy()
+            if not m.empty:
+                score_col = (
+                    "criticality_score_weighted"
+                    if "criticality_score_weighted" in m.columns
+                    else "criticality_score"
+                )
+                if score_col in m.columns:
+                    m["_score"] = pd.to_numeric(m[score_col], errors="coerce").fillna(0)
+                    m = m.sort_values(by=["_score"], ascending=False)
+                m = m.head(20)
+                opts = {}
+                for _, r in m.iterrows():
+                    sc = _coerce_int(r.get("site_code"))
+                    if sc is None:
+                        continue
+                    city = str(r.get("City") or "").strip()
+                    label = f"{int(sc)} - {city}" if city else f"{int(sc)}"
+                    opts[label] = int(sc)
+                map_search_results.options = opts
+                map_search_results.visible = bool(opts)
+    except Exception:  # noqa: BLE001
+        map_search_results.options = {}
+        map_search_results.visible = False
+
+    _refresh_map_view()
+
+
+def _on_map_search_pick(event=None) -> None:
+    global _map_center_override
+    code = _coerce_int(getattr(event, "new", None))
+    if code is None:
+        return
+
+    try:
+        df_map = _map_df()
+        if isinstance(df_map, pd.DataFrame) and not df_map.empty:
+            m = df_map[df_map["site_code"] == int(code)]
+            if not m.empty and "Latitude" in m.columns and "Longitude" in m.columns:
+                lat = pd.to_numeric(m.iloc[0].get("Latitude"), errors="coerce")
+                lon = pd.to_numeric(m.iloc[0].get("Longitude"), errors="coerce")
+                if pd.notna(lat) and pd.notna(lon):
+                    _map_center_override = {
+                        "lat": float(lat),
+                        "lon": float(lon),
+                        "zoom": 10.5,
+                    }
+    except Exception:  # noqa: BLE001
+        _map_center_override = None
+
+    _apply_drilldown_selection(site_code=int(code))
+    _refresh_map_view()
+
+
+def _on_map_clear_search(event=None) -> None:
+    global _map_search_city, _map_center_override
+    _map_search_city = ""
+    _map_center_override = None
+    map_search_input.value = ""
+    map_search_results.options = {}
+    map_search_results.value = None
+    map_search_results.visible = False
+    _refresh_map_view()
+
+
 def _on_map_click(event) -> None:
     try:
         cd = event.new
@@ -1457,9 +1801,11 @@ def _update_site_view(event=None) -> None:
         and "KPI" in site_df.columns
     ):
         sev = {
-            "PERSISTENT_DEGRADED": 3,
-            "DEGRADED": 2,
+            "PERSISTENT_DEGRADED": 4,
+            "DEGRADED": 3,
+            "NOTIFY": 2,
             "RESOLVED": 1,
+            "NOTIFY_RESOLVED": 0,
             "OK": 0,
             "NO_DATA": -1,
         }
@@ -1739,12 +2085,15 @@ def _build_site_heatmap(
             continue
         rule = _infer_rule_row(rules_df, rat, kpi)
         direction = str(rule.get("direction", "higher_is_better"))
+        policy = str(rule.get("policy", "enforce") or "enforce").strip().lower()
         sla_raw = rule.get("sla", None)
         try:
             sla_val = float(sla_raw) if pd.notna(sla_raw) else None
         except Exception:  # noqa: BLE001
             sla_val = None
 
+        sla_eval = None if policy == "notify" else sla_val
+
         s = site_daily[["date_only", kpi]].dropna(subset=[kpi])
         baseline_mask = (s["date_only"] >= baseline_start) & (
             s["date_only"] <= baseline_end
@@ -1766,7 +2115,7 @@ def _build_site_heatmap(
                 baseline_val,
                 direction,
                 float(rel_threshold_pct.value),
-                sla_val,
+                sla_eval,
             )
             row_z.append(1 if bad else 0)
             row_h.append(
@@ -2449,7 +2798,7 @@ def _apply_preset(event=None) -> None:
         return
 
     key = ["RAT", "KPI"]
-    upd_cols = [c for c in ["direction", "sla"] if c in preset_df.columns]
+    upd_cols = [c for c in ["direction", "sla", "policy"] if c in preset_df.columns]
     preset_df2 = preset_df[key + upd_cols].copy()
 
     merged = pd.merge(cur, preset_df2, on=key, how="left", suffixes=("", "_preset"))
@@ -2550,6 +2899,17 @@ def load_datasets(event=None) -> None:
         delta_table.value = pd.DataFrame()
         map_pane.object = None
         map_message.visible = False
+        try:
+            global _map_search_city, _map_center_override, _map_force_fit
+            _map_search_city = ""
+            _map_center_override = None
+            _map_force_fit = False
+            map_search_input.value = ""
+            map_search_results.options = {}
+            map_search_results.value = None
+            map_search_results.visible = False
+        except Exception:  # noqa: BLE001
+            pass
         site_kpi_table.value = pd.DataFrame()
         trend_plot_pane.object = None
         heatmap_plot_pane.object = None
@@ -2597,13 +2957,14 @@ def load_datasets(event=None) -> None:
             )
 
             for kpi in kpi_cols:
-                direction = infer_kpi_direction(kpi)
+                direction = infer_kpi_direction(kpi, rat)
                 rules_rows.append(
                     {
                         "RAT": rat,
                         "KPI": kpi,
                         "direction": direction,
-                        "sla": infer_kpi_sla(kpi, direction),
+                        "sla": infer_kpi_sla(kpi, direction, rat),
+                        "policy": infer_kpi_policy(kpi, rat),
                     }
                 )
 
@@ -3026,6 +3387,11 @@ snapshot_download.callback = _snapshot_download_callback
 
 map_pane.param.watch(_on_map_click, "click_data")
 
+map_fit_button.on_click(_on_map_fit_click)
+map_search_go.on_click(_on_map_search)
+map_search_clear.on_click(_on_map_clear_search)
+map_search_results.param.watch(_on_map_search_pick, "value")
+
 _refresh_presets()
 _refresh_profiles()
 _refresh_complaint_sites()
@@ -3073,6 +3439,11 @@ corr_window_select.param.watch(_on_drilldown_change, "value")
 map_show_site_codes.param.watch(lambda e: _refresh_map_view(), "value")
 map_max_labels.param.watch(lambda e: _refresh_map_view(), "value")
 
+map_top_n.param.watch(lambda e: _refresh_map_view(), "value")
+map_impacted_rat.param.watch(lambda e: _refresh_map_view(), "value")
+map_status_filter.param.watch(lambda e: _refresh_map_view(), "value")
+map_auto_fit.param.watch(lambda e: _refresh_map_view(), "value")
+
 analysis_range.param.watch(_on_drilldown_params_change, "value")
 baseline_days.param.watch(_on_drilldown_params_change, "value")
 recent_days.param.watch(_on_drilldown_params_change, "value")
@@ -3380,7 +3751,17 @@ _tab_delta = pn.Column(
 _tab_map = pn.Column(
     pn.pane.Markdown("## Map"),
     map_message,
-    pn.Row(map_show_site_codes, map_max_labels),
+    pn.Row(
+        map_search_input, map_search_go, map_search_clear, map_fit_button, map_auto_fit
+    ),
+    pn.Row(
+        map_top_n,
+        map_impacted_rat,
+        map_status_filter,
+        map_show_site_codes,
+        map_max_labels,
+    ),
+    pn.Row(map_search_results),
     pn.Column(map_pane, sizing_mode="stretch_both", min_height=700),
     sizing_mode="stretch_both",
 )

process_kpi/kpi_health_check/engine.py

@@ -30,8 +30,8 @@ def is_bad(
 
     thr = float(rel_threshold_pct) / 100.0
     if direction == "higher_is_better":
-        return bad or (value < baseline * (1.0 - thr))
-    return bad or (value > baseline * (1.0 + thr))
+        return bad or (value < baseline - abs(baseline) * thr)
+    return bad or (value > baseline + abs(baseline) * thr)
 
 
 def max_consecutive_days(dates: list[date]) -> int:
@@ -83,12 +83,15 @@ def evaluate_health_check(
         for kpi in kpis:
             rule = rat_rules[rat_rules["KPI"] == kpi].iloc[0]
             direction = str(rule.get("direction", "higher_is_better"))
+            policy = str(rule.get("policy", "enforce") or "enforce").strip().lower()
             sla = rule.get("sla", np.nan)
             try:
                 sla_val = float(sla) if pd.notna(sla) else None
             except Exception:
                 sla_val = None
 
+            sla_eval = None if policy == "notify" else sla_val
+
             s = g_site[["date_only", kpi]].dropna(subset=[kpi])
             if s.empty:
                 rows.append(
@@ -125,7 +128,7 @@ def evaluate_health_check(
                         float(baseline) if pd.notna(baseline) else None,
                         direction,
                         rel_threshold_pct,
-                        sla_val,
+                        sla_eval,
                     ):
                         bad_dates.append(d)
 
@@ -137,7 +140,7 @@ def evaluate_health_check(
                 float(baseline) if pd.notna(baseline) else None,
                 direction,
                 rel_threshold_pct,
-                sla_val,
+                sla_eval,
             )
 
             is_bad_current = is_bad_recent
@@ -149,19 +152,27 @@ def evaluate_health_check(
                     float(baseline) if pd.notna(baseline) else None,
                     direction,
                     rel_threshold_pct,
-                    sla_val,
+                    sla_eval,
                 )
 
             had_bad_recent = (len(bad_dates) > 0) or bool(is_bad_recent)
 
-            if is_bad_current and persistent:
-                status = "PERSISTENT_DEGRADED"
-            elif is_bad_current:
-                status = "DEGRADED"
-            elif had_bad_recent:
-                status = "RESOLVED"
+            if policy == "notify":
+                if is_bad_current:
+                    status = "NOTIFY"
+                elif had_bad_recent:
+                    status = "NOTIFY_RESOLVED"
+                else:
+                    status = "OK"
             else:
-                status = "OK"
+                if is_bad_current and persistent:
+                    status = "PERSISTENT_DEGRADED"
+                elif is_bad_current:
+                    status = "DEGRADED"
+                elif had_bad_recent:
+                    status = "RESOLVED"
+                else:
+                    status = "OK"
 
             rows.append(
                 {
@@ -171,6 +182,7 @@ def evaluate_health_check(
                     "KPI": kpi,
                     "direction": direction,
                     "sla": sla_val,
+                    "policy": policy,
                     "baseline_median": baseline,
                     "recent_median": recent,
                     "bad_days_recent": len(bad_dates),

process_kpi/kpi_health_check/presets.py

@@ -56,7 +56,7 @@ def save_preset(name: str, rules_df: pd.DataFrame) -> str:
     if df.empty:
         raise ValueError("Rules dataframe is empty")
 
-    keep = [c for c in ["RAT", "KPI", "direction", "sla"] if c in df.columns]
+    keep = [c for c in ["RAT", "KPI", "direction", "sla", "policy"] if c in df.columns]
     df = df[keep].copy()
 
     obj = {

process_kpi/kpi_health_check/rules.py

@@ -1,5 +1,78 @@
-def infer_kpi_direction(kpi: str) -> str:
-    k = str(kpi).lower()
+import json
+import os
+import re
+from functools import lru_cache
+
+
+def _norm(value: str) -> str:
+    s = str(value or "").strip().lower()
+    s = re.sub(r"[^0-9a-z]+", " ", s)
+    s = re.sub(r"\s+", " ", s).strip()
+    return s
+
+
+def _project_root() -> str:
+    return os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+
+
+def _load_curated_rows() -> list[dict]:
+    path = os.path.join(
+        _project_root(), "data", "kpi_health_check_presets", "Test1.json"
+    )
+    if not os.path.exists(path):
+        return []
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            txt = f.read()
+        txt = re.sub(r"\bNaN\b", "null", txt)
+        obj = json.loads(txt)
+        rows = obj.get("rules", []) if isinstance(obj, dict) else []
+        return rows if isinstance(rows, list) else []
+    except Exception:  # noqa: BLE001
+        return []
+
+
+@lru_cache(maxsize=1)
+def _curated_rules_map() -> dict[tuple[str, str], dict]:
+    out: dict[tuple[str, str], dict] = {}
+    for r in _load_curated_rows():
+        if not isinstance(r, dict):
+            continue
+        rat = _norm(r.get("RAT"))
+        kpi = _norm(r.get("KPI"))
+        if not rat or not kpi:
+            continue
+
+        direction = str(r.get("direction") or "").strip()
+        policy_raw = str(r.get("policy") or "").strip().lower()
+        policy = policy_raw if policy_raw in {"enforce", "notify"} else None
+        sla_raw = r.get("sla", None)
+        try:
+            sla = float(sla_raw) if sla_raw is not None else None
+        except Exception:  # noqa: BLE001
+            sla = None
+
+        out[(rat, kpi)] = {
+            "direction": direction or None,
+            "sla": sla,
+            "policy": policy,
+        }
+    return out
+
+
+def _curated_rule(kpi: str, rat: str | None = None) -> dict | None:
+    if not kpi or not rat:
+        return None
+    key = (_norm(rat), _norm(kpi))
+    return _curated_rules_map().get(key)
+
+
+def infer_kpi_direction(kpi: str, rat: str | None = None) -> str:
+    curated = _curated_rule(kpi, rat)
+    if curated and curated.get("direction"):
+        return str(curated["direction"])
+
+    k = _norm(kpi)
     lower_is_better = [
         "drop",
         "dcr",
@@ -18,8 +91,15 @@ def infer_kpi_direction(kpi: str) -> str:
     return "higher_is_better"
 
 
-def infer_kpi_sla(kpi: str, direction: str) -> float | None:
-    k = str(kpi).lower()
+def infer_kpi_sla(kpi: str, direction: str, rat: str | None = None) -> float | None:
+    curated = _curated_rule(kpi, rat)
+    if curated and curated.get("sla") is not None:
+        try:
+            return float(curated["sla"])
+        except Exception:  # noqa: BLE001
+            pass
+
+    k = _norm(kpi)
     if direction == "higher_is_better" and any(
         x in k for x in ["availability", "cssr", "success", " sr"]
     ):
@@ -29,3 +109,14 @@ def infer_kpi_sla(kpi: str, direction: str) -> float | None:
     ):
         return 2.0
     return None
+
+
+def infer_kpi_policy(kpi: str, rat: str | None = None) -> str:
+    curated = _curated_rule(kpi, rat)
+    if curated and curated.get("policy"):
+        return str(curated["policy"])
+
+    k = _norm(kpi)
+    if "distance" in k:
+        return "notify"
+    return "enforce"