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

feat: add multi-model Open-Meteo data collection and implement scan terminal dashboard service

Browse Source
This commit is contained in:
2569718930@qq.com
2026-04-26 06:55:16 +08:00
parent 901b870240
commit b25c9312da
4 changed files with 369 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/components/dashboard/ScanTerminalDashboard.tsx
+58 -22
View File
@@ -466,6 +466,10 @@ function isFullEnoughForDeepAnalysis(detail?: CityDetail | null) {
);
}
function waitForDeepAnalysisQueue(ms: number) {
return new Promise((resolve) => window.setTimeout(resolve, ms));
}
function AiCityTemperatureChart({ detail }: { detail: CityDetail }) {
const { locale } = useI18n();
const chartData = useMemo(
@@ -674,7 +678,7 @@ function AiPinnedCityCard({
body: JSON.stringify({
city: detailCityName,
force_refresh: aiRefreshToken > 0,
locale,
locale: "zh-CN",
}),
})
.then(async (response) => {
@@ -715,7 +719,7 @@ function AiPinnedCityCard({
return () => {
cancelled = true;
};
}, [aiForecastKey, aiRefreshToken, detailCityName, locale]);
}, [aiForecastKey, aiRefreshToken, detailCityName]);
const aiCityForecast = aiForecast.payload?.city_forecast || null;
const localizedFinalJudgment =
@@ -1215,6 +1219,8 @@ function ScanTerminalScreen() {
const [themeMode, setThemeMode] = useState<ThemeMode>("dark");
const lastMapSelectedCityRef = useRef<string>("");
const aiFullHydrationRef = useRef<Set<string>>(new Set());
const aiHydrationQueueRef = useRef<string[]>([]);
const aiHydrationRunningRef = useRef(false);
const timeSortedRows = useMemo(
() => sortRowsByUserTime(terminalData?.rows || []),
@@ -1373,6 +1379,48 @@ function ScanTerminalScreen() {
}
}, [activeDetailRow, store.cityDetailsByName, store.ensureCityDetail]);
const runAiHydrationQueue = useCallback(async () => {
if (aiHydrationRunningRef.current) return;
aiHydrationRunningRef.current = true;
try {
while (aiHydrationQueueRef.current.length > 0) {
const nextCity = aiHydrationQueueRef.current.shift();
const key = normalizeCityKey(nextCity || "");
if (!nextCity || !key) continue;
const existingDetail = findDetailForCity(store.cityDetailsByName, nextCity);
try {
const detail = await store.ensureCityDetail(
nextCity,
Boolean(existingDetail) && !isFullEnoughForDeepAnalysis(existingDetail),
"full",
);
if (!isFullEnoughForDeepAnalysis(detail)) {
aiFullHydrationRef.current.delete(key);
}
} catch {
aiFullHydrationRef.current.delete(key);
}
await waitForDeepAnalysisQueue(1200);
}
} finally {
aiHydrationRunningRef.current = false;
if (aiHydrationQueueRef.current.length > 0) {
void runAiHydrationQueue();
}
}
}, [store.cityDetailsByName, store.ensureCityDetail]);
const queueAiFullHydration = useCallback(
(cityName: string) => {
const key = normalizeCityKey(cityName);
if (!key || aiFullHydrationRef.current.has(key)) return;
aiFullHydrationRef.current.add(key);
aiHydrationQueueRef.current.push(cityName);
void runAiHydrationQueue();
},
[runAiHydrationQueue],
);
const addAiPinnedCity = useCallback((cityName: string) => {
const cleanName = String(cityName || "").trim();
const key = normalizeCityKey(cleanName);
@@ -1404,22 +1452,15 @@ function ScanTerminalScreen() {
}
return [nextItem, ...current].slice(0, 8);
});
aiFullHydrationRef.current.delete(key);
aiFullHydrationRef.current.add(key);
void store
.ensureCityDetail(cleanName, true, "full")
.then((detail) => {
if (!isFullEnoughForDeepAnalysis(detail)) {
aiFullHydrationRef.current.delete(key);
}
})
.catch(() => {
aiFullHydrationRef.current.delete(key);
});
}, [locale, store.ensureCityDetail, timeSortedRows]);
queueAiFullHydration(matchedRow?.city || cleanName);
}, [locale, queueAiFullHydration, timeSortedRows]);
const removeAiPinnedCity = useCallback((cityName: string) => {
const key = normalizeCityKey(cityName);
aiFullHydrationRef.current.delete(key);
aiHydrationQueueRef.current = aiHydrationQueueRef.current.filter(
(queuedCity) => normalizeCityKey(queuedCity) !== key,
);
setAiPinnedCities((current) =>
current.filter((item) => normalizeCityKey(item.cityName) !== key),
);
@@ -1432,14 +1473,9 @@ function ScanTerminalScreen() {
const detail = findDetailForCity(store.cityDetailsByName, item.cityName);
const needsFullHydration = !isFullEnoughForDeepAnalysis(detail);
if (!needsFullHydration) return;
aiFullHydrationRef.current.add(key);
void store
.ensureCityDetail(item.cityName, Boolean(detail), "full")
.catch(() => {
aiFullHydrationRef.current.delete(key);
});
queueAiFullHydration(item.cityName);
});
}, [aiPinnedCities, store.cityDetailsByName, store.ensureCityDetail]);
}, [aiPinnedCities, queueAiFullHydration, store.cityDetailsByName]);
const handleMapCitySelect = useCallback((cityName: string) => {
setMapSelectedCityName(cityName);
frontend/hooks/useDashboardStore.tsx
+234 -18
View File
@@ -101,6 +101,10 @@ function getInitialProAccessState(): ProAccessState {
if (isBrowserLocalFullAccess()) {
return getLocalDevProAccessState();
}
const cached = readStoredProAccess();
if (cached) {
return cached;
}
return {
loading: true,
authenticated: false,
@@ -116,13 +120,124 @@ function getInitialProAccessState(): ProAccessState {
}
const SELECTED_CITY_STORAGE_KEY = "polyWeather_selected_city_v1";
const PRO_ACCESS_STORAGE_KEY = "polyWeather_pro_access_v1";
const CITY_LOAD_RETRY_DELAYS_MS = [700, 1600];
const PRO_ACCESS_FALLBACK_TTL_MS = 24 * 60 * 60 * 1000;
type CityDetailDepth = "panel" | "market" | "nearby" | "full";
type StoredProAccessState = ProAccessState & {
cachedAt: number;
expiresAtMs: number;
version: 1;
};
function wait(ms: number) {
return new Promise((resolve) => window.setTimeout(resolve, ms));
}
function getSubscriptionExpiryMs(access: Pick<
ProAccessState,
"subscriptionExpiresAt" | "subscriptionTotalExpiresAt"
>) {
const raw =
access.subscriptionTotalExpiresAt || access.subscriptionExpiresAt || "";
const parsed = Date.parse(raw);
return Number.isFinite(parsed) ? parsed : 0;
}
function clearStoredProAccess() {
if (typeof window === "undefined") return;
try {
window.localStorage.removeItem(PRO_ACCESS_STORAGE_KEY);
} catch {
// Ignore storage failures; backend auth remains the source of truth.
}
}
function readStoredProAccess(): ProAccessState | null {
if (typeof window === "undefined") return null;
try {
const raw = window.localStorage.getItem(PRO_ACCESS_STORAGE_KEY);
if (!raw) return null;
const parsed = JSON.parse(raw) as Partial<StoredProAccessState>;
if (parsed.version !== 1) {
clearStoredProAccess();
return null;
}
if (!parsed.authenticated || !parsed.subscriptionActive || !parsed.userId) {
clearStoredProAccess();
return null;
}
const expiresAtMs = Number(parsed.expiresAtMs || 0);
if (!Number.isFinite(expiresAtMs) || expiresAtMs <= Date.now()) {
clearStoredProAccess();
return null;
}
return {
loading: false,
authenticated: true,
userId: String(parsed.userId),
subscriptionActive: true,
subscriptionPlanCode: parsed.subscriptionPlanCode ?? null,
subscriptionExpiresAt: parsed.subscriptionExpiresAt ?? null,
subscriptionTotalExpiresAt: parsed.subscriptionTotalExpiresAt ?? null,
subscriptionQueuedDays: Math.max(
0,
Number(parsed.subscriptionQueuedDays ?? 0),
),
points: Number(parsed.points ?? 0),
error: null,
};
} catch {
clearStoredProAccess();
return null;
}
}
function writeStoredProAccess(access: ProAccessState) {
if (typeof window === "undefined") return;
if (!access.authenticated || !access.subscriptionActive || !access.userId) {
clearStoredProAccess();
return;
}
const explicitExpiryMs = getSubscriptionExpiryMs(access);
const expiresAtMs =
explicitExpiryMs > Date.now()
? explicitExpiryMs
: Date.now() + PRO_ACCESS_FALLBACK_TTL_MS;
const payload: StoredProAccessState = {
...access,
loading: false,
error: null,
cachedAt: Date.now(),
expiresAtMs,
version: 1,
};
try {
window.localStorage.setItem(PRO_ACCESS_STORAGE_KEY, JSON.stringify(payload));
} catch {
// Storage can be unavailable in private mode; keep the in-memory state.
}
}
function mergeWithStoredProAccess(
next: ProAccessState,
reason: string,
): ProAccessState {
if (next.subscriptionActive || !next.authenticated) return next;
const cached = readStoredProAccess();
if (!cached) return next;
if (next.userId && cached.userId !== next.userId) return next;
const payloadExpiryMs = getSubscriptionExpiryMs(next);
if (payloadExpiryMs > 0 && payloadExpiryMs <= Date.now()) return next;
return {
...cached,
loading: false,
points: Math.max(cached.points, next.points),
error: reason,
};
}
async function buildAuthMeHeaders(): Promise<HeadersInit> {
const headers: Record<string, string> = {
Accept: "application/json",
@@ -247,6 +362,52 @@ function hasMeaningfulDailyModelMap(
);
}
function countModelMapEntries(value: Record<string, number | null> | undefined) {
if (!value || typeof value !== "object") return 0;
return Object.values(value).filter((entry) => Number.isFinite(Number(entry))).length;
}
function pickRicherModelMap(
currentValue: CityDetail["multi_model"] | undefined,
incomingValue: CityDetail["multi_model"] | undefined,
) {
return countModelMapEntries(incomingValue) >= countModelMapEntries(currentValue)
? incomingValue || currentValue
: currentValue;
}
function mergeDailyModelMap(
currentValue: CityDetail["multi_model_daily"] | undefined,
incomingValue: CityDetail["multi_model_daily"] | undefined,
) {
if (!hasMeaningfulDailyModelMap(incomingValue)) return currentValue;
if (!hasMeaningfulDailyModelMap(currentValue)) return incomingValue;
const merged = { ...(currentValue || {}) };
Object.entries(incomingValue || {}).forEach(([date, incomingDay]) => {
const currentDay = merged[date];
const incomingCount = countModelMapEntries(incomingDay?.models || undefined);
const currentCount = countModelMapEntries(currentDay?.models || undefined);
if (incomingCount >= currentCount) {
merged[date] = {
...(currentDay || {}),
...(incomingDay || {}),
models: incomingDay?.models || currentDay?.models,
};
}
});
return merged;
}
function pickRicherForecast(
currentValue: CityDetail["forecast"] | undefined,
incomingValue: CityDetail["forecast"] | undefined,
) {
return countForecastDays({ forecast: incomingValue } as CityDetail) >=
countForecastDays({ forecast: currentValue } as CityDetail)
? incomingValue || currentValue
: currentValue;
}
function pickPreferredNearbyStations(
currentValue: CityDetail["official_nearby"] | CityDetail["mgm_nearby"],
incomingValue: CityDetail["official_nearby"] | CityDetail["mgm_nearby"],
@@ -264,12 +425,17 @@ function mergeCityDetail(
incoming: CityDetail,
): CityDetail {
if (!current) return incoming;
if (incoming.detail_depth !== "market") return incoming;
const currentDepth = normalizeDetailDepth(current);
const incomingDepth = normalizeDetailDepth(incoming);
const mergedDepth =
current.detail_depth === "full" || current.detail_depth === "nearby"
? current.detail_depth
: incoming.detail_depth;
currentDepth === "full" || incomingDepth === "full"
? "full"
: currentDepth === "nearby" || incomingDepth === "nearby"
? "nearby"
: currentDepth === "market" || incomingDepth === "market"
? "market"
: "panel";
return {
...current,
@@ -280,16 +446,12 @@ function mergeCityDetail(
deb: incoming.deb || current.deb,
probabilities: incoming.probabilities || current.probabilities,
trend: incoming.trend || current.trend,
multi_model: hasMeaningfulModelMap(incoming.multi_model)
? incoming.multi_model
: current.multi_model,
multi_model_daily: hasMeaningfulDailyModelMap(incoming.multi_model_daily)
? {
...(current.multi_model_daily || {}),
...(incoming.multi_model_daily || {}),
}
: current.multi_model_daily,
forecast: current.forecast || incoming.forecast,
multi_model: pickRicherModelMap(current.multi_model, incoming.multi_model),
multi_model_daily: mergeDailyModelMap(
current.multi_model_daily,
incoming.multi_model_daily,
),
forecast: pickRicherForecast(current.forecast, incoming.forecast),
official_nearby: pickPreferredNearbyStations(
current.official_nearby,
incoming.official_nearby,
@@ -641,12 +803,14 @@ export function DashboardStoreProvider({
const refreshProAccess = async () => {
if (isBrowserLocalFullAccess()) {
setProAccess(getLocalDevProAccessState());
const localAccess = getLocalDevProAccessState();
writeStoredProAccess(localAccess);
setProAccess(localAccess);
return;
}
setProAccess((current) => ({
...current,
loading: true,
loading: current.subscriptionActive ? false : true,
error: null,
}));
try {
@@ -667,8 +831,10 @@ export function DashboardStoreProvider({
subscription_total_expires_at?: string | null;
subscription_queued_days?: number | null;
points?: number;
degraded_auth_profile?: boolean | null;
degraded_reason?: string | null;
};
setProAccess({
const nextAccess: ProAccessState = {
loading: false,
authenticated: Boolean(payload.authenticated),
userId: payload.user_id ?? null,
@@ -683,8 +849,30 @@ export function DashboardStoreProvider({
),
points: payload.points ?? 0,
error: null,
});
};
const mergedAccess = mergeWithStoredProAccess(
nextAccess,
payload.degraded_auth_profile
? String(payload.degraded_reason || "degraded_auth_profile")
: "using_cached_pro_access",
);
if (mergedAccess.subscriptionActive) {
writeStoredProAccess(mergedAccess);
} else if (!mergedAccess.authenticated || payload.subscription_active === false) {
clearStoredProAccess();
}
setProAccess(mergedAccess);
} catch (error) {
const cachedAccess = readStoredProAccess();
if (cachedAccess) {
setProAccess({
...cachedAccess,
loading: false,
error: String(error),
});
return;
}
clearStoredProAccess();
setProAccess({
loading: false,
authenticated: false,
@@ -727,6 +915,34 @@ export function DashboardStoreProvider({
void refreshProAccess();
}, []);
useEffect(() => {
if (!hasSupabasePublicEnv()) return;
const {
data: { subscription },
} = getSupabaseBrowserClient().auth.onAuthStateChange((event) => {
if (event === "SIGNED_OUT") {
clearStoredProAccess();
setProAccess({
loading: false,
authenticated: false,
userId: null,
subscriptionActive: false,
subscriptionPlanCode: null,
subscriptionExpiresAt: null,
subscriptionTotalExpiresAt: null,
subscriptionQueuedDays: 0,
points: 0,
error: null,
});
return;
}
if (event === "SIGNED_IN" || event === "TOKEN_REFRESHED") {
void refreshProAccess();
}
});
return () => subscription.unsubscribe();
}, []);
const ensureCitySummary = async (cityName: string, force = false) => {
const existing = citySummariesRef.current[cityName];
if (!force && existing) {
src/data_collection/nws_open_meteo_sources.py
+71
View File
@@ -139,6 +139,66 @@ def _parse_open_meteo_multi_model_daily(
return [str(d) for d in dates], daily_forecasts, model_metadata, model_keys
def _count_models(values: Any) -> int:
if not isinstance(values, dict):
return 0
count = 0
for value in values.values():
try:
float(value)
count += 1
except (TypeError, ValueError):
continue
return count
def _merge_multi_model_result_with_cache(
cached: Dict[str, Any],
fresh: Dict[str, Any],
) -> Dict[str, Any]:
"""Avoid downgrading a rich multi-model cache when Open-Meteo returns a sparse subset."""
if not isinstance(cached, dict):
return fresh
cached_forecasts = cached.get("forecasts") if isinstance(cached.get("forecasts"), dict) else {}
fresh_forecasts = fresh.get("forecasts") if isinstance(fresh.get("forecasts"), dict) else {}
if _count_models(fresh_forecasts) >= _count_models(cached_forecasts):
return fresh
merged_daily: Dict[str, Dict[str, float]] = {}
cached_daily = cached.get("daily_forecasts") if isinstance(cached.get("daily_forecasts"), dict) else {}
fresh_daily = fresh.get("daily_forecasts") if isinstance(fresh.get("daily_forecasts"), dict) else {}
for date in sorted({*cached_daily.keys(), *fresh_daily.keys()}):
cached_day = cached_daily.get(date) if isinstance(cached_daily.get(date), dict) else {}
fresh_day = fresh_daily.get(date) if isinstance(fresh_daily.get(date), dict) else {}
merged_daily[str(date)] = (
dict(fresh_day)
if _count_models(fresh_day) >= _count_models(cached_day)
else {**dict(fresh_day), **dict(cached_day)}
)
merged = {
**cached,
**fresh,
"forecasts": {**dict(fresh_forecasts), **dict(cached_forecasts)},
"daily_forecasts": merged_daily or cached_daily or fresh_daily,
"model_metadata": {
**(fresh.get("model_metadata") if isinstance(fresh.get("model_metadata"), dict) else {}),
**(cached.get("model_metadata") if isinstance(cached.get("model_metadata"), dict) else {}),
},
"model_keys": {
**(fresh.get("model_keys") if isinstance(fresh.get("model_keys"), dict) else {}),
**(cached.get("model_keys") if isinstance(cached.get("model_keys"), dict) else {}),
},
"partial_refresh": True,
"partial_refresh_model_count": _count_models(fresh_forecasts),
"preserved_cached_model_count": _count_models(cached_forecasts),
}
merged_dates = list(dict.fromkeys([*(fresh.get("dates") or []), *(cached.get("dates") or [])]))
if merged_dates:
merged["dates"] = merged_dates
return merged
class NwsOpenMeteoSourceMixin:
def fetch_nws(self, lat: float, lon: float) -> Optional[Dict]:
"""
@@ -693,6 +753,17 @@ class NwsOpenMeteoSourceMixin:
"unit": "fahrenheit" if use_fahrenheit else "celsius",
"attribution": "Open-Meteo forecast model API; underlying models from ECMWF, DWD, ECCC, NOAA and JMA.",
}
with self._multi_model_cache_lock:
previous = self._multi_model_cache.get(cache_key)
previous_data = previous.get("data") if isinstance(previous, dict) else None
if isinstance(previous_data, dict):
result = _merge_multi_model_result_with_cache(previous_data, result)
if result.get("partial_refresh"):
logger.info(
"🔬 Multi-model sparse refresh preserved cache: fresh={} cached={}",
result.get("partial_refresh_model_count"),
result.get("preserved_cached_model_count"),
)
with self._multi_model_cache_lock:
self._multi_model_cache[cache_key] = {
"t": time.time(),
web/scan_terminal_service.py
+6 -12
View File
@@ -66,7 +66,7 @@ SCAN_CITY_AI_MAX_TOKENS = max(
8192,
int(os.getenv("POLYWEATHER_SCAN_CITY_AI_MAX_TOKENS", "8192")),
)
SCAN_CITY_AI_PROMPT_VERSION = "city-airport-read-v2"
SCAN_CITY_AI_PROMPT_VERSION = "city-airport-read-v3"
def _safe_float(value: Any) -> Optional[float]:
@@ -1067,9 +1067,6 @@ def _call_deepseek_city_ai(ai_input: Dict[str, Any], *, locale: str = "zh-CN") -
if not api_key:
raise RuntimeError("POLYWEATHER_DEEPSEEK_API_KEY is not configured")
normalized_locale = _normalize_locale(locale)
primary_language = "English" if normalized_locale == "en-US" else "Simplified Chinese"
primary_suffix = "_en" if normalized_locale == "en-US" else "_zh"
secondary_suffix = "_zh" if normalized_locale == "en-US" else "_en"
system_prompt = (
"你是 PolyWeather 的 Deepseek V4-Pro 城市最高温预测员。你必须直接阅读用户给出的城市 JSON，"
@@ -1085,20 +1082,18 @@ def _call_deepseek_city_ai(ai_input: Dict[str, Any], *, locale: str = "zh-CN") -
"涉及 TAF 或云雨扰动时必须给出报文中的有效时间、BECMG/TEMPO/FM 时间窗或说明“未给出明确时间”；"
"如果没有 TAF 时间依据，不要笼统写“峰值窗口云雨扰动风险”。"
"如果峰值窗口尚未到来，不能过早下最终结论；如果峰值窗口已过或实测已创高，需要更重视 METAR 实测。"
f"当前用户界面语言是 {normalized_locale}，所有面向用户的主要自然语言字段必须使用 {primary_language}。"
f"重点填写 {primary_suffix} 字段；{secondary_suffix} 字段只填空字符串，前端不读取它。"
"所有面向用户的自然语言字段必须同时填写简体中文和英文两套内容："
"_zh 字段写简体中文，_en 字段写英文。前端会按用户界面语言直接切换字段，不能留空。"
"risks 最多 2 条，每条必须包含触发条件或方向来源；reasoning、model_cluster_note 各 1 句，metar_read 可用 2-4 句。"
"只返回 JSON object，不要 Markdown。"
)
user_payload = {
"locale": normalized_locale,
"primary_language": primary_language,
"task": (
"Return strict JSON with: predicted_max, range_low, range_high, unit, confidence, "
"final_judgment_zh, final_judgment_en, metar_read_zh, metar_read_en, "
"reasoning_zh, reasoning_en, risks_zh, risks_en, model_cluster_note_zh, model_cluster_note_en. "
f"Primary output language is {primary_language}; the UI will read fields ending with {primary_suffix}. "
f"Fields ending with {secondary_suffix} must be empty strings or empty arrays to avoid truncation. "
"Fill every *_zh field in Simplified Chinese and every *_en field in English in the same response. "
"Keep final_judgment one short decision sentence. metar_read must explain the latest airport bulletin "
"with report time, temperature, wind direction/speed, cloud/weather/visibility/dewpoint if available. "
"For wind, explicitly say whether the current wind tends to warm, cool, or be neutral for today's high, "
@@ -1225,7 +1220,6 @@ def _call_deepseek_city_ai(ai_input: Dict[str, Any], *, locale: str = "zh-CN") -
"content": json.dumps(
{
"locale": normalized_locale,
"primary_language": primary_language,
"required_schema": [
"predicted_max",
"range_low",
@@ -1247,7 +1241,7 @@ def _call_deepseek_city_ai(ai_input: Dict[str, Any], *, locale: str = "zh-CN") -
"previous_assistant_content": _truncate_ai_text(content, 5000),
"city_snapshot": ai_input,
"instruction": (
f"Primary UI language is {primary_language}. "
"Fill *_zh fields in Simplified Chinese and *_en fields in English; do not leave either language empty. "
"Make final_judgment one direct sentence about today's high temperature. "
"metar_read must interpret the latest airport bulletin with report time, temperature, "
"wind direction/speed, cloud/weather/visibility/dewpoint if available. State whether "
@@ -1348,7 +1342,7 @@ def build_scan_city_ai_forecast_payload(
normalized_locale,
SCAN_AI_MODEL,
)
cache_key = f"city_forecast:{SCAN_CITY_AI_PROMPT_VERSION}:{city_name.lower()}:{normalized_locale}"
cache_key = f"city_forecast:{SCAN_CITY_AI_PROMPT_VERSION}:{city_name.lower()}"
if not force_refresh:
with _SCAN_CITY_AI_CACHE_LOCK:
cached = _SCAN_CITY_AI_CACHE.get(cache_key)