From bbd7c768f86a27f1efb3e7338232f5e6db123945 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Wed, 27 May 2026 08:42:08 +0800 Subject: [PATCH] feat: implement live temperature threshold charting component with SSE patch support and data collection logic --- .../LiveTemperatureThresholdChart.tsx | 2 +- .../__tests__/ssePatchArchitecture.test.ts | 3 +- ...temperatureDefaultVisibilityPolicy.test.ts | 142 ++++++++++++++++ .../scan-terminal/temperature-chart-logic.ts | 93 +++++++++-- frontend/lib/dashboard-types.ts | 2 + src/data_collection/amos_station_sources.py | 36 ++++ src/data_collection/weather_sources.py | 22 ++- src/utils/telegram_push.py | 154 +++++++++++------- tests/test_amos_station_sources.py | 3 + 9 files changed, 380 insertions(+), 77 deletions(-) diff --git a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx index 5dda005f..2b1576cf 100644 --- a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx +++ b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx @@ -117,7 +117,7 @@ export function LiveTemperatureThresholdChart({ const latestPatch = useLatestPatch(city); const resyncVersion = useSseResyncVersion(); const timeframe = "1D"; - const [viewMode, setViewMode] = useState<"auto" | "full">("auto"); + const [viewMode, setViewMode] = useState<"auto" | "full">("full"); const [userToggledKeys, setUserToggledKeys] = useState>({}); const [liveTemp, setLiveTemp] = useState(null); const [isHourlyLoading, setIsHourlyLoading] = useState(false); diff --git a/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts index e0d4c520..96c97197 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts @@ -153,7 +153,8 @@ export function runTests() { "SSE replay resync should refresh full detail in the background without showing the loading overlay", ); assert(chart.includes("viewMode"), "temperature chart must expose a view mode for DEB-peak auto view versus full-day view"); - assert(chart.includes("getDebPeakWindowRange"), "temperature chart must derive its default view from the DEB peak window"); + assert(chart.includes('useState<"auto" | "full">("full")'), "temperature chart must default every city panel to the all-day view"); + assert(chart.includes("getDebPeakWindowRange"), "temperature chart must still derive the optional Peak view from the DEB peak window"); assert( chart.includes('isEn ? "Peak" : "高温"') && chart.includes('isEn ? "All Day" : "全天"'), "temperature chart view-mode labels must translate 高温/全天 as Peak/All Day", diff --git a/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts index 02f8bab2..5c48bcbe 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts @@ -73,6 +73,32 @@ export function runTests() { ] as any).state === "watch", "US Fahrenheit charts should convert Celsius thresholds against observed highs before deciding peak glow state", ); + assert( + __getPeakGlowStateForTest({ temp_symbol: "°C", current_max_so_far: 25.0 } as any, [ + { ts: Date.UTC(2026, 4, 27, 0, 0), hourly_forecast: 22.0, runway: 25.0 }, + { ts: Date.UTC(2026, 4, 27, 4, 0), hourly_forecast: 21.6, runway: 24.4 }, + { ts: Date.UTC(2026, 4, 27, 8, 12), hourly_forecast: 22.1, runway: 25.0 }, + { ts: Date.UTC(2026, 4, 27, 12, 0), hourly_forecast: 26.8, runway: null }, + { ts: Date.UTC(2026, 4, 27, 15, 0), hourly_forecast: 28.0, runway: null }, + { ts: Date.UTC(2026, 4, 27, 18, 0), hourly_forecast: 25.0, runway: null }, + ] as any, [ + { + key: "runway_20R_02L", + label: "20R/02L", + source: "Runway", + color: "#009688", + values: [25.0, 24.4, 25.0, null, null, null], + }, + { + key: "hourly_forecast", + label: "DEB Forecast", + source: "DEB Hourly", + color: "#f97316", + values: [22.0, 21.6, 22.1, 26.8, 28.0, 25.0], + }, + ] as any).state === "none", + "morning observations near the intraday observed high should not trigger peak glow before the forecast hot window", + ); const guangzhou = { city: "guangzhou", @@ -452,6 +478,122 @@ export function runTests() { "empty runwayPlateHistory should fall back to AMOS runway_obs so runway cities still draw runway curves", ); + const busanWithRunwayHistory = __buildTemperatureChartDataForTest( + { + city: "busan", + local_date: "2026-05-27", + local_time: "08:20", + tz_offset_seconds: 9 * 60 * 60, + temp_symbol: "°C", + } as any, + { + localTime: "08:20", + times: ["00:00", "12:00", "18:00", "23:00"], + temps: [19.6, 21.1, 20.0, 19.0], + airportPrimary: { + source_code: "amos", + source_label: "AMOS", + temp: 21.0, + obs_time: "2026-05-26T23:20:00Z", + }, + airportPrimaryTodayObs: [ + ["2026-05-26T23:19:00Z", 21.0], + ["2026-05-26T23:20:00Z", 21.0], + ], + runwayPlateHistory: { + "SR/SL": [ + { time: "2026-05-26T23:19:00Z", temp: 20.9 }, + { time: "2026-05-26T23:20:00Z", temp: 21.1 }, + ], + }, + } as any, + "1D", + ); + assert( + !seriesByKey(busanWithRunwayHistory.series, "madis"), + "Busan should not render the AMOS aggregate airport-primary series when runway sensor data is available", + ); + const busanRunway = seriesByKey(busanWithRunwayHistory.series, runwayKey("SR/SL")) as any; + assert(busanRunway, "Busan SR/SL runway history should render as the runway curve"); + assert(busanRunway.featured === true, "Busan SR/SL should be treated as the settlement runway"); + assert(busanRunway.label.includes("结算跑道"), "Busan SR/SL should be labeled as the settlement runway"); + + const busanMergedHourly = __mergePatchIntoHourlyForTest( + { + localTime: "08:19", + times: ["00:00", "12:00", "18:00", "23:00"], + temps: [19.6, 21.1, 20.0, 19.0], + runwayPlateHistory: { + "SR/SL": [{ time: "2026-05-26T23:19:00Z", temp: 20.9 }], + }, + } as any, + { + type: "city_observation_patch.v1", + city: "busan", + revision: 21, + changes: { + temp: 21.1, + obs_time: "2026-05-26T23:20:00Z", + source: "amos", + amos: { + source: "amos", + icao: "RKPK", + runway_obs: { + runway_pairs: [["S R", "S L"]], + temperatures: [[21.1, 12.4]], + }, + }, + }, + } as any, + ); + const busanMergedChart = __buildTemperatureChartDataForTest( + { + city: "busan", + local_date: "2026-05-27", + local_time: "08:20", + tz_offset_seconds: 9 * 60 * 60, + temp_symbol: "°C", + } as any, + busanMergedHourly as any, + "1D", + ); + const busanMergedRunway = seriesByKey(busanMergedChart.series, runwayKey("SR/SL")) as any; + assert(busanMergedRunway, "AMOS runway_obs patch should append Busan SR/SL into runway history"); + assert( + busanMergedRunway.values.some((value: number | null) => value === 21.1), + "AMOS runway_obs patch should use the runway temperature, not ignore the SR/SL point", + ); + + const busanCurrentOnly = __buildTemperatureChartDataForTest( + { + city: "busan", + local_date: "2026-05-27", + local_time: "08:20", + tz_offset_seconds: 9 * 60 * 60, + temp_symbol: "°C", + } as any, + { + localTime: "08:20", + times: ["00:00", "12:00", "18:00", "23:00"], + temps: [19.6, 21.1, 20.0, 19.0], + amos: { + source: "amos", + observation_time: "2026-05-26T23:20:00Z", + runway_obs: { + runway_pairs: [["S R", "S L"]], + temperatures: [[21.1, 12.4]], + }, + }, + } as any, + "1D", + ); + const busanCurrentRunway = seriesByKey(busanCurrentOnly.series, runwayKey("SR/SL")) as any; + const busanCurrentValues = (busanCurrentRunway?.values || []).filter((value: number | null) => value !== null); + assert( + !busanCurrentValues.includes(12.4), + "AMOS temp/dew tuples should not be misread as two runway temperature samples", + ); + const newYorkMetrics = __getObservationDisplayMetricsForTest( { city: "new york", diff --git a/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts b/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts index 68140e9b..46172105 100644 --- a/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts +++ b/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts @@ -22,6 +22,7 @@ const SETTLEMENT_RUNWAY_PAIRS: Record> = { chongqing: [["20R", "02L"]], wuhan: [["04", "22"]], seoul: [["15R", "33L"]], + busan: [["SR", "SL"]], }; function normalizeRunwayLabel(value?: string | null) { @@ -117,18 +118,22 @@ function buildRunwayPlates( if (!Array.isArray(pair) || pair.length < 2) return; const isSettlement = settlementKeys.has(pairKey(pair)); - const tdz = validNumber(pointTemps[index]?.tdz_temp); - const mid = validNumber(pointTemps[index]?.mid_temp); - const end = validNumber(pointTemps[index]?.end_temp); + const pointTemp = pointTemps[index] as any; + const aggregateRunwayTemp = validNumber(pointTemp?.temp) ?? validNumber(pointTemp?.target_runway_max); + const tdz = validNumber(pointTemp?.tdz_temp); + const mid = validNumber(pointTemp?.mid_temp); + const end = validNumber(pointTemp?.end_temp); + const isAmosTempDewTuple = String(amos.source || "").toLowerCase() === "amos"; - const historyVals = Array.isArray(runwayTemps[index]) + const historyVals = !isAmosTempDewTuple && Array.isArray(runwayTemps[index]) ? (runwayTemps[index] as Array).map(validNumber).filter((v): v is number => v !== null) : []; + const aggregateVal = aggregateRunwayTemp !== null ? [aggregateRunwayTemp] : []; const tdzVal = tdz !== null ? [tdz] : []; const midVal = mid !== null ? [mid] : []; const endVal = end !== null ? [end] : []; - const allVals = [...historyVals, ...tdzVal, ...midVal, ...endVal]; + const allVals = [...historyVals, ...aggregateVal, ...tdzVal, ...midVal, ...endVal]; const maxTemp = allVals.length ? Math.max(...allVals) : null; const dailyHigh = historyVals.length ? Math.max(...historyVals) : maxTemp; @@ -606,6 +611,37 @@ function runwayLabelFromPair(rawPair: unknown, index: number) { return `RWY ${index + 1}`; } +function runwayTemperatureFromPairTuple(rawTemp: unknown) { + if (Array.isArray(rawTemp)) return validNumber(rawTemp[0]); + return validNumber(rawTemp); +} + +function runwayPatchPointsFromRunwayObs(runwayObs: any) { + const directPoints = Array.isArray(runwayObs?.point_temperatures) + ? runwayObs.point_temperatures + : []; + if (directPoints.length) return directPoints; + + const runwayPairs = Array.isArray(runwayObs?.runway_pairs) + ? runwayObs.runway_pairs + : []; + const temperatures = Array.isArray(runwayObs?.temperatures) + ? runwayObs.temperatures + : []; + + return runwayPairs + .map((pair: unknown, index: number) => { + const temp = runwayTemperatureFromPairTuple(temperatures[index]); + if (temp === null) return null; + return { + runway: runwayLabelFromPair(pair, index), + temp, + target_runway_max: temp, + }; + }) + .filter((point: any): point is { runway: string; temp: number; target_runway_max: number } => point !== null); +} + type HourlyForecast = { forecastTodayHigh?: number | null; debPrediction?: number | null; @@ -874,8 +910,8 @@ function mergePatchIntoHourly( const runwayObs = amosChanges?.runway_obs; const runwayPoints = Array.isArray(changes.runway_points) ? changes.runway_points - : runwayObs && Array.isArray(runwayObs.point_temperatures) - ? runwayObs.point_temperatures + : runwayObs + ? runwayPatchPointsFromRunwayObs(runwayObs) : []; if (runwayPoints.length && obsTimeVal) { const history: Record>> = {}; @@ -1014,6 +1050,7 @@ function buildRunwayHistorySeries( const runwayPairs = runwayObs?.runway_pairs || []; const runwayTemps = runwayObs?.temperatures || []; const pointTemps = runwayObs?.point_temperatures || []; + const isAmosTempDewTuple = String(amos?.source || "").toLowerCase() === "amos"; const anchor = getCityLocalUtcTimestamp(amos?.observation_time_local || amos?.observation_time || hourly?.localTime || row?.local_time, tzOffset, localDateStr) ?? getCityLocalUtcTimestamp(row?.local_time, tzOffset, localDateStr); @@ -1025,14 +1062,20 @@ function buildRunwayHistorySeries( if (!Array.isArray(rawTemps)) return null; const rwy = runwayLabelFromPair(runwayPairs[index], index); const isSettlement = isSettlementRunway(row, rwy); - const pointTemp = Array.isArray(pointTemps) ? pointTemps[index] : null; + const pointTemp = Array.isArray(pointTemps) ? (pointTemps[index] as any) : null; + const aggregateRunwayTemp = + validNumber(pointTemp?.temp) ?? + validNumber(pointTemp?.target_runway_max) ?? + (isAmosTempDewTuple ? runwayTemperatureFromPairTuple(rawTemps) : null); const snapshotValues = [ - validNumber((pointTemp as any)?.tdz_temp), - validNumber((pointTemp as any)?.mid_temp), - validNumber((pointTemp as any)?.end_temp), - validNumber((pointTemp as any)?.target_runway_max), + aggregateRunwayTemp, + validNumber(pointTemp?.tdz_temp), + validNumber(pointTemp?.mid_temp), + validNumber(pointTemp?.end_temp), ].filter((value): value is number => value !== null); - const samples = rawTemps.map(validNumber).filter((value): value is number => value !== null); + const samples = isAmosTempDewTuple + ? [] + : rawTemps.map(validNumber).filter((value): value is number => value !== null); const valuesForLine = samples.length > 1 ? samples : snapshotValues.length > 1 @@ -1336,13 +1379,26 @@ function buildFullDayChartData( const isAmscSource = (hourly?.airportPrimary as any)?.source === "amsc_awos" || String(hourly?.airportPrimary?.source_label || "").toLowerCase().includes("amsc"); + const isKoreanAmosSource = + (settlementCityKey === "seoul" || settlementCityKey === "busan") && + ( + String( + (hourly?.airportPrimary as any)?.source || + hourly?.airportPrimary?.source_code || + hourly?.airportPrimary?.source_label || + hourly?.amos?.source || + "", + ).toLowerCase().includes("amos") || + Boolean(hourly?.amos?.runway_obs) + ); + const isRunwaySensorAggregateSource = isAmscSource || isKoreanAmosSource; const shouldRenderMetar = metarObs.length > 0 && !observationSetContains(finalMadisObs, metarObs); const timelineSet = new Set(); runwayHistorySeries.forEach((rhs) => rhs.points.forEach((point) => timelineSet.add(point.ts))); normBandObs.forEach((point) => timelineSet.add(point.ts)); finalSettlementObs.forEach((point) => timelineSet.add(point.ts)); - if (!isAmscSource) finalMadisObs.forEach((point) => timelineSet.add(point.ts)); + if (!isRunwaySensorAggregateSource) finalMadisObs.forEach((point) => timelineSet.add(point.ts)); if (shouldRenderMetar) metarObs.forEach((point) => timelineSet.add(point.ts)); let debPath: ReturnType | null = null; @@ -1416,7 +1472,7 @@ function buildFullDayChartData( // ── Airport Primary (MADIS / AMSC AWOS) ── // Skip this series for AMSC AWOS cities — their data is redundant with // runway sensor data and adds a confusing "AMSC AWOS" label to the chart. - if (finalMadisObs.length && !isAmscSource) { + if (finalMadisObs.length && !isRunwaySensorAggregateSource) { const madisVals = valuesAtTimeline(n, indexByTs, finalMadisObs); if (madisVals.some((v) => v !== null)) { series.push({ @@ -1672,6 +1728,13 @@ function getPeakGlowState( observedHigh, }; + const hotWindowRange = getDebPeakWindowRange(data, series); + const hotWindowStart = + hotWindowRange ? validNumber(data[hotWindowRange[0]]?.ts) : null; + if (hotWindowStart !== null && latest.ts < hotWindowStart) { + return { state: "none", ...metaBase }; + } + const nearThreshold = chartDeltaForCelsius(row, 0.5); const watchThreshold = chartDeltaForCelsius(row, 1); const flatTrendFloor = -chartDeltaForCelsius(row, 0.2); diff --git a/frontend/lib/dashboard-types.ts b/frontend/lib/dashboard-types.ts index 450c131a..9291ae0e 100644 --- a/frontend/lib/dashboard-types.ts +++ b/frontend/lib/dashboard-types.ts @@ -953,9 +953,11 @@ export interface AmosData { temperatures?: Array<[number | null, number | null]> | null; point_temperatures?: Array<{ runway?: string | null; + temp?: number | null; tdz_temp?: number | null; mid_temp?: number | null; end_temp?: number | null; + target_runway_max?: number | null; }> | null; pressures_hpa?: Array | null; wind_directions?: Array<[number, number, number] | null> | null; diff --git a/src/data_collection/amos_station_sources.py b/src/data_collection/amos_station_sources.py index e73c4ad3..b4f25d6c 100644 --- a/src/data_collection/amos_station_sources.py +++ b/src/data_collection/amos_station_sources.py @@ -103,6 +103,40 @@ def _amos_is_runway_token(value: str) -> bool: ) +def _amos_normalize_runway_label(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "").strip().upper()) + + +def _amos_runway_pair_label(pair: Any, index: int) -> str: + if isinstance(pair, (list, tuple)) and len(pair) >= 2: + left = _amos_normalize_runway_label(pair[0]) + right = _amos_normalize_runway_label(pair[1]) + if left and right: + return f"{left}/{right}" + return f"RWY {index + 1}" + + +def _amos_build_point_temperatures( + runway_pairs: list[tuple[str, str]], + temperatures: list[tuple[Any, Any]], +) -> list[dict[str, Any]]: + points: list[dict[str, Any]] = [] + for index, pair in enumerate(runway_pairs): + if index >= len(temperatures): + continue + temp = _amos_safe_float(temperatures[index][0]) + if temp is None: + continue + points.append( + { + "runway": _amos_runway_pair_label(pair, index), + "temp": temp, + "target_runway_max": temp, + } + ) + return points + + def _amos_parse_cell_table(lines: list[str]) -> Optional[dict[str, Any]]: """Parse the actual AMOS HTML table after it has been flattened to cells.""" runway_rows: list[dict[str, Any]] = [] @@ -200,6 +234,7 @@ def _amos_parse_cell_table(lines: list[str]) -> Optional[dict[str, Any]]: return { "runway_pairs": runway_pairs, "temperatures": temperatures, + "point_temperatures": _amos_build_point_temperatures(runway_pairs, temperatures), "pressures_hpa": pressures_hpa, "wind_directions": wind_directions, "wind_speeds": wind_speeds, @@ -323,6 +358,7 @@ def _amos_parse_runway_table(text: str) -> dict[str, Any]: return { "runway_pairs": runway_pairs, "temperatures": temperatures, + "point_temperatures": _amos_build_point_temperatures(runway_pairs, temperatures), "pressures_hpa": pressures_hpa, "wind_directions": wind_directions, "wind_speeds": wind_speeds, diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 4b83541f..c45f07f1 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -1371,14 +1371,32 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour runway_obs = amos_data.get("runway_obs") or {} rw_pairs = runway_obs.get("runway_pairs") or [] rw_temps = runway_obs.get("temperatures") or [] + point_temps = runway_obs.get("point_temperatures") or [] for i, (pair, (t, _d)) in enumerate(zip(rw_pairs, rw_temps)): - if t is not None and i < 4: + point = point_temps[i] if i < len(point_temps) and isinstance(point_temps[i], dict) else {} + runway_label = str(point.get("runway") or "").strip().upper() + if not runway_label and isinstance(pair, (list, tuple)) and len(pair) >= 2: + runway_label = f"{str(pair[0]).replace(' ', '').upper()}/{str(pair[1]).replace(' ', '').upper()}" + point_temp = point.get("temp") if point else None + if point_temp is None and point: + point_temp = point.get("target_runway_max") + if point_temp is None: + point_temp = t + if point_temp is not None and i < 4: DBManager().append_airport_obs( icao=f"{amos_data.get('icao', '')}_RWY_{i}", city=city_lower, - temp_c=t, + temp_c=point_temp, obs_time=amos_data.get("observation_time") or datetime.now().isoformat(), ) + if point_temp is not None and runway_label: + DBManager().append_runway_obs( + icao=amos_data.get("icao") or "", + city=city_lower, + runway=runway_label, + target_runway_max=point_temp, + otime_utc=amos_data.get("observation_time") or datetime.now().isoformat(), + ) except Exception: logger.exception("airport_obs_log append failed for amos city={}", city_lower) else: diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index df37acec..942abed7 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -61,6 +61,80 @@ _AIRPORT_EXECUTOR_LOCK = threading.Lock() _AIRPORT_EXECUTOR_MAX_WORKERS: int = 0 +def _get_airport_executor(max_workers: int) -> ThreadPoolExecutor: + global _AIRPORT_EXECUTOR, _AIRPORT_EXECUTOR_MAX_WORKERS + if _AIRPORT_EXECUTOR is None or _AIRPORT_EXECUTOR_MAX_WORKERS != max_workers: + with _AIRPORT_EXECUTOR_LOCK: + if _AIRPORT_EXECUTOR is None or _AIRPORT_EXECUTOR_MAX_WORKERS != max_workers: + if _AIRPORT_EXECUTOR is not None: + _AIRPORT_EXECUTOR.shutdown(wait=False) + _AIRPORT_EXECUTOR = ThreadPoolExecutor(max_workers=max_workers) + _AIRPORT_EXECUTOR_MAX_WORKERS = max_workers + return _AIRPORT_EXECUTOR + + +def _rate_limited_send(bot: Any, chat_id: str, message: str, **kwargs: Any) -> None: + """Throttle bot.send_message calls to avoid hitting Telegram rate limits.""" + global _SEND_MSG_LAST_TS + with _SEND_MSG_LOCK: + now = time.time() + wait = _SEND_MSG_MIN_INTERVAL_SEC - (now - _SEND_MSG_LAST_TS) + if wait > 0: + time.sleep(wait) + _SEND_MSG_LAST_TS = time.time() + bot.send_message(chat_id, message, **kwargs) + + +def _load_city_thread_ids() -> dict: + global _city_thread_ids + if _city_thread_ids: + return _city_thread_ids + paths = [ + _CITY_THREAD_IDS_PATH, + "/var/lib/polyweather/city_thread_ids.json", + "/app/data/city_thread_ids.json", + ] + for path in paths: + if os.path.isfile(path): + try: + with open(path, "r", encoding="utf-8") as f: + _city_thread_ids = json.load(f) + logger.info("loaded city_thread_ids from {}: {} cities", path, len(_city_thread_ids)) + +# Forum topic routing: maps city_key -> message_thread_id for the push forum group. +# Created by scripts/create_forum_topics.py, stored in the runtime data dir. +_CITY_THREAD_IDS_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), + "data", "city_thread_ids.json", +) +_FORUM_CHAT_ID = "-1003927451869" +_city_thread_ids: dict = {} + +# Shared HTTP session for AROME and auxiliary queries (connection reuse) +_HTTP_SESSION: Optional[requests_lib.Session] = None +_HTTP_SESSION_LOCK = threading.Lock() + +# Bot send_message rate limiter: max N messages per second across all threads +_SEND_MSG_LOCK = threading.Lock() +_SEND_MSG_LAST_TS: float = 0.0 +_SEND_MSG_MIN_INTERVAL_SEC = float(os.getenv("TELEGRAM_SEND_RATE_LIMIT_SEC", "0.05")) + + +def _get_http_session() -> requests_lib.Session: + global _HTTP_SESSION + if _HTTP_SESSION is None: + with _HTTP_SESSION_LOCK: + if _HTTP_SESSION is None: + _HTTP_SESSION = requests_lib.Session() + return _HTTP_SESSION + + +# Reusable executor for airport push cycles (avoids thread pool churn) +_AIRPORT_EXECUTOR: Optional[ThreadPoolExecutor] = None +_AIRPORT_EXECUTOR_LOCK = threading.Lock() +_AIRPORT_EXECUTOR_MAX_WORKERS: int = 0 + + def _get_airport_executor(max_workers: int) -> ThreadPoolExecutor: global _AIRPORT_EXECUTOR, _AIRPORT_EXECUTOR_MAX_WORKERS if _AIRPORT_EXECUTOR is None or _AIRPORT_EXECUTOR_MAX_WORKERS != max_workers: @@ -108,10 +182,10 @@ def _load_city_thread_ids() -> dict: def _resolve_thread_id(chat_id: str, city: str) -> int: """Return message_thread_id for a given chat and city, or 0 if not a forum topic.""" - if str(chat_id) != _FORUM_CHAT_ID: + if chat_id != _FORUM_CHAT_ID: return 0 mapping = _load_city_thread_ids() - city_key = str(city or "").strip().lower() + city_key = (city or "").strip().lower() return int(mapping.get(city_key) or 0) @@ -197,7 +271,7 @@ def _bucket_value(row: Dict[str, Any]) -> Optional[float]: n = _safe_float(row.get(key)) if n is not None: return n - label = str(row.get("label") or "").strip() + label = (row.get("label") or "").strip() m = re.search(r"(-?\d+(?:\.\d+)?)", label) if not m: return None @@ -208,7 +282,7 @@ def _bucket_bounds(row: Dict[str, Any]) -> Optional[Tuple[Optional[float], Optio value = _bucket_value(row) if value is None: return None - label = str(row.get("label") or "").strip().lower() + label = (row.get("label") or "").strip().lower() is_upper_tail = any(key in label for key in ("+", "or higher", "or above", "and above")) is_lower_tail = any(key in label for key in ("<=", "or lower", "or below", "and below")) if is_upper_tail and not is_lower_tail: @@ -369,7 +443,7 @@ def _severity_ok(alert_payload: Dict[str, Any], min_severity: str, min_trigger_c trigger_count = int(alert_payload.get("trigger_count") or 0) if trigger_count < min_trigger_count: return False - severity = str(alert_payload.get("severity") or "none").lower() + severity = (alert_payload.get("severity") or "none").lower() return SEVERITY_RANK.get(severity, 0) >= SEVERITY_RANK.get(min_severity, 0) @@ -384,8 +458,8 @@ def _market_price_cap_ok( if not isinstance(primary_market, dict): primary_market = {} market_slug = ( - str(market.get("selected_slug") or "").strip() - or str(primary_market.get("slug") or "").strip() + (market.get("selected_slug") or "").strip() + or (primary_market.get("slug") or "").strip() or "--" ) active = market.get("market_active") @@ -404,12 +478,12 @@ def _market_price_cap_ok( ) accepting_orders = _optional_bool(accepting_orders) market_tradable = _optional_bool(market.get("market_tradable")) - tradable_reason = str( + tradable_reason = ( market.get("market_tradable_reason") or primary_market.get("tradable_reason") or "" ).strip() - ended_at = str( + ended_at = ( market.get("market_ended_at_utc") or primary_market.get("ended_at_utc") or "" @@ -442,12 +516,12 @@ def _market_price_cap_ok( settle_ref = market.get("anchor_settlement") if settle_ref is None: settle_ref = market.get("open_meteo_settlement") - anchor_model = str(market.get("anchor_model") or "").strip() or "--" + anchor_model = (market.get("anchor_model") or "").strip() or "--" yes_buy = None bucket_label = None if isinstance(forecast_bucket, dict): yes_buy = _norm_prob(forecast_bucket.get("yes_buy")) - bucket_label = str(forecast_bucket.get("label") or "").strip() or None + bucket_label = (forecast_bucket.get("label") or "").strip() or None observed_floor = _observed_settlement_floor(alert_payload) bucket_bounds = _bucket_bounds(forecast_bucket) if isinstance(forecast_bucket, dict) else None @@ -482,14 +556,14 @@ def _market_price_cap_ok( def _trigger_type_key(alert_payload: Dict[str, Any]) -> str: trigger_types = sorted( - str(alert.get("type") or "").strip() + (alert.get("type") or "").strip() for alert in (alert_payload.get("triggered_alerts") or []) if alert.get("type") ) market = alert_payload.get("market_snapshot") or {} if isinstance(market, dict) and market.get("available"): - signal = str(market.get("signal_label") or "").strip() - bucket = str(market.get("selected_bucket") or "").strip() + signal = (market.get("signal_label") or "").strip() + bucket = (market.get("selected_bucket") or "").strip() if signal: trigger_types.append(f"mkt:{signal}:{bucket}") return "|".join(trigger_types) @@ -530,7 +604,7 @@ def _evidence_brief(alert_payload: Dict[str, Any]) -> str: forecast_bucket = market.get("forecast_bucket") or {} if isinstance(forecast_bucket, dict): - label = str(forecast_bucket.get("label") or "").strip() + label = (forecast_bucket.get("label") or "").strip() yes_buy = forecast_bucket.get("yes_buy") if label: parts.append(f"bucket={label}") @@ -659,7 +733,7 @@ _FUNCTION_HASHTAGS_EN = { def _city_hashtag(city: Optional[str]) -> Optional[str]: - text = str(city or "").strip() + text = (city or "").strip() if not text: return None parts = [part for part in re.split(r"[^A-Za-z0-9]+", text.title()) if part] @@ -669,7 +743,7 @@ def _city_hashtag(city: Optional[str]) -> Optional[str]: def _station_hashtag(station: Optional[str]) -> Optional[str]: - text = re.sub(r"[^A-Za-z0-9]+", "", str(station or "").upper()) + text = re.sub(r"[^A-Za-z0-9]+", "", (station or "").upper()) return f"#{text}" if text else None @@ -723,7 +797,7 @@ def _fmt(value: Any) -> str: def _normalize_runway_label(value: Any) -> str: - return re.sub(r"[^0-9A-Z]+", "", str(value or "").strip().upper()) + return re.sub(r"[^0-9A-Z]+", "", (value or "").strip().upper()) def _runway_pair_key(r1: Any, r2: Any) -> Tuple[str, str]: @@ -774,13 +848,13 @@ def _select_focus_runway_obs( def _settlement_runway_for_city(city: str) -> Optional[Tuple[str, str]]: """Return the settlement runway pair for a city, if configured.""" - pairs = SETTLEMENT_RUNWAY_PAIRS.get(str(city or "").strip().lower(), set()) + pairs = SETTLEMENT_RUNWAY_PAIRS.get((city or "").strip().lower(), set()) return next(iter(pairs)) if pairs else None def _is_settlement_runway(city: str, r1: str, r2: str) -> bool: """Check if a runway pair is the settlement anchor for this city.""" - pair_set = SETTLEMENT_RUNWAY_PAIRS.get(str(city or "").strip().lower(), set()) + pair_set = SETTLEMENT_RUNWAY_PAIRS.get((city or "").strip().lower(), set()) return _runway_pair_key(r1, r2) in pair_set @@ -788,7 +862,7 @@ def _wind_regime_label(city: str, wind_dir: Optional[int], language: Optional[st """Classify wind direction into a thermal regime label.""" if wind_dir is None: return None - regimes = WIND_REGIME.get(str(city or "").strip().lower(), {}) + regimes = WIND_REGIME.get((city or "").strip().lower(), {}) sea = regimes.get("sea_breeze") warm = regimes.get("warm_advection") if sea and sea[0] != sea[1] and sea[0] <= wind_dir <= sea[1]: @@ -993,42 +1067,6 @@ def _build_airport_status_message( language = _normalize_push_language(language or _telegram_push_language()) _AIRPORT_EN = {"seoul": "Incheon", "singapore": "Changi", "busan": "Gimhae", "tokyo": "Haneda", "ankara": "Esenboğa", "helsinki": "Vantaa", "amsterdam": "Schiphol", - "istanbul": "Airport", "paris": "Le Bourget", - "hong kong": "Observatory", "shenzhen": "LFS Observatory", - "taipei": "Songshan", "beijing": "Capital", "shanghai": "Pudong", - "guangzhou": "Baiyun", "qingdao": "Jiaodong", - "chengdu": "Shuangliu", "chongqing": "Jiangbei", "wuhan": "Tianhe", - "new york": "LaGuardia", "los angeles": "LAX", "chicago": "O'Hare", - "denver": "Buckley", "atlanta": "Hartsfield", "miami": "Intl", - "san francisco": "SFO", "houston": "Hobby", "dallas": "Love Field", - "austin": "Bergstrom", "seattle": "Sea-Tac", - "tel aviv": "Ben Gurion"} - en_name = city.title() - ap_name = _AIRPORT_EN.get(city, "") - time_suffix = f" · {local_time}" if local_time else "" - - amos = city_weather.get("amos") or {} - runway_data = amos.get("runway_obs") or {} - runway_pairs = runway_data.get("runway_pairs") or [] - runway_temps = runway_data.get("temperatures") or [] - point_temps = runway_data.get("point_temperatures") or [] - is_amsc = amos.get("source") in ("amsc_awos", "amos") - has_runway = bool(runway_pairs and (runway_temps or point_temps)) - amos_icao = amos.get("icao") or HIGH_FREQ_AIRPORT_ICAO.get(city, "") - settlement_pair = _settlement_runway_for_city(city) - - # ── Display temp: settlement runway max first, then airport temp ── - settlement_temp: Optional[float] = None - display_temp: Optional[float] = None - if point_temps: - for pt in point_temps: - rw = str(pt.get("runway") or "") - rw_parts = [p.strip() for p in str(rw).split("/") if p.strip()] - if settlement_pair and len(rw_parts) >= 2 and _runway_pair_key(rw_parts[0], rw_parts[1]) == _runway_pair_key(*settlement_pair): - tmax = pt.get("target_runway_max") - if tmax is not None: - settlement_temp = float(tmax) - break if settlement_temp is not None: display_temp = settlement_temp if display_temp is None: @@ -1303,7 +1341,7 @@ def _in_peak_time_window(city: str, city_weather: Dict[str, Any]) -> bool: if fallback and ((first_h is None) or (last_h is not None and last_h - first_h < 3)): first_h, last_h = fallback local_time = city_weather.get("local_time") or "" - if first_h is None or not local_time: + if first_h is None or last_h is None or not local_time: return False try: current_h, current_m = int(local_time[:2]), int(local_time[3:5]) diff --git a/tests/test_amos_station_sources.py b/tests/test_amos_station_sources.py index 0f7161c2..b71a1632 100644 --- a/tests/test_amos_station_sources.py +++ b/tests/test_amos_station_sources.py @@ -130,5 +130,8 @@ def test_amos_parser_handles_flattened_html_cells_and_busan_runway_labels(): assert parsed["runway_pairs"] == [("N L", "N R"), ("S R", "S L")] assert parsed["temperatures"] == [(None, None), (15.4, 9.0)] + assert parsed["point_temperatures"] == [ + {"runway": "SR/SL", "temp": 15.4, "target_runway_max": 15.4} + ] assert parsed["pressures_hpa"] == [None, 1018.2] assert parsed["wind_speeds"] == [(4.4, 3.6, 5.2), (4.8, 4.0, 5.8)]