504 lines
20 KiB
Python
504 lines
20 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Figuras para log_2026-10-30.txt (MicrotrackingFer v1.1.1-ota).
|
|
|
|
Dos versiones por figura:
|
|
full : todo el dia 10:00-18:21 (incluye horas basura >= 14:00, deriva F5)
|
|
pre14 : filtrado, solo 10:00-14:00 (rastreo sano, post-correccion F5)
|
|
|
|
Calibracion (PCB V2, ver docker_tracker_hub/app/config.py):
|
|
ADC122C04 24-bit, Vref 2.048 V.
|
|
C = canal1 fotocorriente (shunt 0.260 ohm = 260 mOhm, gain 20) -> mA
|
|
P = canal2 piranometro (24-bit signed, escala 0.2 mV por W/m2) -> W/m2
|
|
V = canal3 Voc
|
|
|
|
Salida: figures/2026-10-30/*.png + *.pdf
|
|
Uso: py -3.12 figures_2026_10_30.py
|
|
"""
|
|
import io
|
|
import os
|
|
import re
|
|
from datetime import datetime, timedelta as _timedelta
|
|
|
|
import numpy as np
|
|
import matplotlib
|
|
matplotlib.use("Agg")
|
|
import matplotlib.pyplot as plt
|
|
import matplotlib.dates as mdates
|
|
|
|
BASE = r"G:\Mi unidad\microtracking\MicrotrackingFer"
|
|
LOG = os.path.join(BASE, "logs_nuevos", "log_2026-10-30.txt")
|
|
OUT = os.path.join(BASE, "figures", "2026-10-30")
|
|
os.makedirs(OUT, exist_ok=True)
|
|
|
|
CUT = datetime(2026, 10, 30, 14, 0, 0) # "horas basura" empiezan a las 2 pm
|
|
CUT12 = datetime(2026, 10, 30, 12, 0, 0) # ventana limpia: post-fix F5, pre-deriva
|
|
|
|
ADC_MAX = 8388607.0
|
|
GAIN_SHUNT = 20.0
|
|
R_SHUNT = 0.260 # PCB V2 (260 mOhm)
|
|
PYRANO_DIV = 0.2 # W/m2 por (mV/1000) en el canal del piranometro
|
|
RAW2MA = ADC_MAX * GAIN_SHUNT * R_SHUNT / 2048.0 # counts de C (o dC) -> mA
|
|
|
|
|
|
def isc_ma(c_raw):
|
|
vsh_mV = c_raw / ADC_MAX * 2048.0 / GAIN_SHUNT
|
|
return (vsh_mV / 1000.0) / R_SHUNT * 1000.0
|
|
|
|
|
|
def g_wm2(p_signed):
|
|
return p_signed / ADC_MAX * 2048.0 / PYRANO_DIV
|
|
|
|
|
|
def sign24(raw):
|
|
raw &= 0xFFFFFF
|
|
if raw & 0x00800000:
|
|
raw -= 0x1000000
|
|
return raw
|
|
|
|
|
|
RE_MAIN = re.compile(
|
|
r"LOG:\[(?P<t>[^\]]+)\]\s*X:(?P<x>-?\d+),Y:(?P<y>-?\d+),C:(?P<c>-?\d+),"
|
|
r"V:(?P<v>-?\d+),P:(?P<p>-?\d+),S:(?P<s>-?\d+),er:(?P<er>-?\d+)")
|
|
RE_TEL = re.compile(
|
|
r"LOG:\[(?P<t>[^\]]+)\]\s*TEL: t=(?P<tsec>\d+)s, steps=(?P<steps>\d+)\(\+?(?P<delta>\d+)\)")
|
|
RE_FT = re.compile(
|
|
r"LOG:\[(?P<t>[^\]]+)\]\s*FineTune: dC=(?P<dc>\d+), dynE=\d+, e=\d+, near=(?P<near>\d+)")
|
|
RE_F5 = re.compile(
|
|
r"LOG:\[(?P<t>[^\]]+)\]\s*F5: Deriva detectada (?P<drift>\d+) pasos "
|
|
r"\(dX=(-?\d+), dY=(-?\d+)\), re-alineando a \((-?\d+), (-?\d+)\)")
|
|
RE_CLOUD = re.compile(r"LOG:\[(?P<t>[^\]]+)\]\s*\[CLOUD\] (?P<msg>.+)")
|
|
|
|
|
|
def parse_ts(s):
|
|
s = s.strip()
|
|
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
|
|
try:
|
|
return datetime.strptime(s, fmt)
|
|
except ValueError:
|
|
pass
|
|
return None
|
|
|
|
|
|
def parse_log(path):
|
|
M, T, F, F5, CL = [], [], [], [], []
|
|
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
|
for line in f:
|
|
m = RE_MAIN.search(line)
|
|
if m:
|
|
ts = parse_ts(m.group("t"))
|
|
if ts:
|
|
M.append((ts, int(m.group("x")), int(m.group("y")),
|
|
int(m.group("c")), int(m.group("v")),
|
|
sign24(int(m.group("p"))), int(m.group("er"))))
|
|
continue
|
|
m = RE_TEL.search(line)
|
|
if m:
|
|
ts = parse_ts(m.group("t"))
|
|
if ts:
|
|
T.append((ts, int(m.group("steps"))))
|
|
continue
|
|
m = RE_FT.search(line)
|
|
if m:
|
|
ts = parse_ts(m.group("t"))
|
|
if ts:
|
|
F.append((ts, int(m.group("dc")), int(m.group("near"))))
|
|
continue
|
|
m = RE_F5.search(line)
|
|
if m:
|
|
ts = parse_ts(m.group("t"))
|
|
if ts:
|
|
F5.append((ts, int(m.group("drift"))))
|
|
continue
|
|
m = RE_CLOUD.search(line)
|
|
if m:
|
|
ts = parse_ts(m.group("t"))
|
|
if ts:
|
|
CL.append((ts, m.group("msg").strip()))
|
|
return M, T, F, F5, CL
|
|
|
|
|
|
def unwrap_steps(tel):
|
|
"""El contador de steps (TEL) se reinicia en cada reboot/OTA:
|
|
10:27:58->10:30:08 y 11:43:29->11:45:37. Lo desenrolla a una serie continua."""
|
|
out = []
|
|
off = 0
|
|
prev = None
|
|
for (ts, st) in tel:
|
|
if prev is not None and st < prev:
|
|
off += prev
|
|
out.append((ts, st + off))
|
|
prev = st
|
|
return out
|
|
|
|
|
|
def to_arrays(M):
|
|
ts = np.array([r[0] for r in M], dtype=object)
|
|
X = np.array([r[1] for r in M])
|
|
Y = np.array([r[2] for r in M])
|
|
C = np.array([r[3] for r in M], dtype=float)
|
|
V = np.array([r[4] for r in M], dtype=float)
|
|
P = np.array([r[5] for r in M], dtype=float)
|
|
er = np.array([r[6] for r in M])
|
|
return dict(t=ts, X=X, Y=Y, C=C, V=V, P=np.maximum(0, P), er=er)
|
|
|
|
|
|
def subset(L, lo=None, hi=None):
|
|
if lo is None and hi is None:
|
|
return L
|
|
sel = np.array([(t >= lo if lo is not None else True) and (t < hi if hi is not None else True)
|
|
for t in L["t"]])
|
|
return {k: L[k][sel] for k in ("t", "X", "Y", "C", "V", "P", "er")}
|
|
|
|
|
|
def reversals(arr):
|
|
d = np.diff(arr.astype(float))
|
|
s = np.sign(d)
|
|
s = s[s != 0]
|
|
if len(s) < 2:
|
|
return 0
|
|
return int((s[1:] != s[:-1]).sum())
|
|
|
|
|
|
def metrics(L, tel, f5):
|
|
n = len(L["t"])
|
|
out = {"n": n}
|
|
if n == 0:
|
|
return out
|
|
t0, t1 = L["t"][0], L["t"][-1]
|
|
mins = max((t1 - t0).total_seconds() / 60.0, 1e-9)
|
|
out["t0"], out["t1"], out["mins"] = t0, t1, mins
|
|
out["Cpeak_mA"] = isc_ma(L["C"].max())
|
|
out["Ppeak_W"] = g_wm2(L["P"].max())
|
|
out["distinct"] = len(set(zip(L["X"].tolist(), L["Y"].tolist())))
|
|
# on-target: C >= 80% del pico del dia (con luz)
|
|
cmax = L["C"].max()
|
|
p_w = g_wm2(L["P"])
|
|
on = (L["C"] >= 0.8 * cmax) & (p_w > 50.0)
|
|
out["ontarget_pct"] = 100.0 * on.mean()
|
|
ft = [(ts, dc) for (ts, dc, _nr) in FT if L["t"][0] <= ts <= t1]
|
|
out["ndC"] = float(np.mean([dc for _, dc in ft])) / RAW2MA if ft else float("nan") # raw->mA
|
|
out["revX"] = reversals(L["X"])
|
|
out["revY"] = reversals(L["Y"])
|
|
telw = [(ts, st) for (ts, st) in tel if L["t"][0] <= ts <= t1]
|
|
# delta de la ventana (la serie es acumulada desde el boot)
|
|
out["steps_total"] = (telw[-1][1] - telw[0][1]) if len(telw) > 1 else 0
|
|
out["steps_min"] = out["steps_total"] / mins
|
|
f5w = [(ts, d) for (ts, d) in f5 if L["t"][0] <= ts <= t1]
|
|
out["nF5"] = len(f5w)
|
|
out["drift_steps"] = sum(d for _, d in f5w)
|
|
gaps = [int((b - a).total_seconds()) for (a, b) in zip(L["t"][:-1], L["t"][1:]) if (b - a).total_seconds() > 15]
|
|
out["gaps"] = gaps
|
|
return out
|
|
|
|
|
|
M, TEL, FT, F5, CLOUD = parse_log(LOG)
|
|
TEL = unwrap_steps(TEL)
|
|
print("Muestras: %d TEL: %d FineTune: %d F5: %d CLOUD: %d" % (len(M), len(TEL), len(FT), len(F5), len(CLOUD)))
|
|
|
|
L_FULL = to_arrays(M)
|
|
L_PRE = subset(L_FULL, hi=CUT)
|
|
L_NOON = subset(L_FULL, lo=CUT12, hi=CUT)
|
|
for name, L in (("FULL ", L_FULL), ("PRE14", L_PRE), ("NOON ", L_NOON)):
|
|
m = metrics(L, TEL, F5)
|
|
print("[%s] %s -> %s (%.0f min) n=%d pos.distinct=%d Cpeak=%.0f mA Ppeak=%.0f W/m2" % (
|
|
name, m["t0"].strftime("%H:%M:%S"), m["t1"].strftime("%H:%M:%S"), m["mins"],
|
|
m["n"], m["distinct"], m["Cpeak_mA"], m["Ppeak_W"]))
|
|
print(" on-target=%.1f%% dC medio=%.2f mA revX=%d revY=%d steps=%d (%.1f/min) F5=%d (drift %d pasos) gaps>15s=%d" % (
|
|
m["ontarget_pct"], m["ndC"], m["revX"], m["revY"], m["steps_total"], m["steps_min"],
|
|
m["nF5"], m["drift_steps"], len(m["gaps"])))
|
|
m_full = metrics(L_FULL, TEL, F5)
|
|
m_pre = metrics(L_PRE, TEL, F5)
|
|
m_noon = metrics(L_NOON, TEL, F5)
|
|
|
|
|
|
def savefig(fig, base):
|
|
for ext in ("pdf", "png"):
|
|
buf = io.BytesIO()
|
|
fig.savefig(buf, format=ext)
|
|
dest = os.path.join(OUT, base + "." + ext)
|
|
if os.path.exists(dest):
|
|
try:
|
|
os.remove(dest)
|
|
except OSError:
|
|
pass
|
|
with open(dest, "wb") as f:
|
|
f.write(buf.getvalue())
|
|
|
|
|
|
def fmt_ax(ax, fmt="%H:%M"):
|
|
ax.xaxis.set_major_formatter(mdates.DateFormatter(fmt))
|
|
ax.xaxis.set_major_locator(mdates.AutoDateLocator())
|
|
ax.grid(alpha=0.3)
|
|
|
|
|
|
# ---------------------------------------------------------------- heatmap
|
|
def build_heatmap(tag, label, L, tel, f5):
|
|
xs, ys, c_mA, p_w = L["X"], L["Y"], isc_ma(L["C"]), g_wm2(L["P"])
|
|
xlo, xhi = 0, int(np.ceil(xs.max() / 5000.0) * 5000) + 5000
|
|
ylo, yhi = 0, int(np.ceil(ys.max() / 5000.0) * 5000) + 5000
|
|
xbins = np.linspace(xlo, xhi, 61)
|
|
ybins = np.linspace(ylo, yhi, 46)
|
|
|
|
def bin_max(x, y, vals):
|
|
xi = np.clip(np.searchsorted(xbins, x, side="right") - 1, 0, len(xbins) - 2)
|
|
yi = np.clip(np.searchsorted(ybins, y, side="right") - 1, 0, len(ybins) - 2)
|
|
grid = np.full((len(ybins) - 1, len(xbins) - 1), -np.inf)
|
|
np.maximum.at(grid, (yi, xi), vals)
|
|
grid[grid == -np.inf] = np.nan
|
|
return grid
|
|
|
|
maxC = bin_max(xs, ys, c_mA)
|
|
maxP = bin_max(xs, ys, p_w)
|
|
nC = np.nan_to_num(maxC / np.nanmax(maxC))
|
|
nP = np.nan_to_num(maxP / np.nanmax(maxP))
|
|
co = np.ma.masked_invalid(np.minimum(nC, nP))
|
|
|
|
xc = (xbins[:-1] + xbins[1:]) / 2
|
|
yc = (ybins[:-1] + ybins[1:]) / 2
|
|
pi, pj = np.unravel_index(np.nanargmax(co), co.shape)
|
|
co_x, co_y = xc[pj], yc[pi]
|
|
|
|
w = np.where(co > 0.5 * np.nanmax(co), co, 0.0)
|
|
if w.sum() > 0:
|
|
com_x = np.nansum(w * xc[None, :]) / np.nansum(w)
|
|
com_y = np.nansum(w * yc[:, None]) / np.nansum(w)
|
|
else:
|
|
com_x = com_y = np.nan
|
|
|
|
fig, axes = plt.subplots(1, 3, figsize=(20, 6.4), sharex=True, sharey=True)
|
|
|
|
def draw_heat(ax, grid, title, cunit):
|
|
im = ax.imshow(np.ma.masked_invalid(grid),
|
|
extent=[xbins[0], xbins[-1], ybins[0], ybins[-1]],
|
|
origin="lower", aspect="auto", cmap="inferno",
|
|
norm=matplotlib.colors.Normalize(vmin=0,
|
|
vmax=np.nanmax(grid) if np.isfinite(np.nanmax(grid)) else 1))
|
|
ax.set_title(title, fontsize=11)
|
|
ax.set_xlabel("X [steps]")
|
|
ax.set_ylabel("Y [steps]")
|
|
fig.colorbar(im, ax=ax, label=cunit)
|
|
|
|
draw_heat(axes[0], maxC, "Photocurrent (maximum per bin)", "Isc [mA]")
|
|
draw_heat(axes[1], maxP, "Irradiance (maximum per bin)", "G [W/m$^2$]")
|
|
draw_heat(axes[2], co, "Normalized overlap min(C*, P*)", "Normalized overlap")
|
|
|
|
ci, cj = np.unravel_index(np.nanargmax(maxC), maxC.shape)
|
|
axes[0].plot(xc[cj], yc[ci], "w*", ms=16, mec="white", mew=1.2, zorder=5)
|
|
pi_, pj_ = np.unravel_index(np.nanargmax(maxP), maxP.shape)
|
|
axes[1].plot(xc[pj_], yc[pi_], "w*", ms=16, mec="white", mew=1.2, zorder=5)
|
|
|
|
axes[2].plot(xs, ys, "-", color="cyan", lw=0.25, alpha=0.35, zorder=2, label="Recorded trajectory")
|
|
axes[2].plot(co_x, co_y, "o", ms=16, mfc="none", mec="lime", mew=2.2, zorder=6, label="Overlap maximum")
|
|
axes[2].plot(com_x, com_y, "x", ms=14, color="yellow", mew=2.4, zorder=7, label="Weighted centroid")
|
|
axes[2].plot(xc[cj], yc[ci], "+", ms=14, color="red", mew=2.0, zorder=7, label="Current maximum")
|
|
axes[2].plot(xc[pj_], yc[pi_], "+", ms=14, color="orange", mew=2.0, zorder=7, label="Irradiance maximum")
|
|
axes[2].legend(loc="upper left", fontsize=8, framealpha=0.8)
|
|
|
|
fig.suptitle("Stage-coordinate maps - %s (%d samples, %s)" % (label, len(L["t"]),
|
|
L["t"][0].strftime("%d/%m %H:%M") + "-" + L["t"][-1].strftime("%H:%M")), fontsize=13, y=1.00)
|
|
fig.tight_layout()
|
|
savefig(fig, tag + "_heatmap")
|
|
plt.close(fig)
|
|
print(" co-pico: X~%.0f Y~%.0f | centro de masa: X~%.0f Y~%.0f" % (co_x, co_y, com_x, com_y))
|
|
|
|
|
|
# ------------------------------------------------------- tracking figures
|
|
def build_tracking(tag, label, L, tel, f5, show_cut):
|
|
t = L["t"]
|
|
tx = mdates.date2num([t[i] for i in range(len(t))])
|
|
c_mA = isc_ma(L["C"])
|
|
p_w = g_wm2(L["P"])
|
|
|
|
fig, axes = plt.subplots(2, 2, figsize=(16, 9), dpi=200)
|
|
fig.suptitle("Tracking performance - %s" % label, fontsize=14, fontweight="bold")
|
|
|
|
ax = axes[0, 0]
|
|
ax.plot(tx, p_w, color="#ff7f0e", lw=0.6, label="P [W/m$^2$]")
|
|
ax.set_xlabel("Time"); ax.set_ylabel("G [W/m$^2$]", color="#ff7f0e")
|
|
ax2 = ax.twinx()
|
|
ax2.plot(tx, c_mA, color="#d62728", lw=0.6, label="C [mA]")
|
|
ax2.set_ylabel("C [mA]", color="#d62728")
|
|
for (fts, _d) in f5:
|
|
if t[0] <= fts <= t[-1]:
|
|
ax.axvline(mdates.date2num(fts), color="red", ls="--", lw=0.8, alpha=0.6)
|
|
ax.text(mdates.date2num(t[0]) + (tx[-1] - tx[0]) * 0.01, ax2.get_ylim()[1] * 0.9, "F5 (drift)", color="red", fontsize=8)
|
|
if show_cut:
|
|
ax.axvline(mdates.date2num(CUT), color="purple", ls="-", lw=1.4)
|
|
ax.text(mdates.date2num(CUT) + (tx[-1] - tx[0]) * 0.01, ax2.get_ylim()[1] * 0.98, "14:00 (degraded tracking)",
|
|
color="purple", fontsize=8, rotation=90, va="top")
|
|
fmt_ax(ax)
|
|
h1, l1 = ax.get_legend_handles_labels(); h2, l2 = ax2.get_legend_handles_labels()
|
|
ax.legend(h1 + h2, l1 + l2, fontsize=8, loc="upper right")
|
|
ax.set_title("Irradiance and photocurrent (red dashed: F5 events)")
|
|
|
|
ax = axes[0, 1]
|
|
ftw = [(ts, dc) for (ts, dc, _nr) in FT if t[0] <= ts <= t[-1]]
|
|
if ftw:
|
|
ftx = mdates.date2num([x[0] for x in ftw])
|
|
ax.plot(ftx, [x[1] for x in ftw], color="tab:blue", lw=0.5)
|
|
ax.set_xlabel("Time"); ax.set_ylabel("dC [ADC counts]")
|
|
if show_cut:
|
|
ax.axvline(mdates.date2num(CUT), color="purple", ls="-", lw=1.4)
|
|
fmt_ax(ax)
|
|
ax.set_title("Fine-tune current-change diagnostic")
|
|
|
|
ax = axes[1, 0]
|
|
ax.step(tx, L["er"], where="post", color="tab:purple", lw=1.2)
|
|
ax.set_xlabel("Time"); ax.set_ylabel("er (adaptive threshold)")
|
|
if show_cut:
|
|
ax.axvline(mdates.date2num(CUT), color="purple", ls="-", lw=1.4)
|
|
fmt_ax(ax)
|
|
ax.set_title("Adaptive threshold er")
|
|
|
|
ax = axes[1, 1]
|
|
wins = 15 * 60.0
|
|
tsec = np.array([(x - t[0]).total_seconds() for x in t])
|
|
cmax = L["C"].max()
|
|
on = ((L["C"] >= 0.8 * cmax) & (g_wm2(L["P"]) > 50.0)).astype(float)
|
|
wi = (tsec // wins).astype(int)
|
|
frac = []
|
|
for w in range(wi.max() + 1):
|
|
sel = wi == w
|
|
frac.append(on[sel].mean() if sel.any() else np.nan)
|
|
wmid = [t[0] + _timedelta(seconds=(w + 0.5) * wins) for w in range(wi.max() + 1)]
|
|
ax.step(mdates.date2num(wmid), frac, where="post", color="tab:green", lw=1.4)
|
|
ax.set_ylim(-0.05, 1.05)
|
|
ax.set_ylabel("On-target sample fraction")
|
|
ax.set_xlabel("Time")
|
|
if show_cut:
|
|
ax.axvline(mdates.date2num(CUT), color="purple", ls="-", lw=1.4)
|
|
ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M"))
|
|
ax.xaxis.set_major_locator(mdates.AutoDateLocator())
|
|
ax.grid(alpha=0.3)
|
|
ax.set_title("On-target occupancy (15-min bins)")
|
|
|
|
fig.tight_layout(rect=[0, 0, 1, 0.96])
|
|
savefig(fig, tag + "_tracking")
|
|
plt.close(fig)
|
|
|
|
|
|
# --------------------------------------------------------- energy figures
|
|
def build_energy(tag, label, L, tel, f5, show_cut):
|
|
t = L["t"]
|
|
tx = mdates.date2num([t[i] for i in range(len(t))])
|
|
telw = [(ts, st) for (ts, st) in TEL if t[0] <= ts <= t[-1]]
|
|
f5w = [(ts, d) for (ts, d) in F5 if t[0] <= ts <= t[-1]]
|
|
m = metrics(L, tel, f5)
|
|
|
|
fig, axes = plt.subplots(2, 2, figsize=(16, 9), dpi=200)
|
|
fig.suptitle("Motion activity - %s" % label, fontsize=14, fontweight="bold")
|
|
|
|
ax = axes[0, 0]
|
|
if telw:
|
|
ax.plot(mdates.date2num([x[0] for x in telw]), [x[1] for x in telw], color="tab:blue", lw=1.2)
|
|
ymax = max(x[1] for x in telw)
|
|
else:
|
|
ymax = 1
|
|
for rb, rbl in ((datetime(2026, 10, 30, 10, 30, 8), "OTA reboot"),
|
|
(datetime(2026, 10, 30, 11, 45, 37), "OTA reboot (F5 update)")):
|
|
if t[0] <= rb <= t[-1]:
|
|
ax.axvline(mdates.date2num(rb), color="green", ls=":", lw=1.2)
|
|
ax.text(mdates.date2num(rb), ymax * 0.03, rbl, color="green", fontsize=7, rotation=90, va="bottom")
|
|
ax.set_xlabel("Time"); ax.set_ylabel("Cumulative steps")
|
|
if show_cut:
|
|
ax.axvline(mdates.date2num(CUT), color="purple", ls="-", lw=1.4)
|
|
fmt_ax(ax)
|
|
ax.set_title("Cumulative fine-tune steps (TEL; reboot-adjusted)")
|
|
|
|
ax = axes[0, 1]
|
|
if len(telw) > 1:
|
|
tdx = np.diff(mdates.date2num([x[0] for x in telw])) * 86400.0
|
|
dsteps = np.diff([x[1] for x in telw])
|
|
rate = dsteps / (tdx / 60.0)
|
|
tm = mdates.date2num([x[0] for x in telw][1:])
|
|
ax.plot(tm, rate, color="tab:orange", lw=0.8)
|
|
med = np.median(rate)
|
|
ax.axhline(med, color="gray", ls=":", lw=1)
|
|
ax.text(tm[0], med * 1.05, "Median %.0f steps/min" % med, fontsize=8, color="gray")
|
|
ax.set_xlabel("Time"); ax.set_ylabel("steps/min")
|
|
if show_cut:
|
|
ax.axvline(mdates.date2num(CUT), color="purple", ls="-", lw=1.4)
|
|
fmt_ax(ax)
|
|
ax.set_title("Movement rate")
|
|
|
|
ax = axes[1, 0]
|
|
if f5w:
|
|
fx = mdates.date2num([x[0] for x in f5w])
|
|
ax.bar(fx, [x[1] for x in f5w], width=0.004, color="red", alpha=0.75)
|
|
ax.set_yscale("log")
|
|
ax.set_xlabel("Time"); ax.set_ylabel("Corrected drift [steps] (log)")
|
|
if show_cut:
|
|
ax.axvline(mdates.date2num(CUT), color="purple", ls="-", lw=1.4)
|
|
fmt_ax(ax)
|
|
ax.set_title("F5 drift corrections (%d events, %d steps)" % (m["nF5"], m["drift_steps"]))
|
|
|
|
ax = axes[1, 1]
|
|
ax.axis("off")
|
|
txt = [
|
|
label, "",
|
|
"Window: %s -> %s (%.0f min)" % (m["t0"].strftime("%H:%M:%S"), m["t1"].strftime("%H:%M:%S"), m["mins"]),
|
|
"X/Y samples: %d (distinct positions: %d)" % (m["n"], m["distinct"]),
|
|
"Peak current: %.0f mA | Peak irradiance: %.0f W/m2" % (m["Cpeak_mA"], m["Ppeak_W"]),
|
|
"On-target: %.1f%% (C >= 80%% of window peak; G > 50 W/m2)" % m["ontarget_pct"],
|
|
"Mean dC: %.1f mA" % m["ndC"],
|
|
"Reversals: X=%d Y=%d" % (m["revX"], m["revY"]),
|
|
"Total steps: %d (%.1f steps/min)" % (m["steps_total"], m["steps_min"]),
|
|
"F5 events: %d (corrected drift: %d steps = %.1f%% of counted steps)" % (
|
|
m["nF5"], m["drift_steps"], 100.0 * m["drift_steps"] / max(m["steps_total"], 1)),
|
|
"Gaps > 15 s: %d" % len(m["gaps"]),
|
|
]
|
|
ax.text(0.0, 1.0, "\n".join(txt), family="monospace", fontsize=10, va="top",
|
|
bbox=dict(boxstyle="round,pad=0.6", fc="#f7f7f7", ec="gray"))
|
|
|
|
fig.tight_layout(rect=[0, 0, 1, 0.96])
|
|
savefig(fig, tag + "_energy")
|
|
plt.close(fig)
|
|
|
|
|
|
# ---------------------------------------------------- comparacion (corroborar)
|
|
def build_compare():
|
|
rows = [
|
|
("Total counted steps", m_full["steps_total"], m_pre["steps_total"], m_noon["steps_total"]),
|
|
("steps/min", m_full["steps_min"], m_pre["steps_min"], m_noon["steps_min"]),
|
|
("F5 events", m_full["nF5"], m_pre["nF5"], m_noon["nF5"]),
|
|
("Corrected drift [steps]", m_full["drift_steps"], m_pre["drift_steps"], m_noon["drift_steps"]),
|
|
("On-target [%]", m_full["ontarget_pct"], m_pre["ontarget_pct"], m_noon["ontarget_pct"]),
|
|
("Mean dC [mA]", m_full["ndC"], m_pre["ndC"], m_noon["ndC"]),
|
|
("Direction reversals (X+Y)", m_full["revX"] + m_full["revY"], m_pre["revX"] + m_pre["revY"], m_noon["revX"] + m_noon["revY"]),
|
|
("Distinct positions", m_full["distinct"], m_pre["distinct"], m_noon["distinct"]),
|
|
]
|
|
fig, axes = plt.subplots(2, 4, figsize=(18, 8.5), dpi=200)
|
|
fig.suptitle("Window comparison: full record (10:00-18:21), pre-14:00 (10:00-14:00), midday (12:00-14:00)",
|
|
fontsize=14, fontweight="bold")
|
|
for ax, (name, vfull, vpre, vnoon) in zip(axes.ravel(), rows):
|
|
vals = (vfull, vpre, vnoon)
|
|
bars = ax.bar([0, 1, 2], vals, color=["#d62728", "#ff7f0e", "#2ca02c"], alpha=0.85)
|
|
ax.set_xticks([0, 1, 2])
|
|
ax.set_xticklabels(["Full", "Pre-14:00", "Midday"])
|
|
ax.set_title(name, fontsize=10)
|
|
ax.grid(axis="y", alpha=0.3)
|
|
for b, v in zip(bars, vals):
|
|
ax.annotate("%.3g" % v, (b.get_x() + b.get_width() / 2, v),
|
|
ha="center", va="bottom", fontsize=8)
|
|
ax.set_yscale("symlog" if max(abs(v) for v in vals) > 1e4 else "linear")
|
|
fig.tight_layout(rect=[0, 0, 1, 0.94])
|
|
savefig(fig, "compare_full_vs_pre14_vs_noon")
|
|
plt.close(fig)
|
|
|
|
|
|
VERSIONS = [
|
|
("2026-10-30_full", "Full record, 30 October 2026 (10:00-18:21)", L_FULL, True),
|
|
("2026-10-30_pre14", "Pre-14:00 window, 30 October 2026 (10:00-14:00)", L_PRE, False),
|
|
("2026-10-30_noon", "Midday window, 30 October 2026 (12:00-14:00)", L_NOON, False),
|
|
]
|
|
|
|
for tag, label, L, show_cut in VERSIONS:
|
|
print("Generando", tag, "...")
|
|
build_heatmap(tag, label, L, TEL, F5)
|
|
build_tracking(tag, label, L, TEL, F5, show_cut)
|
|
build_energy(tag, label, L, TEL, F5, show_cut)
|
|
|
|
print("Generando comparacion ...")
|
|
build_compare()
|
|
print("DONE ->", OUT)
|