feat(plotting.py): 从空壳重写为完整的 Plotly 图表实现

原代码为轻量级存根,所有函数返回 None 以禁用可视化依赖。
本次重写实现了 10 个图表函数:

1. blank_figure()      - 返回空占位图(含 'Plot not available' 文本)
2. scatter_3d()        - 3D 散点图(飞行轨迹 + 气体浓度着色)
3. scatter_2d()        - 2D 散点图
4. time_series()       - 时序折线图(风速等)
5. background_plotting() - 背景校正可视化(原始数据/拟合基线/信号点/背景点)
6. windrose()           - 风玫瑰图(16 扇区 bar_polar)
7. outliers()           - 异常值检测箱线图(IQR 方法)
8. contour_krig()       - 克里金插值等高线图(go.Contour + RdBu_r 色标)
   注意:np.mgrid 生成 (x_nodes,y_nodes) 形状的数组,xx 沿 axis=0 变化,yy 沿 axis=1 变化,
   field 需转置 (.T) 后方可与 Plotly 的 z[y][x] 索引匹配
9. heatmap_krig()       - 克里金网格热力图(go.Heatmap + zmid=0 对称色标)
10. semivariogram_plot() - 半变异函数图(实验值散点 + 拟合模型曲线)
11. create_kml_file()   - KML 文件生成(保持返回 None,未实现)

所有图表使用 plotly.graph_objects,数据通过参数传入,异常时优雅降级
This commit is contained in:
DXC
2026-07-10 14:48:55 +08:00
parent d755367df0
commit e3acd46da9

View File

@ -1,44 +1,477 @@
""" """
Lightweight stub plotting module to disable heavy visualization dependencies. Plotting module for GasFlux visualization.
All functions return None so callers can safely check for truthiness. Generates interactive Plotly figures for reports.
""" """
import numpy as np
import pandas as pd
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
def blank_figure(): def blank_figure():
return None """Return an empty Plotly figure."""
fig = go.Figure()
fig.update_layout(
def scatter_3d(*args, **kwargs): title="No data available",
return None xaxis={"visible": False},
yaxis={"visible": False},
annotations=[{"text": "Plot not available", "showarrow": False, "font": {"size": 16}}],
def scatter_2d(*args, **kwargs): )
return None return fig
def time_series(*args, **kwargs): def scatter_3d(
return None df: pd.DataFrame,
gas: str,
x: str = "utm_easting",
def background_plotting(*args, **kwargs): y: str = "utm_northing",
return None z: str = "height_ato",
color_col: str | None = None,
) -> go.Figure:
def windrose(*args, **kwargs): """
return None Create a 3D scatter plot of flight path with gas concentration coloring.
Parameters:
def outliers(*args, **kwargs): df: DataFrame with position and gas data.
return None gas: Gas name (e.g. 'co2', 'ch4').
x, y, z: Column names for coordinates.
color_col: Column to use for color scale. Defaults to normalised gas column.
def contour_krig(*args, **kwargs):
return None Returns:
plotly.graph_objects.Figure
"""
def heatmap_krig(*args, **kwargs): if color_col is None:
return None norm_col = f"{gas}_normalised"
color_col = norm_col if norm_col in df.columns else gas
def create_kml_file(*args, **kwargs): if color_col not in df.columns:
return blank_figure()
valid = df[[x, y, z, color_col]].dropna()
if len(valid) == 0:
return blank_figure()
fig = go.Figure()
fig.add_trace(
go.Scatter3d(
x=valid[x],
y=valid[y],
z=valid[z],
mode="markers",
marker={
"size": 3,
"color": valid[color_col],
"colorscale": "Viridis",
"colorbar": {"title": f"{gas.upper()} (ppm)"},
"opacity": 0.8,
},
text=[f"{v:.2f} ppm" for v in valid[color_col]],
hoverinfo="text",
name="Flight path",
)
)
fig.update_layout(
title=f"3D Flight Path – {gas.upper()} Concentration",
scene={
"xaxis_title": "UTM Easting (m)",
"yaxis_title": "UTM Northing (m)",
"zaxis_title": "Height ATO (m)",
},
margin={"l": 0, "r": 0, "t": 40, "b": 0},
)
return fig
def scatter_2d(
df: pd.DataFrame,
x: str = "x",
y: str = "height_ato",
color_col: str | None = None,
title: str = "2D Scatter",
) -> go.Figure:
"""Create a 2D scatter plot."""
if color_col is None or color_col not in df.columns:
color_col = None
fig = px.scatter(
df,
x=x,
y=y,
color=color_col,
title=title,
color_continuous_scale="Viridis" if color_col else None,
)
fig.update_layout(margin={"l": 0, "r": 0, "t": 40, "b": 0})
return fig
def time_series(
df: pd.DataFrame,
x_col: str = "timestamp",
y_col: str = "windspeed",
title: str = "Wind Speed Over Time",
) -> go.Figure:
"""Create a time series line chart."""
if x_col not in df.columns or y_col not in df.columns:
return blank_figure()
fig = go.Figure()
fig.add_trace(
go.Scatter(
x=df[x_col],
y=df[y_col],
mode="lines",
name=y_col,
line={"color": "#3498db", "width": 1.5},
)
)
fig.update_layout(
title=title,
xaxis_title="Time",
yaxis_title=y_col,
margin={"l": 0, "r": 0, "t": 40, "b": 0},
)
return fig
def background_plotting(df: pd.DataFrame, gas: str) -> go.Figure:
"""
Visualise background correction: raw data, fitted baseline, signal and background points.
Parameters:
df: DataFrame with gas column, fitted baseline, and signal mask.
gas: Gas name.
Returns:
plotly.graph_objects.Figure
"""
required = [gas, f"{gas}_fit", f"{gas}_normalised"]
missing = [c for c in required if c not in df.columns]
if missing:
return blank_figure()
signal_mask = df.get(f"{gas}_signal", pd.Series([False] * len(df)))
fig = go.Figure()
# Raw data
fig.add_trace(
go.Scatter(
y=df[gas],
mode="lines",
name=f"{gas.upper()} raw",
line={"color": "#2c3e50", "width": 1},
opacity=0.6,
)
)
# Fitted baseline
fig.add_trace(
go.Scatter(
y=df[f"{gas}_fit"],
mode="lines",
name="Baseline fit",
line={"color": "#e74c3c", "width": 2},
)
)
# Background points
bg_df = df[~signal_mask]
if len(bg_df) > 0:
fig.add_trace(
go.Scatter(
y=bg_df[f"{gas}_normalised"],
mode="markers",
name="Background",
marker={"color": "#3498db", "size": 4, "symbol": "circle-open"},
)
)
# Signal points
sig_df = df[signal_mask]
if len(sig_df) > 0:
fig.add_trace(
go.Scatter(
y=sig_df[f"{gas}_normalised"],
mode="markers",
name="Signal",
marker={"color": "#e74c3c", "size": 5},
)
)
fig.update_layout(
title=f"Background Correction – {gas.upper()}",
xaxis_title="Sample Index",
yaxis_title=f"{gas.upper()} (ppm)",
margin={"l": 0, "r": 0, "t": 40, "b": 0},
legend={"orientation": "h", "yanchor": "bottom", "y": 1.02},
)
return fig
def windrose(
df: pd.DataFrame,
winddir_col: str = "winddir",
windspeed_col: str = "windspeed",
) -> go.Figure:
"""
Create a wind rose plot using polar bar chart.
Parameters:
df: DataFrame with wind direction and speed.
winddir_col: Wind direction column (0-360 degrees).
windspeed_col: Wind speed column.
Returns:
plotly.graph_objects.Figure
"""
if winddir_col not in df.columns or windspeed_col not in df.columns:
return blank_figure()
valid = df[[winddir_col, windspeed_col]].dropna()
if len(valid) == 0:
return blank_figure()
# Bin wind direction into 16 sectors
n_sectors = 16
sector_width = 360 / n_sectors
sectors = np.arange(0, 360, sector_width)
sector_labels = ["N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE",
"S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW"]
mean_speeds = []
for i, start in enumerate(sectors):
end = start + sector_width
mask = (valid[winddir_col] >= start) & (valid[winddir_col] < end)
if mask.any():
mean_speeds.append(valid.loc[mask, windspeed_col].mean())
else:
mean_speeds.append(0)
fig = go.Figure()
fig.add_trace(
go.Barpolar(
r=mean_speeds,
theta=sector_labels,
name="Wind Speed (m/s)",
marker_color="#3498db",
marker_line_color="#2980b9",
opacity=0.7,
)
)
fig.update_layout(
title="Wind Rose",
polar={
"radialaxis": {"title": "Wind Speed (m/s)", "visible": True},
"angularaxis": {"direction": "clockwise", "rotation": 90},
},
margin={"l": 0, "r": 0, "t": 40, "b": 0},
)
return fig
def outliers(
df: pd.DataFrame, column: str, name: str = ""
) -> go.Figure | None:
"""
Visualise outlier detection using IQR method.
Parameters:
df: DataFrame.
column: Column to check for outliers.
name: Dataset name for title.
Returns:
plotly.graph_objects.Figure or None
"""
if column not in df.columns:
return None
valid = df[column].dropna()
if len(valid) == 0:
return None
q1 = valid.quantile(0.25)
q3 = valid.quantile(0.75)
iqr = q3 - q1
fence_low = q1 - 3 * iqr
fence_high = q3 + 3 * iqr
outliers_mask = (valid < fence_low) | (valid > fence_high)
fig = go.Figure()
fig.add_trace(
go.Box(
y=valid,
name=column,
boxpoints="outliers",
marker={"color": "#e74c3c", "size": 4},
line={"color": "#2c3e50"},
)
)
title = f"Outlier Detection – {column}"
if name:
title += f" ({name})"
fig.update_layout(
title=title,
yaxis_title=column,
margin={"l": 0, "r": 0, "t": 40, "b": 0},
)
return fig
def contour_krig(krig_variables: dict) -> go.Figure:
"""
Create a contour plot of the kriging interpolation field.
Parameters:
krig_variables: Dictionary containing 'xx', 'yy', 'field', and 'gas'.
Returns:
plotly.graph_objects.Figure
"""
xx = krig_variables.get("xx")
yy = krig_variables.get("yy")
field = krig_variables.get("field")
gas = krig_variables.get("gas", "gas")
if xx is None or yy is None or field is None:
return blank_figure()
# np.mgrid produces shape (x_nodes, y_nodes).
# xx varies along axis 0 → xx[:, 0] gives unique x values
# yy varies along axis 1 → yy[0, :] gives unique y values
# field[x][y] needs transpose → field.T for plotly's z[y][x]
x_1d = xx[:, 0] if xx.ndim > 1 else xx.flatten()
y_1d = yy[0, :] if yy.ndim > 1 else yy.flatten()
fig = go.Figure()
fig.add_trace(
go.Contour(
z=field.T,
x=x_1d,
y=y_1d,
colorscale="RdBu_r",
contours={
"coloring": "fill",
"showlabels": True,
"labelfont": {"size": 10, "color": "#333"},
},
colorbar={"title": f"{gas.upper()} Flux (kg/h/m²)"},
)
)
fig.update_layout(
title=f"Kriging Interpolation – {gas.upper()} Flux Contour",
xaxis_title="Distance along plane (m)",
yaxis_title="Height ATO (m)",
margin={"l": 0, "r": 0, "t": 40, "b": 0},
)
return fig
def heatmap_krig(krig_variables: dict) -> go.Figure:
"""
Create a heatmap of the kriging grid with overlaid data points.
Parameters:
krig_variables: Dictionary with kriging output.
Returns:
plotly.graph_objects.Figure
"""
xx = krig_variables.get("xx")
yy = krig_variables.get("yy")
field = krig_variables.get("field")
gas = krig_variables.get("gas", "gas")
if xx is None or yy is None or field is None:
return blank_figure()
# np.mgrid produces shape (x_nodes, y_nodes).
# xx varies along axis 0 → xx[:, 0] gives unique x values
# yy varies along axis 1 → yy[0, :] gives unique y values
# field[x][y] needs transpose → field.T for plotly's z[y][x]
x_1d = xx[:, 0] if xx.ndim > 1 else xx.flatten()
y_1d = yy[0, :] if yy.ndim > 1 else yy.flatten()
fig = go.Figure()
fig.add_trace(
go.Heatmap(
z=field.T,
x=x_1d,
y=y_1d,
colorscale="RdBu_r",
colorbar={"title": f"{gas.upper()} Flux (kg/h/m²)"},
zmid=0,
)
)
fig.update_layout(
title=f"Kriging Grid – {gas.upper()} Flux Heatmap",
xaxis_title="Distance along plane (m)",
yaxis_title="Height ATO (m)",
margin={"l": 0, "r": 0, "t": 40, "b": 0},
)
return fig
def semivariogram_plot(semivariogram, gas: str = "") -> go.Figure:
"""
Create a semivariogram plot from a scikit-gstat Variogram object.
Parameters:
semivariogram: scikit-gstat Variogram object.
gas: Gas name for title.
Returns:
plotly.graph_objects.Figure
"""
try:
bins = semivariogram.bins
experimental = semivariogram.experimental
fig = go.Figure()
# Experimental semivariogram points
fig.add_trace(
go.Scatter(
x=bins,
y=experimental,
mode="markers",
name="Experimental",
marker={"size": 8, "color": "#3498db"},
)
)
# Fitted model line
if hasattr(semivariogram, "model") and semivariogram.model is not None:
x_line = np.linspace(0, bins.max(), 100)
y_line = semivariogram.model(x_line)
fig.add_trace(
go.Scatter(
x=x_line,
y=y_line,
mode="lines",
name=f"Fitted ({semivariogram.model.__class__.__name__})",
line={"color": "#e74c3c", "width": 2},
)
)
title = "Semivariogram"
if gas:
title += f" – {gas.upper()}"
fig.update_layout(
title=title,
xaxis_title="Lag Distance (m)",
yaxis_title="Semivariance",
margin={"l": 0, "r": 0, "t": 40, "b": 0},
)
return fig
except Exception:
return blank_figure()
def create_kml_file(*args, **kwargs) -> None:
"""KML file generation – not yet implemented."""
return None return None