def load_rmm_from_file(path: str) -> pd.DataFrame:
"""
Parse a local BoM RMM text file.
Returns DataFrame with ['rmm1','rmm2','phase','amplitude'] indexed by datetime.
Skips headers and drops missing (1.E36 or 999).
"""
rows = []
with open(path, "r", encoding="utf-8", errors="ignore") as f:
for raw in f:
line = raw.strip()
if not line:
continue
c0 = line[0]
# Keep only lines beginning with a year (numeric or leading minus)
if not (c0.isdigit() or (c0 == "-" and len(line) > 1 and line[1].isdigit())):
continue
parts = line.split()
if len(parts) < 7:
continue
try:
y = int(parts[0]); m = int(parts[1]); d = int(parts[2])
r1 = float(parts[3]); r2 = float(parts[4]); ph = int(parts[5]); amp = float(parts[6])
except ValueError:
continue
# Drop missing per header: 1.E36 or 999
if any(abs(v) >= 1e35 for v in (r1, r2, amp)) or any(v == 999 for v in (r1, r2, amp)):
continue
rows.append({
"date": datetime(y, m, d),
"rmm1": r1, "rmm2": r2, "phase": ph, "amplitude": amp
})
if not rows:
raise ValueError("No RMM rows parsed from file.")
return pd.DataFrame(rows).set_index("date").sort_index()
def plot_rmm_phase_space(df: pd.DataFrame, *, highlight_days: int = 60) -> go.Figure:
if df.empty:
raise ValueError("Empty RMM DataFrame provided.")
fig = go.Figure()
# Full track (muted)
fig.add_trace(go.Scatter(
x=df["rmm1"], y=df["rmm2"],
mode="lines", line=dict(color="rgba(150,150,150,0.35)", width=1),
name="Full track",
))
# Highlight recent segment
cutoff = df.index.max() - pd.Timedelta(days=highlight_days)
recent = df[df.index >= cutoff]
if len(recent) > 1:
t = (recent.index - recent.index.min()).days.astype(float)
fig.add_trace(go.Scatter(
x=recent["rmm1"], y=recent["rmm2"],
mode="lines+markers",
line=dict(color="#1f77b4", width=2),
marker=dict(size=6, color=t, colorscale="Viridis", showscale=True,
colorbar=dict(title="Days from start")),
name=f"Last {highlight_days} days",
))
# Unit circle (amplitude = 1)
theta = np.linspace(0, 2*np.pi, 361)
fig.add_trace(go.Scatter(
x=np.cos(theta), y=np.sin(theta),
mode="lines", line=dict(color="black", width=1, dash="dash"),
name="Amplitude = 1",
))
# Axes at zero
fig.add_shape(type="line", x0=-5, x1=5, y0=0, y1=0, line=dict(color="gray", width=1, dash="dot"))
fig.add_shape(type="line", x0=0, x1=0, y0=-5, y1=5, line=dict(color="gray", width=1, dash="dot"))
# Phase labels
labels = {
1: (0.35, 1.05), 2: (1.05, 0.35), 3: (1.05, -0.35), 4: (0.35, -1.05),
5: (-0.35, -1.05), 6: (-1.05, -0.35), 7: (-1.05, 0.35), 8: (-0.35, 1.05),
}
for ph, (x, y) in labels.items():
fig.add_annotation(x=x, y=y, text=str(ph), showarrow=False, font=dict(size=12))
# Latest point
latest = df.iloc[-1]
fig.add_trace(go.Scatter(
x=[latest["rmm1"]], y=[latest["rmm2"]],
mode="markers",
marker=dict(color="crimson", size=10, line=dict(color="white", width=1)),
name=f"Latest: {df.index[-1].date()} (phase {latest['phase']}, amp {latest['amplitude']:.2f})",
))
fig.update_layout(
title="RMM Phase Space (RMM2 vs RMM1)",
xaxis_title="RMM1", yaxis_title="RMM2",
xaxis=dict(constrain="domain", scaleanchor="y", scaleratio=1, range=[-4, 4], zeroline=False),
yaxis=dict(range=[-4, 4], zeroline=False),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left", x=0),
margin=dict(l=40, r=10, t=60, b=40),
hovermode="closest",
)
return fig
