Python天气预报可视化毕设：从API集成到交互式图表的完整技术实现-开发者社区

Python天气预报可视化毕设：从API集成到交互式图表的完整技术实现

摘要：许多同学在“Python天气预报可视化”毕设里被 API 限流、数据格式混乱、图表静态丑到哭。本文用一次真实开发流水账，带你把 OpenWeatherMap 的数据一路薅到 PyEcharts 的交互大屏，并给出可直接跑的模块化代码。读完能扛住答辩，也能扛住生产。

1. 学生党常见三大痛点

刚拿到免费 API Key，5 分钟就被 429 打回原型，还不懂重试。
JSON 字段忽多忽少，代码里一堆if 'temp' in data，一跑就 KeyError。
用 matplotlib 画完静态图，导师一句“能不能点点看”直接社死。

2. 技术栈 30 秒对比

需求场景	方案 A（入门）	方案 B（毕设够用）	怎么选
拉取数据	requests	aiohttp + asyncio	单城市 requests 够，多城市批量上 aiohttp
清洗&合并	csv 手写	Pandas	时间戳转本地、空值补全一行代码
可视化	matplotlib	PyEcharts / Plotly	前者论文图多，后者鼠标能点，答辩加分
缓存	无	diskcache / pickle	省配额、断网可演示

3. 核心实现拆解

3.1 异步请求封装（含指数退避）

单文件写死循环会崩，拆成weather_client.py：

import aiohttp, asyncio, os, time from datetime import datetime API_KEY = os.getenv("OWM_KEY") # 安全：不硬编码 BASE_URL = "https://api.openweathermap.org/data/2.5/forecast" class WeatherClient: def __init__(self, session: aiohttp.ClientSession): self.session = session async def get_forecast(self, city: str, retries=4) -> dict: params = {"q": city, "appid": API_KEY, "units": "metric", "lang": "zh_cn"} for attempt in range(1, retries+1): async with self.session.get(BASE_URL, params=params) as resp: if resp.status == 200: return await resp.json() elif resp.status == 429: wait = 2 ** attempt # 指数退避 await asyncio.sleep(wait) else: resp.raise_for_status() raise RuntimeError("仍被限流，请明天再试")

调用侧：

async def main(cities): async with aiohttp.ClientSession() as session: client = WeatherClient(session) tasks = [client.get_forecast(c) for c in cities] return await asyncio.gather(*tasks)

3.2 JSON 解析容错

天气 API 的list字段偶尔为空，封装“安全取值”小函数：

def safe_get(weather_dict, key, default=None): return weather_dict.get(key) or default

解析主函数只关心：dt,main.temp,weather[0].description，其余字段一律用safe_get，后续 DataFrame 统一补 NaN。

3.3 时间戳→本地时间

OpenWeatherMap 返回 UTC 时间戳，用 Pandas 一行转：

df["local_time"] = pd.to_datetime(df["dt"], unit="s").dt.tz_localize("UTC").dt.tz_convert("Asia/Shanghai")

4. 完整可运行示例（单文件 demo）

把上面模块拼成run.py，函数职责单一，方便单元测试：

import asyncio, os, pandas as pd, pickle, diskcache from weather_client import WeatherClient from pyecharts.charts import Line from pyecharts import options as opts CACHE = diskcache.Cache("./tmp_cache") async def fetch_or_load(cities): key = "_".join(sorted(cities)) if key in CACHE: return CACHE[key] async with aiohttp.ClientSession() as s: client = WeatherClient(s) raw = await asyncio.gather(*[client.get_forecast(c) for c in cities]) CACHE[key] = raw return raw def json_to_df(raw_list): frames = [] for raw in raw_list: city = raw["city"]["name"] for item in raw["list"]: frames.append({ "city": city, "dt": item["dt"], "temp": item["main"]["temp"], "desc": item["weather"][0]["description"] }) df = pd.DataFrame(frames) df["time"] = pd.to_datetime(df["dt"], unit="s").dt.tz_localize("UTC").dt.tz_convert("Asia/Shanghai") return df def build_line(df): line = Line() for city, grp in df.groupby("city"): line.add_xaxis(grp["time"].dt.strftime("%m%d-%H").tolist()) line.add_yaxis(city, grp["temp"].round(1).tolist(), is_smooth=True) line.set_global_opts(title_opts=opts.TitleOpts(title="72h 温度趋势"), datazoom_opts=[opts.DataZoomOpts()]) line.render("weather.html") if __name__ == "__main__": cities = ["Beijing", "Shanghai", "Guangzhou"] raw = asyncio.run(fetch_or_load(cities)) df = json_to_df(raw) build_line(df)

跑完会在当前目录生成weather.html，双击就能缩放、拖拽，导师鼠标点一点，好感+10。

5. 性能 & 安全小补丁

缓存：diskcache 自带 TTL，可CACHE.set(key, value, expire=3600)，演示前跑一遍，断网也能播。
API Key：写进.env，用 python-dotenv 加载，GitHub 仓库加.gitignore，答辩完公开代码不泄露。
限流：指数退避最大 4 次，总等待 2+4+8+16=30 s，基本覆盖免费套餐 60 调用/分钟。

6. 生产环境避坑指南

坐标解析失败：用户输入“徐家汇”可能返回多条，用city["coord"]同时记录经纬度，前端地图打点。
时区处理错误：系统若部署在 Docker 默认 UTC，一定把容器TZ=Asia/Shanghai，否则时间轴错位。
中文描述乱码：PyEcharts 默认 UTF-8，但 Windows PowerShell 可能 GBK，渲染前export PYTHONIOENCODING=utf-8。
缓存击穿：演示当天一口气刷新页面，可把diskcache换成redis，或者把 TTL 拉长到 6 h。