From 9978ac3e010797941bc2f543ab83cefd3e875867 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 25 May 2026 06:43:42 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=AE=9E=E6=97=B6=E6=BB=9A?= =?UTF-8?q?=E5=8A=A8=E6=B8=A9=E5=BA=A6=E8=B5=B0=E5=8A=BF=E5=9B=BE=EF=BC=9A?= =?UTF-8?q?API=E7=AB=AF=E7=82=B9+=E5=89=8D=E7=AB=AF=E7=BB=84=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端 city_realtime_stream.py：循环缓冲区(deque maxlen=1440)，best_temp() METAR优先。路由 /api/city/{name}/realtime-stream 返回 {points, thresholds}。前端 RealtimeScrollChart：每30秒轮询，一条温度线+多条阈值横线，横轴随时间推进。 --- .../LiveTemperatureThresholdChart.tsx | 16 +- .../scan-terminal/RealtimeScrollChart.tsx | 160 ++++++++++++++++++ web/routers/city.py | 8 + web/services/city_realtime_stream.py | 118 +++++++++++++ 4 files changed, 299 insertions(+), 3 deletions(-) create mode 100644 frontend/components/dashboard/scan-terminal/RealtimeScrollChart.tsx create mode 100644 web/services/city_realtime_stream.py diff --git a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx index 0e4fc885..d66552fc 100644 --- a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx +++ b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx @@ -341,12 +341,20 @@ export function LiveTemperatureThresholdChart({ isEn: boolean; row: ScanOpportunityRow | null; }) { +const hourlyCache = new Map(); +const HOURLY_CACHE_TTL_MS = 30 * 60 * 1000; // 30 min + const [hourly, setHourly] = useState(null); const city = String(row?.city || "").toLowerCase().trim(); useEffect(() => { - setHourly(null); if (!city) return; + const cached = hourlyCache.get(city); + if (cached && Date.now() - cached.ts < HOURLY_CACHE_TTL_MS) { + setHourly(cached.data); + return; + } + setHourly(null); let cancelled = false; fetch(`/api/city/${encodeURIComponent(city)}/detail?depth=panel&force_refresh=false`, { cache: "no-store", @@ -358,13 +366,15 @@ export function LiveTemperatureThresholdChart({ }) .then((json) => { if (cancelled || !json?.hourly) return; - setHourly({ + const data: HourlyForecast = { forecastTodayHigh: json.forecast?.today_high ?? null, localTime: json.local_time || null, times: json.hourly.times || [], temps: json.hourly.temps || [], modelCurves: json.models_hourly?.curves || undefined, - }); + }; + hourlyCache.set(city, { ts: Date.now(), data }); + setHourly(data); }) .catch(() => {}); return () => { cancelled = true; }; diff --git a/frontend/components/dashboard/scan-terminal/RealtimeScrollChart.tsx b/frontend/components/dashboard/scan-terminal/RealtimeScrollChart.tsx new file mode 100644 index 00000000..1498fc16 --- /dev/null +++ b/frontend/components/dashboard/scan-terminal/RealtimeScrollChart.tsx @@ -0,0 +1,160 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { + CartesianGrid, + Line, + LineChart as ReLineChart, + ReferenceLine, + ResponsiveContainer, + Tooltip, + XAxis, + YAxis, +} from "recharts"; +import { Panel } from "@/components/dashboard/scan-terminal/Panel"; + +type StreamPoint = { timestamp: string; temp: number; source: string }; +type Threshold = { label: string; threshold_c: number; breached: boolean }; + +type StreamPayload = { + points: StreamPoint[]; + thresholds: Threshold[]; +}; + +const POLL_INTERVAL_MS = 30_000; + +export function RealtimeScrollChart({ + city, + isEn, +}: { + city: string; + isEn: boolean; +}) { + const [payload, setPayload] = useState({ points: [], thresholds: [] }); + + useEffect(() => { + if (!city) return; + let cancelled = false; + + const fetchStream = () => { + fetch(`/api/city/${encodeURIComponent(city)}/realtime-stream`, { + cache: "no-store", + headers: { Accept: "application/json" }, + }) + .then(async (res) => { + if (!res.ok) return null; + return res.json() as Promise; + }) + .then((data) => { + if (cancelled || !data) return; + setPayload(data); + }) + .catch(() => {}); + }; + + fetchStream(); + const interval = setInterval(fetchStream, POLL_INTERVAL_MS); + return () => { + cancelled = true; + clearInterval(interval); + }; + }, [city]); + + const { points, thresholds } = payload; + const latestTemp = points.length ? points[points.length - 1].temp : null; + const breached = thresholds.filter((t) => t.breached); + const domainMin = thresholds.length + ? Math.min(...thresholds.map((t) => t.threshold_c)) - 2 + : "auto"; + const domainMax = thresholds.length + ? Math.max(...thresholds.map((t) => t.threshold_c)) + 2 + : "auto"; + + return ( + +
+ {/* Status bar */} +
+ + {isEn ? "Latest" : "最新"}:{" "} + + {latestTemp !== null ? `${latestTemp.toFixed(1)}°` : "--"} + + + + {isEn ? "Points" : "数据点"}: {points.length} + + {breached.length > 0 && ( + + {isEn ? "Breached" : "已触发"}: {breached.map((t) => t.label).join(", ")} + + )} +
+ + {/* Chart */} +
+ {points.length < 2 ? ( +
+ {isEn ? "Collecting data..." : "数据采集中..."} +
+ ) : ( + + + + + `${Number(v).toFixed(1)}°`} + axisLine={{ stroke: "#cbd5e1" }} + tickLine={false} + domain={[domainMin, domainMax]} + width={40} + /> + [`${Number(value).toFixed(2)}°`, "Temp"]} + labelFormatter={(label) => `${label}`} + /> + {/* Temperature line */} + + {/* Threshold lines */} + {thresholds.map((t) => ( + + ))} + + + )} +
+
+ + ); +} diff --git a/web/routers/city.py b/web/routers/city.py index c497926a..d876ff7d 100644 --- a/web/routers/city.py +++ b/web/routers/city.py @@ -11,6 +11,7 @@ from web.services.city_api import ( get_city_summary_payload, list_cities_payload, ) +from web.services.city_realtime_stream import get_realtime_stream_payload router = APIRouter(tags=["city"]) @@ -208,3 +209,10 @@ async def city_holders( "available": True, "condition_id": condition_id, } + + +@router.get("/api/city/{name}/realtime-stream") +async def city_realtime_stream(name: str): + """Return a rolling window of recent temperature readings + market + threshold lines for the scrolling realtime chart.""" + return get_realtime_stream_payload(name) diff --git a/web/services/city_realtime_stream.py b/web/services/city_realtime_stream.py new file mode 100644 index 00000000..fd0c09e2 --- /dev/null +++ b/web/services/city_realtime_stream.py @@ -0,0 +1,118 @@ +"""Lightweight realtime temperature stream for scrolling chart. + +Maintains per-city deque buffers (max 1440 points) fed by _analyze() +refreshes. The /api/city/{name}/realtime-stream endpoint reads from +these buffers and returns a simple {points, thresholds} payload that +the frontend RealtimeScrollChart polls every 30 seconds. +""" + +from __future__ import annotations + +import collections +import threading +import time +from typing import Any, Dict, List, Optional + +from web.analysis_service import _analyze +from web.core import CITIES + +# Per-city ring buffers: city_name → deque of {timestamp, temp, source} +_STREAM_BUFFERS: Dict[str, collections.deque] = {} +_BUFFER_LOCK = threading.Lock() +_MAXLEN = 1440 + + +def _best_temp(data: Dict[str, Any]) -> Optional[float]: + """METAR-first, then runway sensor, then settlement current.""" + airport = data.get("airport_current") or {} + t = airport.get("current", {}).get("temp") if isinstance(airport, dict) else None + if t is not None: + return float(t) + # AMOS / runway sensor + amos = data.get("amos") or {} + if isinstance(amos, dict): + runway_obs = amos.get("runway_obs") or {} + temps = runway_obs.get("temperatures") if isinstance(runway_obs, dict) else [] + if isinstance(temps, list) and temps: + for pair in temps: + vals = pair if isinstance(pair, list) else [] + for v in vals: + try: + if v is not None: + return float(v) + except (TypeError, ValueError): + continue + # Settlement source + curr = data.get("current") or {} + if isinstance(curr, dict) and curr.get("temp") is not None: + return float(curr["temp"]) + return None + + +def _extract_thresholds(data: Dict[str, Any]) -> List[Dict[str, Any]]: + """Extract threshold lines from market data.""" + thresholds: List[Dict[str, Any]] = [] + dist = (data.get("probabilities") or {}).get("distribution") or [] + current_temp = _best_temp(data) + + for bucket in dist: + if not isinstance(bucket, dict): + continue + temp_val = bucket.get("temp") + if temp_val is None: + continue + try: + t = float(temp_val) + except (TypeError, ValueError): + continue + label = str(bucket.get("label") or f"{t}°C") + thresholds.append({ + "label": label, + "threshold_c": t, + "breached": current_temp is not None and current_temp >= t, + }) + + # Sort by temperature ascending + thresholds.sort(key=lambda x: float(x["threshold_c"])) + return thresholds + + +def capture_sample(city: str) -> None: + """Record one sample for *city* into its ring buffer.""" + try: + data = _analyze(city, force_refresh=False, detail_mode="panel") + except Exception: + return + + temp = _best_temp(data) + if temp is None: + return + + ts = time.strftime("%H:%M:%S") + point = {"timestamp": ts, "temp": round(temp, 1), "source": "metar"} + + with _BUFFER_LOCK: + buf = _STREAM_BUFFERS.get(city) + if buf is None: + buf = collections.deque(maxlen=_MAXLEN) + _STREAM_BUFFERS[city] = buf + buf.append(point) + + +def get_realtime_stream_payload(city: str) -> Dict[str, Any]: + """Return {points, thresholds} for the scrolling chart.""" + # Capture a fresh sample + capture_sample(city) + + with _BUFFER_LOCK: + buf = _STREAM_BUFFERS.get(city) + points = list(buf) if buf else [] + + # Build thresholds from cached analysis + try: + data = _analyze(city, force_refresh=False, detail_mode="panel") + thresholds = _extract_thresholds(data) + except Exception: + thresholds = [] + + return {"points": points, "thresholds": thresholds}