From 135198e161f4389a7ec0269a6df6a369318cb8cc Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 16 Jun 2026 18:29:37 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8B=86=E5=88=86=E5=9B=BE=E8=A1=A8=E5=AE=9E?= =?UTF-8?q?=E6=97=B6=E8=A7=82=E6=B5=8B=E5=85=9C=E5=BA=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/api/city/[name]/observation/route.ts | 32 ++ .../LiveTemperatureThresholdChart.tsx | 51 +- .../__tests__/refreshCadencePolicy.test.ts | 23 +- .../__tests__/ssePatchArchitecture.test.ts | 14 +- .../__tests__/temperatureChartData.test.ts | 48 ++ .../scan-terminal/temperature-chart-logic.ts | 161 +++++++ frontend/lib/refresh-policy.ts | 2 + tests/test_city_observation_endpoint.py | 114 +++++ web/routers/city.py | 8 + web/services/city_observation.py | 454 ++++++++++++++++++ 10 files changed, 882 insertions(+), 25 deletions(-) create mode 100644 frontend/app/api/city/[name]/observation/route.ts create mode 100644 tests/test_city_observation_endpoint.py create mode 100644 web/services/city_observation.py diff --git a/frontend/app/api/city/[name]/observation/route.ts b/frontend/app/api/city/[name]/observation/route.ts new file mode 100644 index 00000000..1741c79d --- /dev/null +++ b/frontend/app/api/city/[name]/observation/route.ts @@ -0,0 +1,32 @@ +import { NextRequest, NextResponse } from "next/server"; +import { proxyBackendJsonGet } from "@/lib/api-proxy"; +import { NO_STORE_CACHE_CONTROL } from "@/lib/proxy-cache-policy"; + +const API_BASE = process.env.POLYWEATHER_API_BASE_URL; + +export async function GET( + req: NextRequest, + context: { params: Promise<{ name: string }> }, +) { + if (!API_BASE) { + return NextResponse.json( + { error: "POLYWEATHER_API_BASE_URL is not configured" }, + { + headers: { + "Cache-Control": NO_STORE_CACHE_CONTROL, + "Cloudflare-CDN-Cache-Control": NO_STORE_CACHE_CONTROL, + }, + status: 500, + }, + ); + } + + const { name } = await context.params; + return proxyBackendJsonGet(req, { + cacheControl: NO_STORE_CACHE_CONTROL, + fetchCache: "no-store", + includeSupabaseIdentity: false, + publicMessage: "Failed to fetch live city observation", + url: `${API_BASE}/api/city/${encodeURIComponent(name)}/observation`, + }); +} diff --git a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx index 09ab2a9a..d989632f 100644 --- a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx +++ b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx @@ -20,6 +20,7 @@ import { buildIntDegreeTicks, buildRunwayPlates, fetchHourlyForecastForCity, + fetchLiveObservationForCity, getActiveTemperatureSeries, getDebPeakWindowRange, getPeakGlowState, @@ -28,6 +29,7 @@ import { getVisibleTemperatureSeries, isTemperatureSeriesVisibleByDefault, mergeHourlyWithLiveObservations, + mergeObservationPayloadIntoHourly, mergePatchIntoHourly, mergeRowObservationIntoHourly, normObs, @@ -65,7 +67,7 @@ const PEAK_GLOW_BADGE_CLASS = { const PROBABILITY_REFRESH_AFTER_PATCH_MS = DASHBOARD_REFRESH_POLICY_MS.metar; const FOREGROUND_FULL_DETAIL_REFRESH_DEDUP_MS = 90_000; -const NO_PATCH_CACHED_DETAIL_REFRESH_MS = DASHBOARD_REFRESH_POLICY_MS.observation; +const LIVE_OBSERVATION_FALLBACK_MS = DASHBOARD_REFRESH_POLICY_MS.liveObservationFallback; const DETAIL_LOAD_BATCH_DELAY_MS = 0; const TRANSIENT_DETAIL_RETRY_DELAY_MS = 3_000; const INITIAL_DETAIL_LOAD_SLOTS = 3; @@ -252,6 +254,11 @@ function rowObservationTimeForFreshness(row: ScanOpportunityRow | null) { ).trim() || null; } +function observationPayloadTimeForFreshness(payload: any) { + const block = payload?.airport_current || payload?.airport_primary || payload?.current || {}; + return String(block.obs_time || block.observed_at || block.observation_time || payload?.local_time || "").trim() || null; +} + function patchObservationTimeForFreshness(patch: { changes?: Record } | null | undefined) { const changes = patch?.changes || {}; return String( @@ -848,6 +855,28 @@ export function LiveTemperatureThresholdChart({ applySuccessfulHourlyDetail, }); + const applyLiveObservationPayload = useCallback((payload: any) => { + if (!payload || typeof payload !== "object") return; + const appliedAtMs = Date.now(); + const condition = payload.airport_current || payload.airport_primary || payload.current || {}; + const temp = validNumber(condition.temp); + if (temp !== null) setLiveTemp(temp); + if (typeof payload.local_date === "string" && payload.local_date) { + setCurrentCityLocalDate(payload.local_date); + } + commitHourlySnapshot((prev) => + mergeObservationPayloadIntoHourly( + prev ?? seedHourlyForecastFromRow(getLatestRowSnapshot()), + payload, + ), + ); + setChartFreshness((prev) => ({ + ...prev, + rowAppliedAtMs: appliedAtMs, + rowObservationTime: observationPayloadTimeForFreshness(payload), + })); + }, [commitHourlySnapshot, getLatestRowSnapshot]); + useEffect(() => { if (!city || !currentRowObservationSignature) return; if (lastRowObservationSignatureRef.current === currentRowObservationSignature) return; @@ -1036,37 +1065,35 @@ export function LiveTemperatureThresholdChart({ }; }, [resyncVersion, city, runHourlyDetailFetch]); - // ── SSE fallback: visible charts refresh cached detail at observation cadence if patches stop. ── + // ── SSE fallback: visible charts merge no-store observations if patches stop. ── useEffect(() => { if (!shouldPollLiveChart({ city, compact, isActive, isMaximized })) return; let cancelled = false; - const refreshCachedDetail = () => { + const refreshLiveObservation = () => { const now = Date.now(); lastPatchAtRef.current = now; - void runHourlyDetailFetch({ - source: "network", - fetchOptions: { bypassLocalCache: true }, - applyOptions: { updateLiveTemp: true }, - isCancelled: () => cancelled, - onSettled: () => setIsHourlyLoading(false), + void fetchLiveObservationForCity(city).then((payload) => { + if (cancelled || !payload) return; + applyLiveObservationPayload(payload); }); }; const checkFallback = () => { if (typeof document !== "undefined" && document.visibilityState === "hidden") return; - if (Date.now() - lastPatchAtRef.current < NO_PATCH_CACHED_DETAIL_REFRESH_MS) return; + if (Date.now() - lastPatchAtRef.current < LIVE_OBSERVATION_FALLBACK_MS) return; - refreshCachedDetail(); + refreshLiveObservation(); }; + refreshLiveObservation(); const id = setInterval(checkFallback, 60_000); return () => { cancelled = true; clearInterval(id); }; - }, [city, compact, isActive, isMaximized, targetResolution, runHourlyDetailFetch]); + }, [city, compact, isActive, isMaximized, applyLiveObservationPayload]); useEffect(() => { if (!activationRefreshKey) return; diff --git a/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts index c1395cd9..bd33c297 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts @@ -36,6 +36,7 @@ async function flushMicrotasks() { export async function runTests() { assert(DASHBOARD_REFRESH_POLICY_MS.observation === 60_000, "observation layer should refresh every 60 seconds"); + assert(DASHBOARD_REFRESH_POLICY_MS.liveObservationFallback === 3 * 60_000, "chart observation fallback should poll every 180 seconds when SSE is unavailable"); assert(DASHBOARD_REFRESH_POLICY_MS.scanRows === 2 * 60_000, "region/city rows should refresh every 2 minutes"); assert(DASHBOARD_REFRESH_POLICY_MS.marketOverview === 10 * 60_000, "market overview should refresh every 10 minutes"); assert(DASHBOARD_REFRESH_POLICY_MS.model === 30 * 60_000, "DEB and multi-model data should refresh every 30 minutes"); @@ -86,19 +87,21 @@ export async function runTests() { "selected city chart should consume SSE patches and keep METAR cadence for heavy probability refreshes instead of a 2-minute forced refresh", ); assert( - chartSource.includes("NO_PATCH_CACHED_DETAIL_REFRESH_MS = DASHBOARD_REFRESH_POLICY_MS.observation") && - chartSource.includes("refreshCachedDetail") && - chartSource.includes("runHourlyDetailFetch") && - chartSource.includes("fetchOptions: { bypassLocalCache: true }") && + chartSource.includes("LIVE_OBSERVATION_FALLBACK_MS = DASHBOARD_REFRESH_POLICY_MS.liveObservationFallback") && + chartSource.includes("refreshLiveObservation") && + chartSource.includes("fetchLiveObservationForCity") && + chartSource.includes("mergeObservationPayloadIntoHourly") && chartLogicSource.includes("options.bypassLocalCache") && chartLogicSource.includes("const forceRefresh = Boolean(options.ignoreCache)"), - "visible charts should bypass the five-minute browser detail cache every observation cadence without force-refreshing backend sources", + "visible charts should use the no-store observation endpoint for 180-second SSE fallback without forcing detail-batch refreshes", ); const componentHourlyFetchCalls = chartSource.match(/fetchHourlyForecastForCity\(city,/g) || []; + const hourlyFetcherBlock = + /function useHourlyDetailFetcher\([\s\S]*?\n}\r?\n\r?\n\/\/ 岸岸 Main component/.exec(chartSource)?.[0] || ""; assert( chartSource.includes("function useHourlyDetailFetcher") && componentHourlyFetchCalls.length === 1 && - !/fetchHourlyForecastForCity\(city,[\s\S]*?\)\s*\.then\(/.test(chartSource), + !/fetchHourlyForecastForCity\(city,[\s\S]*?\)\s*\.then\(/.test(hourlyFetcherBlock), "temperature chart should centralize full-detail fetch lifecycle in useHourlyDetailFetcher instead of duplicating then/catch branches across effects", ); assert( @@ -159,7 +162,13 @@ export async function runTests() { assert( chartLogicSource.includes("forceRefresh: boolean") && chartLogicSource.includes('force_refresh: forceRefresh ? "true" : "false"'), - "ignoreCache chart detail refreshes must send force_refresh=true so fresh METAR observations bypass proxy and backend chart caches", + "ignoreCache chart detail refreshes must send force_refresh=true only for heavy model/detail resyncs, not the 180-second observation fallback", + ); + assert( + chartLogicSource.includes("/api/city/${encodeURIComponent(city)}/observation") && + chartLogicSource.includes('cache: "no-store"') && + chartLogicSource.includes("mergeObservationPayloadIntoHourly"), + "live observation fetches must call the no-store per-city observation endpoint and merge without touching cached model detail", ); assert( chartLogicSource.includes("cityDetailBatchQueueKey") && diff --git a/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts index 517df2cb..ff4c816b 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts @@ -197,8 +197,8 @@ export function runTests() { "temperature chart must keep METAR cadence for heavy patch-triggered probability refreshes", ); assert( - chart.includes("NO_PATCH_CACHED_DETAIL_REFRESH_MS = DASHBOARD_REFRESH_POLICY_MS.observation"), - "temperature chart must use observation cadence for lightweight cached no-patch refreshes", + chart.includes("LIVE_OBSERVATION_FALLBACK_MS = DASHBOARD_REFRESH_POLICY_MS.liveObservationFallback"), + "temperature chart must use the 180-second observation fallback cadence when SSE patches stop", ); assert(chart.includes("TemperatureChartCanvas"), "temperature chart shell must compose the extracted chart canvas"); assert(chart.includes("TemperatureStatsBars"), "temperature chart shell must compose the extracted stat bars"); @@ -229,13 +229,15 @@ export function runTests() { chart.includes("ignoreCache: true") && chart.includes("currentCityLocalDate !== loadedLocalDate"), "temperature chart must background-refresh full city detail when the city-local day rolls over", ); - const fallbackRefreshBlock = chart.match(/const refreshCachedDetail = \(\) => \{[\s\S]*?\n \};/)?.[0] || ""; + const fallbackRefreshBlock = chart.match(/const refreshLiveObservation = \(\) => \{[\s\S]*?\n \};/)?.[0] || ""; assert( - fallbackRefreshBlock.includes("runHourlyDetailFetch") && - fallbackRefreshBlock.includes("fetchOptions: { bypassLocalCache: true }") && + fallbackRefreshBlock.includes("fetchLiveObservationForCity") && + fallbackRefreshBlock.includes("applyLiveObservationPayload") && + chart.includes("mergeObservationPayloadIntoHourly") && + !fallbackRefreshBlock.includes("runHourlyDetailFetch") && !fallbackRefreshBlock.includes("ignoreCache: true") && !fallbackRefreshBlock.includes("setIsHourlyLoading(true)"), - "no-patch fallback refresh should revalidate through cached backend detail without force-refreshing sources or showing the loading overlay", + "no-patch fallback refresh should merge no-store observation data without refreshing cached detail-batch or showing the loading overlay", ); const resyncBlock = chart.match(/useEffect\(\(\) => \{\s*if \(!resyncVersion \|\| !city\) return;[\s\S]*?\}, \[resyncVersion, city, runHourlyDetailFetch\]\);/)?.[0] || ""; assert( diff --git a/frontend/components/dashboard/scan-terminal/__tests__/temperatureChartData.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/temperatureChartData.test.ts index 013e7b29..05a4847d 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/temperatureChartData.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/temperatureChartData.test.ts @@ -3,6 +3,7 @@ import type { CityDetail } from "@/lib/dashboard-types"; import { buildChartTimeAxis, buildDebBaselinePath } from "@/lib/temperature-chart-paths"; import { buildFullDayChartData, + mergeObservationPayloadIntoHourly, mergeHourlyWithLiveObservations, mergePatchIntoHourly, mergeRowObservationIntoHourly, @@ -835,6 +836,53 @@ export function runTests() { "instant-restore cache must include live-merged runway history so returning to terminal shows it immediately", ); + const observationOnlyPayload = { + city: "chengdu", + local_date: "2026-06-15", + local_time: "17:45", + airport_current: { + temp: 27.1, + obs_time: "2026-06-15T17:45:00+08:00", + source_code: "amsc_awos", + source_label: "AMSC AWOS", + }, + airport_primary: { + temp: 27.1, + obs_time: "2026-06-15T17:45:00+08:00", + source_code: "amsc_awos", + source_label: "AMSC AWOS", + }, + metar_today_obs: [ + { + time: "17:45", + temp: 27.1, + obs_time: "2026-06-15T17:45:00+08:00", + source_code: "amsc_awos", + }, + ], + runway_plate_history: { + "02L/20R": [{ time: "2026-06-15T17:45:00+08:00", tdz_temp: 27.1, end_temp: 26.8 }], + }, + }; + const observationMergedChengdu = mergeObservationPayloadIntoHourly( + cachedChengduDetail, + observationOnlyPayload as any, + ); + assert( + observationMergedChengdu?.airportCurrent?.temp === 27.1 && + observationMergedChengdu?.airportPrimary?.source_code === "amsc_awos", + "observation endpoint payload should update the current observation block", + ); + assert( + observationMergedChengdu?.modelCurves?.ECMWF?.length === 2 && + observationMergedChengdu?.debHourlyPath?.temps?.includes(30.9), + "observation endpoint payload must not clear DEB or multi-model chart detail", + ); + assert( + (observationMergedChengdu?.runwayPlateHistory?.["02L/20R"] || []).length === 2, + "observation endpoint payload should append fresh runway history onto cached detail history", + ); + _hourlyCache.clear(); for (let i = 0; i < 180; i += 1) { const row = { diff --git a/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts b/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts index f62f3d88..dcdc459b 100644 --- a/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts +++ b/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts @@ -1602,6 +1602,16 @@ function mergeHourlyWithLiveObservations( }; } +function mergeObservationPayloadIntoHourly( + prev: HourlyForecast, + payload: CityObservationPayload | null | undefined, +): HourlyForecast { + const live = observationPayloadToHourly(payload); + if (!prev) return live; + if (!live) return prev; + return mergeHourlyWithLiveObservations(prev, live, null); +} + function mergeRowObservationIntoHourly( prev: HourlyForecast, row: ScanOpportunityRow | null, @@ -1643,6 +1653,22 @@ type HourlyForecastFetchOptions = { resolution?: string; }; +type CityObservationPayload = { + city?: string | null; + local_date?: string | null; + local_time?: string | null; + current?: Record | null; + airport_current?: Record | null; + airport_primary?: Record | null; + amos?: AmosData | null; + runway_plate_history?: Record>> | null; + runway_points?: Array>; + metar_today_obs?: Array>; + timeseries?: { + metar_today_obs?: Array>; + } | null; +}; + type CityDetailBatchPayload = { cities?: string[]; details?: Record; @@ -1669,6 +1695,126 @@ const CITY_DETAIL_BATCH_WINDOW_MS = 100; const CITY_DETAIL_BATCH_MAX_CITIES = 12; const _cityDetailBatchQueues = new Map(); +function normalizeObservationCondition(block: Record | null | undefined): AirportCurrentConditions | null { + if (!block || typeof block !== "object") return null; + const temp = validNumber(block.temp); + const obsTime = String(block.obs_time || block.observed_at || block.observation_time || block.time || "").trim(); + if (temp === null || !obsTime) return null; + return { + ...(block as AirportCurrentConditions), + temp, + obs_time: obsTime, + max_so_far: validNumber(block.max_so_far) ?? validNumber(block.max_temp_so_far) ?? temp, + source_code: String(block.source_code || block.source || "").trim() || null, + source_label: String(block.source_label || block.settlement_source_label || block.source || "").trim() || null, + station_code: String(block.station_code || block.icao || "").trim() || null, + station_label: String(block.station_label || block.station_name || "").trim() || null, + }; +} + +function normalizeObservationPoint(point: Record): ObsPoint | null { + const temp = validNumber(point.temp); + const time = String(point.time || point.obs_time || point.observed_at || point.observation_time || "").trim(); + if (temp === null || !time) return null; + return { time, temp }; +} + +function normalizeObservationRunwayHistory( + history: CityObservationPayload["runway_plate_history"], + runwayPoints: CityObservationPayload["runway_points"], +) { + const normalized: Record>> = {}; + Object.entries(history || {}).forEach(([runway, points]) => { + if (!Array.isArray(points)) return; + const normalizedPoints = points + .map((point): Record | null => { + if (!point || typeof point !== "object") return null; + const value = + parseRunwayHistoryValue(point) ?? + validNumber((point as any).target_runway_max) ?? + validNumber((point as any).tdz_temp) ?? + validNumber((point as any).end_temp); + const time = String((point as any).timestamp || (point as any).time || (point as any).observed_at || "").trim(); + if (value === null || !time) return null; + return { + ...point, + timestamp: time, + value, + temp_c: value, + }; + }) + .filter((point): point is Record => point !== null); + if (normalizedPoints.length) normalized[runway] = normalizedPoints; + }); + + for (const point of runwayPoints || []) { + if (!point || typeof point !== "object") continue; + const runway = String((point as any).runway || "").trim().toUpperCase(); + if (!runway) continue; + const value = + parseRunwayHistoryValue(point) ?? + validNumber((point as any).target_runway_max) ?? + validNumber((point as any).tdz_temp) ?? + validNumber((point as any).end_temp); + const time = String((point as any).timestamp || (point as any).time || (point as any).observed_at || "").trim(); + if (value === null || !time) continue; + normalized[runway] = [ + ...(normalized[runway] || []), + { + ...point, + timestamp: time, + value, + temp_c: value, + }, + ].slice(-MAX_OBS_POINTS); + } + + return Object.keys(normalized).length ? normalized : undefined; +} + +function observationPayloadToHourly(payload: CityObservationPayload | null | undefined): HourlyForecast { + if (!payload || typeof payload !== "object") return null; + const airportCurrent = normalizeObservationCondition(payload.airport_current || payload.current); + const airportPrimary = normalizeObservationCondition(payload.airport_primary || payload.airport_current || payload.current); + const current = payload.current && typeof payload.current === "object" + ? { + ...(payload.current as CurrentConditions), + temp: validNumber(payload.current.temp), + } + : null; + const metarTodayObs = [ + ...((payload.timeseries?.metar_today_obs || []) as Array>), + ...((payload.metar_today_obs || []) as Array>), + ] + .map(normalizeObservationPoint) + .filter((point): point is ObsPoint => point !== null); + const airportPrimaryTodayObs = (airportPrimary?.obs_time && validNumber(airportPrimary.temp) !== null) + ? appendRawObservationPoint(undefined, airportPrimary.obs_time, Number(airportPrimary.temp)) + : undefined; + return { + forecastTodayHigh: null, + debPrediction: null, + debQuality: null, + debHourlyPath: null, + localDate: payload.local_date || null, + localTime: payload.local_time || airportPrimary?.obs_time || airportCurrent?.obs_time || null, + times: [], + temps: [], + modelTimes: undefined, + modelCurves: undefined, + forecastDaily: [], + multiModelDaily: {}, + probabilities: null, + runwayPlateHistory: normalizeObservationRunwayHistory(payload.runway_plate_history, payload.runway_points), + amos: payload.amos || null, + current, + airportCurrent, + airportPrimary, + metarTodayObs: metarTodayObs.length ? metarTodayObs : undefined, + airportPrimaryTodayObs, + }; +} + function parseHourlyForecastFromCityDetail(json: CityDetail | null): HourlyForecast { const hourlySource = (json as any)?.hourly ?? (json as any)?.timeseries?.hourly; if (!json || !hourlySource) return null; @@ -1895,6 +2041,19 @@ async function fetchCityDetailBatchWithTimeout( .finally(() => globalThis.clearTimeout(timeoutId)); } +async function fetchLiveObservationForCity(city: string): Promise { + const headers = await buildBrowserBackendHeaders({ Accept: "application/json" }); + return fetch(`/api/city/${encodeURIComponent(city)}/observation`, { + cache: "no-store", + headers, + }) + .then(async (res) => { + if (!res.ok) return null; + return res.json() as Promise; + }) + .catch(() => null); +} + async function fetchHourlyForecastForCity( city: string, options: HourlyForecastFetchOptions = {}, @@ -3223,6 +3382,7 @@ export { buildModelSummaryCards, buildRunwayPlates, fetchHourlyForecastForCity, + fetchLiveObservationForCity, getActiveTemperatureSeries, getTemperatureSeriesForRunwayDetailsMode, getLiveObservationLabels, @@ -3230,6 +3390,7 @@ export { getVisibleTemperatureSeries, isTemperatureSeriesVisibleByDefault, mergeHourlyWithLiveObservations, + mergeObservationPayloadIntoHourly, mergePatchIntoHourly, mergeRowObservationIntoHourly, normObs, diff --git a/frontend/lib/refresh-policy.ts b/frontend/lib/refresh-policy.ts index aaf618b5..e413faac 100644 --- a/frontend/lib/refresh-policy.ts +++ b/frontend/lib/refresh-policy.ts @@ -1,5 +1,6 @@ export const DASHBOARD_REFRESH_POLICY_SEC = { observation: 60, + liveObservationFallback: 3 * 60, metar: 5 * 60, scanRows: 2 * 60, marketOverview: 10 * 60, @@ -8,6 +9,7 @@ export const DASHBOARD_REFRESH_POLICY_SEC = { export const DASHBOARD_REFRESH_POLICY_MS = { observation: DASHBOARD_REFRESH_POLICY_SEC.observation * 1000, + liveObservationFallback: DASHBOARD_REFRESH_POLICY_SEC.liveObservationFallback * 1000, metar: DASHBOARD_REFRESH_POLICY_SEC.metar * 1000, scanRows: DASHBOARD_REFRESH_POLICY_SEC.scanRows * 1000, marketOverview: DASHBOARD_REFRESH_POLICY_SEC.marketOverview * 1000, diff --git a/tests/test_city_observation_endpoint.py b/tests/test_city_observation_endpoint.py new file mode 100644 index 00000000..485582ca --- /dev/null +++ b/tests/test_city_observation_endpoint.py @@ -0,0 +1,114 @@ +from fastapi.testclient import TestClient + +from web.app import app + + +client = TestClient(app) + + +def test_city_observation_endpoint_is_no_store_and_uses_live_observation_service(monkeypatch): + from web.services import city_observation + + calls = [] + + def fake_payload(name): + calls.append(name) + return { + "city": "chengdu", + "local_date": "2026-06-16", + "local_time": "17:45", + "current": { + "temp": 22.2, + "source_code": "amsc_awos", + "obs_time": "2026-06-16T17:45:00+08:00", + }, + "airport_current": { + "temp": 22.2, + "source_code": "amsc_awos", + "obs_time": "2026-06-16T17:45:00+08:00", + }, + "runway_plate_history": { + "02L/20R": [ + { + "time": "2026-06-16T17:45:00+08:00", + "tdz_temp": 22.2, + "end_temp": 22.0, + } + ] + }, + "metar_today_obs": [ + { + "time": "17:45", + "temp": 22.2, + "obs_time": "2026-06-16T17:45:00+08:00", + "source_code": "amsc_awos", + } + ], + } + + monkeypatch.setattr(city_observation, "get_city_observation_payload", fake_payload) + + response = client.get("/api/city/chengdu/observation") + + assert response.status_code == 200 + assert response.headers["Cache-Control"] == "no-store, max-age=0" + assert response.headers["Cloudflare-CDN-Cache-Control"] == "no-store, max-age=0" + assert calls == ["chengdu"] + payload = response.json() + assert payload["airport_current"]["source_code"] == "amsc_awos" + assert payload["runway_plate_history"]["02L/20R"][0]["tdz_temp"] == 22.2 + + +def test_city_observation_payload_prefers_raw_latest_over_stale_city_cache(monkeypatch): + from web.services import city_observation + + class FakeDB: + def get_city_cache(self, kind, city): + raise AssertionError("observation endpoint must not read city_full_cache") + + def get_latest_raw_observation(self, source, city): + if (source, city) != ("amsc_awos", "chengdu"): + return None + return { + "source": "amsc_awos", + "city": "chengdu", + "station_code": "ZUUU", + "station_name": "Chengdu Shuangliu", + "observed_at": "2026-06-16T17:45:00+08:00", + "payload": { + "source": "amsc_awos", + "source_label": "AMSC AWOS Chengdu Shuangliu", + "icao": "ZUUU", + "temp_c": 22.2, + "observation_time": "2026-06-16T17:45:00+08:00", + "runway_obs": { + "point_temperatures": [ + { + "runway": "02L/20R", + "tdz_temp": 22.2, + "end_temp": 22.0, + } + ] + }, + }, + } + + def list_latest_raw_observations_for_city(self, city, *, limit=100): + return [self.get_latest_raw_observation("amsc_awos", city)] + + def get_airport_obs_recent(self, icao, minutes=180): + return [] + + def get_canonical_temperature(self, city): + return None + + monkeypatch.setattr(city_observation, "_CACHE_DB", FakeDB()) + + payload = city_observation.get_city_observation_payload("chengdu") + + assert payload["city"] == "chengdu" + assert payload["airport_current"]["temp"] == 22.2 + assert payload["airport_current"]["source_code"] == "amsc_awos" + assert payload["airport_current"]["obs_time"] == "2026-06-16T17:45:00+08:00" + assert payload["runway_plate_history"]["02L/20R"][0]["tdz_temp"] == 22.2 + assert payload["metar_today_obs"][-1]["source_code"] == "amsc_awos" diff --git a/web/routers/city.py b/web/routers/city.py index fdd21860..d2fc7364 100644 --- a/web/routers/city.py +++ b/web/routers/city.py @@ -16,6 +16,7 @@ from web.services.city_api import ( get_city_summary_payload, list_cities_payload, ) +from web.services import city_observation as city_observation_service from web.services.city_realtime_stream import get_realtime_stream_payload from web.services.request_timing import attach_server_timing_header @@ -153,6 +154,13 @@ async def city_detail_batch( return payload +@router.get("/api/city/{name}/observation") +async def city_observation(response: Response, name: str): + payload = city_observation_service.get_city_observation_payload(name) + apply_no_store(response.headers) + return payload + + @router.get("/api/city/{name}") async def city_detail( request: Request, diff --git a/web/services/city_observation.py b/web/services/city_observation.py new file mode 100644 index 00000000..b7e2440e --- /dev/null +++ b/web/services/city_observation.py @@ -0,0 +1,454 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any, Optional + +from fastapi import HTTPException + +from src.data_collection.city_registry import ALIASES, CITY_REGISTRY +from src.data_collection.city_time import city_local_datetime, get_city_utc_offset_seconds +from src.database.db_manager import DBManager + +_CACHE_DB = DBManager() + +_SOURCE_LABELS = { + "amsc_awos": "AMSC AWOS", + "amos": "AMOS", + "hko_obs": "HKO", + "mgm": "MGM", + "jma_amedas": "JMA", + "cwa": "CWA", + "metar": "METAR", + "madis_hfmetar": "NOAA MADIS", +} + +_PREFERRED_SOURCES = ( + "amsc_awos", + "amos", + "hko_obs", + "mgm", + "jma_amedas", + "cwa", + "madis_hfmetar", + "metar", +) + + +def _normalize_city(value: Any) -> str: + text = str(value or "").strip().lower() + compact = text.replace(" ", "") + city = ALIASES.get(text, ALIASES.get(compact, text)) + if not city or city not in CITY_REGISTRY: + raise HTTPException(status_code=404, detail="Unknown city") + return city + + +def _to_float(value: Any) -> Optional[float]: + try: + if value is None or value == "": + return None + return float(value) + except (TypeError, ValueError): + return None + + +def _first_text(*values: Any) -> str: + for value in values: + text = str(value or "").strip() + if text: + return text + return "" + + +def _parse_datetime(value: Any) -> Optional[datetime]: + if isinstance(value, datetime): + dt = value + else: + text = str(value or "").strip() + if not text: + return None + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + dt = datetime.fromisoformat(text) + except ValueError: + for fmt in ("%Y-%m-%d %H:%M:%S%z", "%Y-%m-%d %H:%M:%S"): + try: + dt = datetime.strptime(text, fmt) + break + except ValueError: + continue + else: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def _epoch(value: Any) -> int: + dt = _parse_datetime(value) + if dt is None: + return 0 + return int(dt.timestamp()) + + +def _row_observed_at(row: dict[str, Any], payload: dict[str, Any]) -> str: + return _first_text( + payload.get("observation_time"), + payload.get("obs_time"), + payload.get("observed_at"), + row.get("observed_at"), + payload.get("observation_time_local"), + payload.get("observed_at_local"), + row.get("fetched_at"), + ) + + +def _row_temp(row: dict[str, Any], payload: dict[str, Any]) -> Optional[float]: + for key in ("temp", "temp_c", "temperature_c", "temperature", "value"): + temp = _to_float(payload.get(key)) + if temp is not None: + return temp + return _to_float(row.get("value")) + + +def _source_code(row: dict[str, Any], payload: dict[str, Any]) -> str: + return str(row.get("source") or payload.get("source") or "").strip().lower() + + +def _station_code(row: dict[str, Any], payload: dict[str, Any]) -> str: + return _first_text( + payload.get("icao"), + payload.get("station_code"), + payload.get("istNo"), + row.get("station_code"), + ).upper() + + +def _station_name(row: dict[str, Any], payload: dict[str, Any], source_label: str) -> str: + return _first_text( + payload.get("station_label"), + payload.get("station_name"), + payload.get("name"), + row.get("station_name"), + source_label, + ) + + +def _source_label(source: str, payload: dict[str, Any]) -> str: + return _first_text(payload.get("source_label"), _SOURCE_LABELS.get(source), source.upper()) + + +def _candidate_score(city: str, row: dict[str, Any], payload: dict[str, Any]) -> tuple[int, int, int]: + meta = CITY_REGISTRY.get(city) or {} + station = _station_code(row, payload) + settlement = str(meta.get("settlement_station_code") or "").strip().upper() + icao = str(meta.get("icao") or "").strip().upper() + station_priority = 0 + if settlement and station == settlement: + station_priority = 2 + elif icao and station == icao: + station_priority = 1 + source = _source_code(row, payload) + try: + source_priority = len(_PREFERRED_SOURCES) - _PREFERRED_SOURCES.index(source) + except ValueError: + source_priority = 0 + observed = _row_observed_at(row, payload) + return station_priority, _epoch(observed), source_priority + + +def _valid_raw_rows(db: Any, city: str) -> list[tuple[tuple[int, int, int], dict[str, Any], dict[str, Any]]]: + rows: list[dict[str, Any]] = [] + lister = getattr(db, "list_latest_raw_observations_for_city", None) + if callable(lister): + try: + listed = lister(city, limit=100) + except Exception: + listed = [] + if isinstance(listed, list): + rows.extend([row for row in listed if isinstance(row, dict)]) + + if not rows: + getter = getattr(db, "get_latest_raw_observation", None) + if callable(getter): + for source in _PREFERRED_SOURCES: + try: + row = getter(source, city) + except Exception: + row = None + if isinstance(row, dict): + rows.append(row) + + candidates: list[tuple[tuple[int, int, int], dict[str, Any], dict[str, Any]]] = [] + for row in rows: + payload = row.get("payload") + if not isinstance(payload, dict): + payload = {} + if _row_temp(row, payload) is None: + continue + candidates.append((_candidate_score(city, row, payload), row, payload)) + return candidates + + +def _airport_log_candidate(db: Any, city: str) -> Optional[dict[str, Any]]: + meta = CITY_REGISTRY.get(city) or {} + station_codes = [ + str(meta.get("settlement_station_code") or "").strip().upper(), + str(meta.get("icao") or "").strip().upper(), + ] + reader = getattr(db, "get_airport_obs_recent", None) + if not callable(reader): + return None + + latest: Optional[tuple[int, dict[str, Any]]] = None + for station_code in [code for index, code in enumerate(station_codes) if code and code not in station_codes[:index]]: + try: + rows = reader(station_code, minutes=180) + except Exception: + rows = [] + for row in rows if isinstance(rows, list) else []: + if not isinstance(row, dict): + continue + temp = _to_float(row.get("temp_c") if row.get("temp_c") is not None else row.get("temp")) + obs_time = _first_text(row.get("obs_time"), row.get("observed_at")) + if temp is None or not obs_time: + continue + score = _epoch(obs_time) + if latest is None or score > latest[0]: + latest = ( + score, + { + "source": "metar", + "source_label": "METAR", + "station_code": station_code, + "station_name": str(meta.get("airport_name") or station_code), + "temp": round(float(temp), 1), + "obs_time": obs_time, + "observed_at": obs_time, + }, + ) + return latest[1] if latest else None + + +def _canonical_candidate(db: Any, city: str) -> Optional[dict[str, Any]]: + getter = getattr(db, "get_canonical_temperature", None) + if not callable(getter): + return None + try: + row = getter(city) + except Exception: + return None + if not isinstance(row, dict): + return None + payload = row.get("payload") if isinstance(row.get("payload"), dict) else row + temp = _to_float(row.get("value") if row.get("value") is not None else payload.get("value")) + if temp is None: + return None + return { + "source": str(row.get("source") or payload.get("source") or "canonical").strip().lower(), + "source_label": _first_text(row.get("source_label"), payload.get("source_label"), "Canonical"), + "station_code": _first_text(row.get("station_code"), payload.get("station_code")), + "station_name": _first_text(row.get("station_name"), payload.get("station_name")), + "temp": round(float(temp), 1), + "obs_time": _first_text( + row.get("observed_at"), + payload.get("observed_at"), + row.get("updated_at"), + payload.get("fetched_at"), + ), + "observed_at": _first_text(row.get("observed_at"), payload.get("observed_at")), + } + + +def _runway_history(row: dict[str, Any], raw_payload: dict[str, Any], obs_time: str) -> dict[str, list[dict[str, Any]]]: + history: dict[str, list[dict[str, Any]]] = {} + runway_obs = raw_payload.get("runway_obs") + points = runway_obs.get("point_temperatures") if isinstance(runway_obs, dict) else None + if isinstance(points, list): + for point in points: + if not isinstance(point, dict): + continue + runway = str(point.get("runway") or row.get("runway") or "").strip().upper() + if not runway: + continue + entry: dict[str, Any] = {"time": obs_time} + for key in ("temp", "tdz_temp", "mid_temp", "end_temp", "target_runway_max"): + temp = _to_float(point.get(key)) + if temp is not None: + entry[key] = round(float(temp), 1) + if len(entry) > 1: + history.setdefault(runway, []).append(entry) + + runway = str(row.get("runway") or "").strip().upper() + value = _to_float(row.get("value")) + if runway and value is not None and runway not in history: + history[runway] = [{"time": obs_time, "temp": round(float(value), 1)}] + return history + + +def _local_context(city: str, obs_time: str) -> tuple[str, str, int]: + dt = _parse_datetime(obs_time) + if dt is None: + local = city_local_datetime(city) + else: + local = city_local_datetime(city, dt.astimezone(timezone.utc)) + return local.date().isoformat(), local.strftime("%H:%M"), get_city_utc_offset_seconds(city, local) + + +def _observation_block( + *, + city: str, + source: str, + source_label: str, + station_code: str, + station_name: str, + temp: float, + obs_time: str, + observed_at_local: str = "", +) -> dict[str, Any]: + freshness = { + "freshness_status": "fresh", + "observed_at": obs_time or None, + "observed_at_local": observed_at_local or None, + "source_code": source, + "source_label": source_label, + } + return { + "city": city, + "temp": round(float(temp), 1), + "source_code": source, + "source_label": source_label, + "settlement_source": source, + "settlement_source_label": source_label, + "station_code": station_code or None, + "station_name": station_name or source_label, + "observed_at": obs_time or None, + "observed_at_local": observed_at_local or None, + "obs_time": obs_time or observed_at_local, + "freshness": freshness, + "observation_status": "live", + } + + +def _payload_from_raw(city: str, row: dict[str, Any], raw_payload: dict[str, Any]) -> dict[str, Any]: + temp = _row_temp(row, raw_payload) + if temp is None: + raise HTTPException(status_code=404, detail="No live observation") + source = _source_code(row, raw_payload) or "raw_observation" + label = _source_label(source, raw_payload) + obs_time = _row_observed_at(row, raw_payload) + observed_at_local = _first_text( + raw_payload.get("observation_time_local"), + raw_payload.get("observed_at_local"), + ) + local_date, local_time, offset_seconds = _local_context(city, obs_time) + current = _observation_block( + city=city, + source=source, + source_label=label, + station_code=_station_code(row, raw_payload), + station_name=_station_name(row, raw_payload, label), + temp=temp, + obs_time=obs_time, + observed_at_local=observed_at_local, + ) + runway_history = _runway_history(row, raw_payload, obs_time) + metar_point = { + "time": local_time, + "temp": round(float(temp), 1), + "obs_time": obs_time, + "source_code": source, + "source_label": label, + } + return { + "city": city, + "name": city, + "display_name": str((CITY_REGISTRY.get(city) or {}).get("name") or city), + "source": "live_observation", + "local_date": local_date, + "local_time": local_time, + "utc_offset_seconds": offset_seconds, + "updated_at": obs_time or row.get("fetched_at"), + "current": current, + "airport_current": dict(current), + "airport_primary": dict(current), + "amos": dict(raw_payload), + "raw_observation": { + "source": source, + "station_code": _station_code(row, raw_payload) or None, + "observed_at": obs_time or None, + "fetched_at": row.get("fetched_at") or None, + }, + "runway_plate_history": runway_history, + "runway_points": [ + {**point, "runway": runway} + for runway, points in runway_history.items() + for point in points + ], + "metar_today_obs": [metar_point], + "timeseries": {"metar_today_obs": [metar_point]}, + "observation_status": "live", + } + + +def _payload_from_block(city: str, block: dict[str, Any]) -> dict[str, Any]: + temp = _to_float(block.get("temp")) + if temp is None: + raise HTTPException(status_code=404, detail="No live observation") + obs_time = _first_text(block.get("obs_time"), block.get("observed_at")) + local_date, local_time, offset_seconds = _local_context(city, obs_time) + source = str(block.get("source") or block.get("source_code") or "observation").strip().lower() + label = _first_text(block.get("source_label"), _SOURCE_LABELS.get(source), source.upper()) + current = _observation_block( + city=city, + source=source, + source_label=label, + station_code=str(block.get("station_code") or "").strip().upper(), + station_name=str(block.get("station_name") or label).strip(), + temp=temp, + obs_time=obs_time, + ) + metar_point = { + "time": local_time, + "temp": round(float(temp), 1), + "obs_time": obs_time, + "source_code": source, + "source_label": label, + } + return { + "city": city, + "name": city, + "display_name": str((CITY_REGISTRY.get(city) or {}).get("name") or city), + "source": "live_observation", + "local_date": local_date, + "local_time": local_time, + "utc_offset_seconds": offset_seconds, + "updated_at": obs_time, + "current": current, + "airport_current": dict(current), + "airport_primary": dict(current), + "runway_plate_history": {}, + "runway_points": [], + "metar_today_obs": [metar_point], + "timeseries": {"metar_today_obs": [metar_point]}, + "observation_status": "live", + } + + +def get_city_observation_payload(name: str) -> dict[str, Any]: + city = _normalize_city(name) + candidates = _valid_raw_rows(_CACHE_DB, city) + if candidates: + _, row, raw_payload = max(candidates, key=lambda item: item[0]) + return _payload_from_raw(city, row, raw_payload) + + airport_log = _airport_log_candidate(_CACHE_DB, city) + if airport_log: + return _payload_from_block(city, airport_log) + + canonical = _canonical_candidate(_CACHE_DB, city) + if canonical: + return _payload_from_block(city, canonical) + + raise HTTPException(status_code=404, detail="No live observation")