gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve scan terminal freshness and AMSC overlay consistency

Browse Source
This commit is contained in:
2569718930@qq.com
2026-06-15 01:39:19 +08:00
parent 717786b1d8
commit c79e0f36cc
30 changed files with 924 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/cache_warmer_service.py
+10 -10
View File
@@ -190,17 +190,17 @@ class CacheWarmer:
city_provider: Callable[[], Mapping[str, Mapping[str, Any]]],
scan_warmer: Callable[..., Any],
city_panel_warmer: Callable[..., Any],
scan_interval_sec: int = 300,
city_interval_sec: int = 60,
city_batch_size: int = 8,
scan_interval_sec: int = 120,
city_interval_sec: int = 30,
city_batch_size: int = 16,
hot_cities: Optional[Iterable[str]] = None,
) -> None:
self.city_provider = city_provider
self.scan_warmer = scan_warmer
self.city_panel_warmer = city_panel_warmer
self.scan_interval_sec = max(60, int(scan_interval_sec or 300))
self.city_interval_sec = max(30, int(city_interval_sec or 60))
self.city_batch_size = max(1, min(32, int(city_batch_size or 8)))
self.scan_interval_sec = max(60, int(scan_interval_sec or 120))
self.city_interval_sec = max(30, int(city_interval_sec or 30))
self.city_batch_size = max(1, min(32, int(city_batch_size or 16)))
self.hot_cities = tuple(hot_cities or DEFAULT_HOT_CITIES)
self._last_scan_ts = 0.0
self._last_city_ts = 0.0
@@ -279,9 +279,9 @@ def build_default_cache_warmer() -> CacheWarmer:
city_provider=lambda: CITIES,
scan_warmer=build_scan_terminal_payload,
city_panel_warmer=_queue_city_panel_refresh,
scan_interval_sec=_env_int("POLYWEATHER_WARMER_SCAN_INTERVAL_SEC", 300),
city_interval_sec=_env_int("POLYWEATHER_WARMER_CITY_INTERVAL_SEC", 60),
city_batch_size=_env_int("POLYWEATHER_WARMER_CITY_BATCH_SIZE", 8),
scan_interval_sec=_env_int("POLYWEATHER_WARMER_SCAN_INTERVAL_SEC", 120),
city_interval_sec=_env_int("POLYWEATHER_WARMER_CITY_INTERVAL_SEC", 30),
city_batch_size=_env_int("POLYWEATHER_WARMER_CITY_BATCH_SIZE", 16),
hot_cities=hot_cities or DEFAULT_HOT_CITIES,
)
@@ -291,4 +291,4 @@ def warmer_enabled() -> bool:
def warmer_tick_sec() -> int:
return max(10, _env_int("POLYWEATHER_WARMER_TICK_SEC", 60))
return max(10, _env_int("POLYWEATHER_WARMER_TICK_SEC", 30))
web/observation_collector_service.py
+1 -1
View File
@@ -96,7 +96,7 @@ class ObservationCollector:
int(
cache_refresh_workers
if cache_refresh_workers is not None
else _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS", 1)
else _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS", 2)
),
),
)
web/realtime_patch_schema.py
+17
View File
@@ -16,6 +16,15 @@ from src.data_collection.city_time import (
SCHEMA_TYPE = "city_observation_patch"
SCHEMA_VERSION = 1
EVENT_TYPE = "city_observation_patch.v1"
DERIVED_SCAN_NUMBER_FIELDS = (
"signed_gap",
"gap_to_target",
"touch_distance",
"current_reference",
"edge",
"edge_percent",
"deb_prediction",
)
SOURCE_CADENCE_SECONDS = {
"amos": 60,
"amsc_awos": 180,
@@ -178,6 +187,10 @@ def _payload_from_legacy(changes: Dict[str, Any]) -> Dict[str, Any]:
max_so_far = _first_number(changes.get("max_so_far"), changes.get("current_max_so_far"))
if max_so_far is not None:
payload["max_so_far"] = max_so_far
for key in DERIVED_SCAN_NUMBER_FIELDS:
value = _finite_number(changes.get(key))
if value is not None:
payload[key] = value
station_code = str(
changes.get("station_code")
@@ -227,6 +240,10 @@ def _payload_from_v1(raw_payload: Any) -> Dict[str, Any]:
max_so_far = _finite_number(raw_payload.get("max_so_far"))
if max_so_far is not None:
payload["max_so_far"] = max_so_far
for key in DERIVED_SCAN_NUMBER_FIELDS:
value = _finite_number(raw_payload.get(key))
if value is not None:
payload[key] = value
for key in ("station_code", "station_label", "series_key", "unit"):
value = raw_payload.get(key)
web/routers/scan.py
+4
View File
@@ -30,6 +30,8 @@ async def scan_terminal(
time_range: str = "today",
limit: int = 25,
force_refresh: bool = False,
diff: bool = False,
since_snapshot_id: str | None = None,
region: str = "",
trading_region: str = "",
timezone_offset_seconds: int | None = None,
@@ -46,6 +48,8 @@ async def scan_terminal(
time_range=time_range,
limit=limit,
force_refresh=force_refresh,
diff=diff,
since_snapshot_id=since_snapshot_id,
region=region or trading_region or None,
timezone_offset_seconds=timezone_offset_seconds,
)
web/scan_terminal_cache.py
+8
View File
@@ -154,11 +154,19 @@ def set_cached_scan_terminal_payload(
cache_key = scan_terminal_cache_key(filters)
existing = get_scan_terminal_cache_entry(filters) or {}
now = time.time()
existing_success_payload = existing.get("success_payload")
previous_success_payload = existing.get("previous_success_payload")
if isinstance(existing_success_payload, dict):
existing_snapshot_id = existing_success_payload.get("snapshot_id")
next_snapshot_id = payload.get("snapshot_id")
if existing_snapshot_id and existing_snapshot_id != next_snapshot_id:
previous_success_payload = dict(existing_success_payload)
entry = {
"t": now,
"payload": dict(payload),
"success_t": now,
"success_payload": dict(payload),
"previous_success_payload": previous_success_payload,
"last_error": existing.get("last_error"),
"last_failed_at": existing.get("last_failed_at"),
}
web/scan_terminal_payloads.py
+129
View File
@@ -81,6 +81,135 @@ def build_scan_terminal_snapshot_id(
return f"scan-{digest[:10]}"
def _scan_row_id(row: Any) -> Optional[str]:
if not isinstance(row, dict):
return None
row_id = row.get("id")
if row_id is None:
return None
text = str(row_id)
return text if text else None
def _scan_row_digest(row: Dict[str, Any]) -> str:
return hashlib.sha256(
json.dumps(row, ensure_ascii=True, sort_keys=True, default=str).encode("utf-8")
).hexdigest()
def _rows_by_id(payload: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
rows = payload.get("rows")
if not isinstance(rows, list):
return {}
indexed: Dict[str, Dict[str, Any]] = {}
for row in rows:
if not isinstance(row, dict):
continue
row_id = _scan_row_id(row)
if row_id is not None:
indexed[row_id] = row
return indexed
def _full_scan_terminal_incremental_payload(
*,
current_payload: Dict[str, Any],
since_snapshot_id: str,
) -> Dict[str, Any]:
payload = dict(current_payload)
payload["diff"] = {
"mode": "full",
"base_snapshot_id": since_snapshot_id,
"snapshot_id": current_payload.get("snapshot_id"),
"rows_changed": [],
"removed_row_ids": [],
}
return payload
def build_scan_terminal_incremental_payload(
*,
filters: Dict[str, Any],
current_payload: Dict[str, Any],
since_snapshot_id: Optional[str],
base_payload: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
since_snapshot_id = str(since_snapshot_id or "").strip()
current_snapshot_id = current_payload.get("snapshot_id")
if not since_snapshot_id or not current_snapshot_id:
return dict(current_payload)
if current_payload.get("stale") is True or current_payload.get("status") not in (
"ready",
"partial",
):
return _full_scan_terminal_incremental_payload(
current_payload=current_payload,
since_snapshot_id=since_snapshot_id,
)
if since_snapshot_id == current_snapshot_id:
return {
"generated_at": current_payload.get("generated_at"),
"snapshot_id": current_snapshot_id,
"status": "not_modified",
"stale": False,
"stale_reason": None,
"last_success_at": current_payload.get("last_success_at"),
"last_failed_at": current_payload.get("last_failed_at"),
"filters": filters,
"summary": current_payload.get("summary") or {},
"top_signal": current_payload.get("top_signal"),
"rows": [],
"diff": {
"mode": "not_modified",
"base_snapshot_id": since_snapshot_id,
"snapshot_id": current_snapshot_id,
"rows_changed": [],
"removed_row_ids": [],
},
}
if (
not isinstance(base_payload, dict)
or base_payload.get("snapshot_id") != since_snapshot_id
):
return _full_scan_terminal_incremental_payload(
current_payload=current_payload,
since_snapshot_id=since_snapshot_id,
)
current_rows = _rows_by_id(current_payload)
base_rows = _rows_by_id(base_payload)
rows_changed: List[Dict[str, Any]] = []
for row_id, row in current_rows.items():
base_row = base_rows.get(row_id)
if base_row is None or _scan_row_digest(base_row) != _scan_row_digest(row):
rows_changed.append(row)
removed_row_ids = [
row_id for row_id in base_rows.keys() if row_id not in current_rows
]
return {
"generated_at": current_payload.get("generated_at"),
"snapshot_id": current_snapshot_id,
"status": current_payload.get("status") or "ready",
"stale": False,
"stale_reason": current_payload.get("stale_reason"),
"last_success_at": current_payload.get("last_success_at"),
"last_failed_at": current_payload.get("last_failed_at"),
"filters": filters,
"summary": current_payload.get("summary") or {},
"top_signal": current_payload.get("top_signal"),
"rows": [],
"diff": {
"mode": "row_delta",
"base_snapshot_id": since_snapshot_id,
"snapshot_id": current_snapshot_id,
"rows_changed": rows_changed,
"removed_row_ids": removed_row_ids,
},
}
def build_stale_scan_terminal_payload(
*,
filters: Dict[str, Any],
web/scan_terminal_service.py
+69 -3
View File
@@ -35,6 +35,7 @@ from web.scan_terminal_filters import (
)
from web.scan_terminal_payloads import (
build_failed_scan_terminal_payload,
build_scan_terminal_incremental_payload,
build_scan_terminal_snapshot_id,
build_stale_scan_terminal_payload,
compact_ranked_scan_rows_for_payload,
@@ -75,6 +76,36 @@ def _success_payload_within_stale_window(cached_entry: Dict[str, Any]) -> bool:
return (time.time() - success_ts) <= float(SCAN_TERMINAL_STALE_SUCCESS_MAX_AGE_SEC)
def _build_scan_terminal_incremental_from_cache_entry(
*,
filters: Dict[str, Any],
current_payload: Dict[str, Any],
cached_entry: Optional[Dict[str, Any]],
since_snapshot_id: Optional[str],
) -> Dict[str, Any]:
base_payload: Optional[Dict[str, Any]] = None
since_snapshot_id = str(since_snapshot_id or "").strip()
candidates: List[Any] = [current_payload]
if isinstance(cached_entry, dict):
candidates.extend(
[
cached_entry.get("success_payload"),
cached_entry.get("previous_success_payload"),
cached_entry.get("payload"),
]
)
for candidate in candidates:
if isinstance(candidate, dict) and candidate.get("snapshot_id") == since_snapshot_id:
base_payload = candidate
break
return build_scan_terminal_incremental_payload(
filters=filters,
current_payload=current_payload,
since_snapshot_id=since_snapshot_id,
base_payload=base_payload,
)
def _build_stale_payload_for_timeout_if_better_cached(
*,
filters: Dict[str, Any],
@@ -339,6 +370,8 @@ def build_scan_terminal_payload(
raw_filters: Optional[Dict[str, Any]] = None,
*,
force_refresh: bool = False,
diff: bool = False,
since_snapshot_id: Optional[str] = None,
timing_recorder: Any = None,
) -> Dict[str, Any]:
filters = (
@@ -364,6 +397,14 @@ def build_scan_terminal_payload(
)
)
if cached is not None:
if diff and since_snapshot_id:
cached_entry = get_scan_terminal_cache_entry(filters) or {}
return _build_scan_terminal_incremental_from_cache_entry(
filters=filters,
current_payload=cached,
cached_entry=cached_entry,
since_snapshot_id=since_snapshot_id,
)
return cached
cached_entry = (
@@ -381,7 +422,7 @@ def build_scan_terminal_payload(
and _success_payload_within_stale_window(cached_entry)
):
started = _start_scan_terminal_background_refresh(filters)
return build_stale_scan_terminal_payload(
payload = build_stale_scan_terminal_payload(
filters=filters,
success_payload=success_payload,
error_message=(
@@ -389,8 +430,16 @@ def build_scan_terminal_payload(
),
failed_at=cached_entry.get("last_failed_at"),
)
if diff and since_snapshot_id:
return _build_scan_terminal_incremental_from_cache_entry(
filters=filters,
current_payload=payload,
cached_entry=cached_entry,
since_snapshot_id=since_snapshot_id,
)
return payload
started = _start_scan_terminal_background_refresh(filters)
return build_failed_scan_terminal_payload(
payload = build_failed_scan_terminal_payload(
filters=filters,
error_message=(
"市场扫描快照正在初始化"
@@ -399,8 +448,16 @@ def build_scan_terminal_payload(
),
failed_at=cached_entry.get("last_failed_at"),
)
if diff and since_snapshot_id:
return _build_scan_terminal_incremental_from_cache_entry(
filters=filters,
current_payload=payload,
cached_entry=cached_entry,
since_snapshot_id=since_snapshot_id,
)
return payload
return (
payload = (
timing_recorder.measure(
"uncached_build",
lambda: _build_scan_terminal_payload_singleflight(
@@ -411,6 +468,15 @@ def build_scan_terminal_payload(
if timing_recorder is not None
else _build_scan_terminal_payload_singleflight(filters, force_refresh=force_refresh)
)
if diff and since_snapshot_id:
cached_entry = get_scan_terminal_cache_entry(filters) or {}
return _build_scan_terminal_incremental_from_cache_entry(
filters=filters,
current_payload=payload,
cached_entry=cached_entry,
since_snapshot_id=since_snapshot_id,
)
return payload
_SCAN_PREWARM_STARTED = False
web/services/canonical_engine.py
+12 -1
View File
@@ -185,7 +185,18 @@ def build_realtime_event_from_canonical(canonical: dict[str, Any]) -> Optional[d
text = str(canonical.get(key) or "").strip()
if text:
payload[key] = text
for key in ("freshness_sec", "confidence"):
for key in (
"freshness_sec",
"confidence",
"max_so_far",
"signed_gap",
"gap_to_target",
"touch_distance",
"current_reference",
"edge",
"edge_percent",
"deb_prediction",
):
value_meta = canonical.get(key)
if value_meta is not None and value_meta != "":
payload[key] = value_meta
web/services/canonical_temperature.py
+15
View File
@@ -28,6 +28,14 @@ def _to_int(value: Any) -> Optional[int]:
return None
def _first_float(*values: Any) -> Optional[float]:
for value in values:
number = _to_float(value)
if number is not None:
return number
return None
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
@@ -127,6 +135,7 @@ def build_canonical_temperature(
canonical = {
"city": str(city or payload.get("name") or payload.get("city") or "").strip().lower(),
"value": round(value, 2),
"max_so_far": _first_float(current.get("max_so_far"), current.get("max_temp_so_far")),
"temp_symbol": str(payload.get("temp_symbol") or "°C"),
"source": source,
"source_label": source_label,
@@ -142,6 +151,10 @@ def build_canonical_temperature(
}
age_text = f" updated {freshness_sec}s ago" if freshness_sec is not None else ""
canonical["explanation"] = f"{source_label or source or 'Source'}{age_text}."
deb = payload.get("deb") if isinstance(payload.get("deb"), dict) else {}
deb_prediction = _to_float(deb.get("prediction"))
if deb_prediction is not None:
canonical["deb_prediction"] = deb_prediction
return canonical
@@ -190,6 +203,7 @@ def build_city_weather_from_canonical(city: str, canonical: Dict[str, Any]) -> O
}
current = {
"temp": value,
"max_so_far": _first_float(canonical.get("max_so_far"), value),
"source_code": source,
"settlement_source": source,
"settlement_source_label": source_label,
@@ -201,6 +215,7 @@ def build_city_weather_from_canonical(city: str, canonical: Dict[str, Any]) -> O
}
airport_primary = {
"temp": value,
"max_so_far": _first_float(canonical.get("max_so_far"), value),
"source_code": source,
"source_label": source_label,
"obs_time": observed_at or observed_at_local,
web/services/latest_observation_overlay.py
+8 -1
View File
@@ -45,12 +45,19 @@ def _payload_latest_epoch(payload: dict[str, Any], keys: tuple[str, ...]) -> Opt
def _block_epoch(block: Any) -> Optional[int]:
if not isinstance(block, dict):
return None
return _payload_latest_epoch(
canonical_epoch = _payload_latest_epoch(
block,
(
"observed_at",
"observation_time",
"obs_time",
),
)
if canonical_epoch is not None:
return canonical_epoch
return _payload_latest_epoch(
block,
(
"observed_at_local",
"observation_time_local",
),
web/services/scan_api.py
+17
View File
@@ -23,6 +23,17 @@ def _supports_timing_recorder(func: Any) -> bool:
)
def _supports_keyword(func: Any, name: str) -> bool:
try:
params = signature(func).parameters.values()
except (TypeError, ValueError):
return True
return any(
param.name == name or param.kind == Parameter.VAR_KEYWORD
for param in params
)
async def get_scan_terminal_payload(
request: Request,
*,
@@ -36,6 +47,8 @@ async def get_scan_terminal_payload(
time_range: str = "today",
limit: int = 25,
force_refresh: bool = False,
diff: bool = False,
since_snapshot_id: str | None = None,
region: str = "",
timezone_offset_seconds: int | None = None,
) -> Dict[str, Any]:
@@ -67,6 +80,10 @@ async def get_scan_terminal_payload(
async def build_payload():
builder = legacy_routes.build_scan_terminal_payload
kwargs: Dict[str, Any] = {"force_refresh": force_refresh}
if _supports_keyword(builder, "diff"):
kwargs["diff"] = diff
if _supports_keyword(builder, "since_snapshot_id"):
kwargs["since_snapshot_id"] = since_snapshot_id
if _supports_timing_recorder(builder):
kwargs["timing_recorder"] = timer
return await run_in_threadpool(builder, filters, **kwargs)