#!/usr/bin/env python3 """Local utility outage providers for the NWAL Scanner platform. Normalizes Florence Electricity and Sheffield Utilities into the county-shaped JSON payload expected by the existing broadcast display. """ from __future__ import annotations import json import re import ssl import threading import time import urllib.request from datetime import datetime, timezone from typing import Any FLORENCE_URL = "http://68.208.116.30:7576/data/outageSummary.json?v=2" SHEFFIELD_URL = "https://ebill.sheffieldutils.org/maps/Outage_Web_Map/maps/GWT.rpc" SHEFFIELD_ORIGIN = "https://ebill.sheffieldutils.org" SHEFFIELD_REFERER = "https://ebill.sheffieldutils.org/maps/Outage_Web_Map/" SHEFFIELD_MODULE_BASE = "https://ebill.sheffieldutils.org/maps/Outage_Web_Map/maps/" SHEFFIELD_PERMUTATION = "1D9BF682AF8D25FA7709856C83E9910F" SHEFFIELD_BODY = ( "7|0|4|https://ebill.sheffieldutils.org/maps/Outage_Web_Map/maps/|" "612278413EC26C34D54A3907AA0CDFD8|" "coop.nisc.oms.webmap.services.RpcCombinedOutageDetailsService|" "getCombinedOutageDetails|1|2|3|4|0|" ) CACHE_SECONDS = 60 STALE_AFTER_SECONDS = 15 * 60 _cache_lock = threading.Lock() _cache: dict[str, Any] | None = None _cache_time = 0.0 def _utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def _parse_iso_age_seconds(value: Any) -> int | None: if not value: return None try: dt = datetime.fromisoformat(str(value).replace("Z", "+00:00")) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return max(0, int((datetime.now(timezone.utc) - dt.astimezone(timezone.utc)).total_seconds())) except Exception: return None def _fetch_bytes(request: urllib.request.Request, timeout: int = 10) -> bytes: context = ssl.create_default_context() with urllib.request.urlopen(request, timeout=timeout, context=context) as response: return response.read() def fetch_florence() -> dict[str, Any]: request = urllib.request.Request( FLORENCE_URL, headers={ "Accept": "application/json, text/plain, */*", "Cache-Control": "no-cache", "Pragma": "no-cache", "Referer": "http://68.208.116.30:7576/", "User-Agent": "NWALScannerPlatform/018 (+local outage adapter)", }, ) raw = _fetch_bytes(request) payload = json.loads(raw.decode("utf-8-sig")) customers_out = int(payload.get("customersOutNow") or 0) customers_served = int(payload.get("customersServed") or 0) updated = payload.get("updateTime") or payload.get("lastUpdated") or _utc_now_iso() age = _parse_iso_age_seconds(updated) return { "US_Full_FIPS": "01077", "CountyName": "Lauderdale", "StateAbbr": "AL", "CustomersTracked": customers_served, "CustomersOut": customers_out, "OutageEvents": int(payload.get("outageCount") or (1 if customers_out > 0 else 0)), "LastUpdatedDateTime": updated, "ProviderName": "Florence Electricity", "ProviderArea": "Florence Electricity service area", "SourceURL": FLORENCE_URL, "Stale": age is not None and age > STALE_AFTER_SECONDS, "AgeSeconds": age, "RawSummary": { "customersAffected": payload.get("customersAffected"), "customersRestored": payload.get("customersRestored"), "streetsAffected": payload.get("streetsAffected"), }, } def _decode_sheffield_response(text: str) -> dict[str, Any]: if not text.startswith("//OK"): raise ValueError("Sheffield returned a non-OK GWT response") stream = json.loads(text[4:]) if not isinstance(stream, list) or len(stream) < 4 or not isinstance(stream[-3], list): raise ValueError("Unexpected Sheffield GWT response shape") values = stream[:-3] strings = stream[-3] # Active outage IDs are six-or-more digit numeric strings adjacent to the # serialized Outage class entries in the response string table. outage_ids = [ item for item in strings if isinstance(item, str) and re.fullmatch(r"\d{6,}", item) ] # NISC serializes every outage record with its projected X/Y coordinates, # each followed by the java.lang.Double type reference (55 in this # permutation). The current customer count is four scalar fields later. # This avoids relying on a fixed record offset and works for one or many # outage objects in the same response. customer_counts: list[int] = [] coordinates: list[dict[str, float]] = [] for i in range(0, max(0, len(values) - 10)): try: x, x_type, y, y_type = values[i + 1], values[i + 2], values[i + 3], values[i + 4] if ( isinstance(x, (int, float)) and abs(float(x)) > 100000 and x_type == 55 and isinstance(y, (int, float)) and abs(float(y)) > 100000 and y_type == 55 ): count = values[i + 9] if isinstance(count, int) and 0 <= count <= 1_000_000: customer_counts.append(count) coordinates.append({"x": float(x), "y": float(y)}) except (IndexError, TypeError, ValueError): continue # De-duplicate accidental overlapping matches while preserving order. dedup_counts: list[int] = [] dedup_coords: list[dict[str, float]] = [] seen_coords: set[tuple[float, float]] = set() for count, coord in zip(customer_counts, coordinates): key = (coord["x"], coord["y"]) if key in seen_coords: continue seen_coords.add(key) dedup_counts.append(count) dedup_coords.append(coord) # NISC serializes the Customers Summary table in this same response. # IMPORTANT: each ZIP summary carries TWO different integer fields: # customers served, then customers currently out. # Older builds loosely matched ", IntegerRef, ZIPRef", which can # accidentally grab the customers-served field (e.g. 5,135) and publish it # as an outage count. Decode the Region records structurally instead. # # Observed GWT layouts: # served, IntegerRef, out, IntegerRef, ZIPRef, RegionNameRef, RegionClassRef # served, IntegerRef, -IntegerRef, ZIPRef, RegionNameRef, RegionClassRef # The negative IntegerRef is GWT's null/back-reference form here and means # no current outage value, so we treat it as zero. String refs are 1-based. integer_ref = next((i + 1 for i, item in enumerate(strings) if isinstance(item, str) and item.startswith("java.lang.Integer/")), None) region_class_ref = next((i + 1 for i, item in enumerate(strings) if isinstance(item, str) and item.startswith("cc.nisc.oms.clientandserver.v2.pojo.Region/")), None) region_name_ref = next((i + 1 for i, item in enumerate(strings) if item == "Zip Code Regions"), None) zip_refs = {i + 1: item for i, item in enumerate(strings) if isinstance(item, str) and re.fullmatch(r"\d{5}", item)} zip_outages: dict[str, int] = {} zip_customers_served: dict[str, int] = {} if integer_ref is not None and zip_refs: for i, value in enumerate(values): if value not in zip_refs: continue if i + 2 >= len(values): continue if region_name_ref is not None and values[i + 1] != region_name_ref: continue if region_class_ref is not None and values[i + 2] != region_class_ref: continue zip_code = zip_refs[value] outage_count = None served_count = None # Explicit outage integer: served, intRef, out, intRef, zipRef if i >= 4 and values[i - 1] == integer_ref: possible_out = values[i - 2] if isinstance(possible_out, int) and possible_out >= 0: outage_count = possible_out if values[i - 3] == integer_ref: possible_served = values[i - 4] if isinstance(possible_served, int) and possible_served >= 0: served_count = possible_served # Null/no-outage form: served, intRef, -intRef, zipRef elif i >= 3 and values[i - 1] == -integer_ref: outage_count = 0 if values[i - 2] == integer_ref: possible_served = values[i - 3] if isinstance(possible_served, int) and possible_served >= 0: served_count = possible_served if outage_count is not None: zip_outages[zip_code] = outage_count if served_count is not None: zip_customers_served[zip_code] = served_count summary_total = sum(zip_outages.values()) if zip_outages else None # Prefer the utility's own Customers Summary total. Keep marker-derived # counts only as a fallback so a future NISC serialization change does not # make the entire outage feed disappear. if summary_total is not None: customers_out = summary_total parse_quality = "customers-summary-by-zip" elif outage_ids and not dedup_counts: customers_out = len(outage_ids) parse_quality = "event-count-fallback" else: customers_out = sum(dedup_counts) parse_quality = "decoded-marker-counts-fallback" return { "customers_out": customers_out, "outage_events": max(len(outage_ids), len(dedup_counts)), "outage_ids": outage_ids, "customer_counts": dedup_counts, "zip_outages": zip_outages, "zip_customers_served": zip_customers_served, "coordinates": dedup_coords, "parse_quality": parse_quality, } def fetch_sheffield() -> dict[str, Any]: request = urllib.request.Request( SHEFFIELD_URL, data=SHEFFIELD_BODY.encode("utf-8"), method="POST", headers={ "Accept": "*/*", "Cache-Control": "no-cache", "Pragma": "no-cache", "Content-Type": "text/x-gwt-rpc; charset=UTF-8", "Origin": SHEFFIELD_ORIGIN, "Referer": SHEFFIELD_REFERER, "User-Agent": "Mozilla/5.0 NWALScannerPlatform/018", "X-GWT-Module-Base": SHEFFIELD_MODULE_BASE, "X-GWT-Permutation": SHEFFIELD_PERMUTATION, "coop.nisc.outagewebmap.configName": "mv", }, ) raw = _fetch_bytes(request, timeout=12) text = raw.decode("utf-8-sig", errors="replace") decoded = _decode_sheffield_response(text) updated = _utc_now_iso() return { "US_Full_FIPS": "01033", "CountyName": "Colbert", "StateAbbr": "AL", "CustomersTracked": 0, "CustomersOut": int(decoded["customers_out"]), "OutageEvents": int(decoded["outage_events"]), "LastUpdatedDateTime": updated, "ProviderName": "Sheffield Utilities", "ProviderArea": "Sheffield Utilities electric service area", "SourceURL": SHEFFIELD_URL, "Stale": False, "AgeSeconds": 0, "ParseQuality": decoded["parse_quality"], "OutageIDs": decoded["outage_ids"], "CustomerCounts": decoded["customer_counts"], "ZipOutages": decoded.get("zip_outages", {}), "ZipCustomersServed": decoded.get("zip_customers_served", {}), } def get_local_outages(force: bool = False) -> dict[str, Any]: global _cache, _cache_time now = time.time() with _cache_lock: if not force and _cache is not None and now - _cache_time < CACHE_SECONDS: return dict(_cache) previous = _cache counties: list[dict[str, Any]] = [] provider_status: list[dict[str, Any]] = [] for provider_name, fetcher in ( ("Florence Electricity", fetch_florence), ("Sheffield Utilities", fetch_sheffield), ): try: county = fetcher() counties.append(county) provider_status.append({ "provider": provider_name, "ok": True, "customersOut": county.get("CustomersOut", 0), "outageEvents": county.get("OutageEvents", 0), "updated": county.get("LastUpdatedDateTime"), "stale": bool(county.get("Stale")), }) except Exception as exc: provider_status.append({ "provider": provider_name, "ok": False, "error": str(exc), }) print(f"[OUTAGES] {provider_name} fetch failed: {exc}", flush=True) # If one source has a temporary hiccup, retain only that provider's last # known record while clearly marking it stale. This prevents the display # from snapping to zero during a brief utility-site outage. if previous and isinstance(previous.get("counties"), list): current_names = {c.get("ProviderName") for c in counties} for old in previous["counties"]: if old.get("ProviderName") not in current_names: fallback = dict(old) fallback["Stale"] = True fallback["Fallback"] = True counties.append(fallback) order = {"01077": 0, "01033": 1} counties.sort(key=lambda c: order.get(str(c.get("US_Full_FIPS", "")), 99)) result = { "success": bool(counties), "source": "local-utility-aggregator", "generatedAt": _utc_now_iso(), "counties": counties, "providers": provider_status, "totalCustomersOut": sum(int(c.get("CustomersOut") or 0) for c in counties), "totalOutageEvents": sum(int(c.get("OutageEvents") or 0) for c in counties), } with _cache_lock: _cache = result _cache_time = time.time() return dict(result)