351 lines
16 KiB
Python
351 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
# plot_logs.py — Generate candidate paper figures from the three Microtracking logs.
|
|
# Usage: py -3.12 plot_logs.py
|
|
import os, re, io
|
|
from datetime import datetime
|
|
from collections import defaultdict
|
|
import numpy as np
|
|
import matplotlib
|
|
matplotlib.use("Agg")
|
|
import matplotlib.pyplot as plt
|
|
import matplotlib.dates as mdates
|
|
|
|
BASE = r'c:\Users\fdavalos\OneDrive - up.edu.mx\microtracking\MicrotrackingFer'
|
|
OUT = os.path.join(BASE, 'figures')
|
|
os.makedirs(OUT, exist_ok=True)
|
|
|
|
LOGS = [
|
|
('21aug', os.path.join(BASE, 'full_log_21082025.txt'), '21/08/2025 (old firmware)'),
|
|
('14sep_A', os.path.join(BASE, 'log(3).txt'), '14/09/2025 capture A [log(3)]'),
|
|
('14sep_B', os.path.join(BASE, 'gyuggyhoy.txt'), '14/09/2025 capture B [gyuggyhoy]'),
|
|
]
|
|
LOG_COLORS = {'21aug': '#1f77b4', '14sep_A': '#2ca02c', '14sep_B': '#d62728'}
|
|
|
|
logRe = re.compile(r'^(?:LOG:)?\[([0-9\-/ :]+)\] X:(-?\d+),Y:(-?\d+),C:(-?\d+),V:(-?\d+),P:(-?\d+),S:(\d+),er:(-?\d+)\s*$')
|
|
ftRe = re.compile(r'^(?:LOG:)?\[([0-9\-/ :]+)\] FineTune: dC=(-?\d+), dynE=(-?\d+), e=(-?\d+), near=(\d+), Cmax=(-?\d+)\s*$')
|
|
posRe = re.compile(r'^POS:(-?\d+):(-?\d+)\s*$')
|
|
evRe = re.compile(r'^(?:LOG:)?\[([0-9\-/ :]+)\] (.+?)\s*$')
|
|
|
|
def parse_ts(s):
|
|
s = s.strip()
|
|
for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M', '%d/%m/%Y %H:%M:%S', '%d/%m/%Y %H:%M'):
|
|
try:
|
|
return datetime.strptime(s, fmt)
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
def parse_log(path):
|
|
t=[]; X=[]; Y=[]; C=[]; V=[]; P=[]; S=[]; er=[]
|
|
ft_t=[]; dC=[]; dynE=[]; e=[]; near=[]; Cmax=[]
|
|
pos=[]; events=[]
|
|
with open(path, 'r', encoding='utf-8', errors='replace') as f:
|
|
for raw in f:
|
|
line = raw.strip()
|
|
if not line:
|
|
continue
|
|
m = logRe.match(line)
|
|
if m:
|
|
ts = parse_ts(m.group(1))
|
|
if ts is None:
|
|
continue
|
|
t.append(ts); X.append(int(m.group(2))); Y.append(int(m.group(3)))
|
|
C.append(int(m.group(4))); V.append(int(m.group(5))); P.append(int(m.group(6)))
|
|
S.append(int(m.group(7))); er.append(int(m.group(8)))
|
|
continue
|
|
m = ftRe.match(line)
|
|
if m:
|
|
ts = parse_ts(m.group(1))
|
|
if ts is None:
|
|
continue
|
|
ft_t.append(ts); dC.append(int(m.group(2))); dynE.append(int(m.group(3)))
|
|
e.append(int(m.group(4))); near.append(int(m.group(5))); Cmax.append(int(m.group(6)))
|
|
continue
|
|
m = posRe.match(line)
|
|
if m:
|
|
pos.append((int(m.group(1)), int(m.group(2))))
|
|
continue
|
|
m = evRe.match(line)
|
|
if m:
|
|
ts = parse_ts(m.group(1))
|
|
if ts is None:
|
|
continue
|
|
events.append((ts, m.group(2)))
|
|
return dict(t=np.array(t, object), X=np.array(X), Y=np.array(Y), C=np.array(C),
|
|
V=np.array(V), P=np.array(P), S=np.array(S), er=np.array(er),
|
|
ft_t=np.array(ft_t, object), dC=np.array(dC), dynE=np.array(dynE),
|
|
e=np.array(e), near=np.array(near), Cmax=np.array(Cmax), pos=pos, events=events)
|
|
|
|
def curate(events):
|
|
out = []
|
|
for (ts, text) in events:
|
|
u = text.upper()
|
|
if 'APLICADOS' in u or text.startswith('AT:'):
|
|
kind = 'auto-thresh'
|
|
elif 'ERRMAX' in u or 'MINERRMAX' in u:
|
|
kind = 'er-set'
|
|
elif 'DELAY' in u:
|
|
kind = 'delay'
|
|
elif text.startswith('SunPos'):
|
|
kind = 'sun-pos'
|
|
elif 'COM READY' in u:
|
|
kind = 'boot'
|
|
elif 'WS CERRADO' in u or 'DESCONECTADO' in u:
|
|
kind = 'ws-close'
|
|
else:
|
|
continue
|
|
out.append((ts, kind))
|
|
seen = defaultdict(int); res = []
|
|
for (ts, kind) in out:
|
|
if seen[kind] >= 5:
|
|
continue
|
|
seen[kind] += 1
|
|
res.append((ts, kind))
|
|
return res
|
|
|
|
def derive(L):
|
|
L['CMa'] = L['C'] / 6144.0
|
|
L['PW'] = L['P'] / 8192.0
|
|
L['dCMa'] = L['dC'] / 6144.0
|
|
L['events_c'] = curate(L['events'])
|
|
n = len(L['t']); L['n'] = n
|
|
if n:
|
|
L['t0'] = L['t'][0]; L['tend'] = L['t'][-1]
|
|
span = (L['tend'] - L['t0']).total_seconds() or 1
|
|
L['minutes'] = span / 60.0
|
|
L['hour'] = np.array([(x - L['t0']).total_seconds() / 3600.0 for x in L['t']])
|
|
sec = [(x - L['t0']).total_seconds() for x in L['t']]
|
|
L['gaps'] = [int(g) for g in np.diff(sec) if g > 15]
|
|
else:
|
|
L['t0']=None; L['tend']=None; L['minutes']=0.0; L['hour']=np.array([]); L['gaps']=[]
|
|
tr = []
|
|
for i in range(1, len(L['er'])):
|
|
if L['er'][i] != L['er'][i-1]:
|
|
tr.append((L['t'][i], int(L['er'][i])))
|
|
L['er_trans'] = tr
|
|
L['er_seq'] = []
|
|
prev = None
|
|
for i in range(len(L['er'])):
|
|
if L['er'][i] != prev:
|
|
L['er_seq'].append(int(L['er'][i])); prev = L['er'][i]
|
|
drops = []
|
|
if n > 1:
|
|
d = np.diff(L['C'].astype(float))
|
|
L['n_drops'] = int((d < -100000).sum())
|
|
for j in np.argsort(d):
|
|
if d[j] < -100000:
|
|
drops.append((L['t'][j+1], int(-d[j])))
|
|
if len(drops) >= 6:
|
|
break
|
|
else:
|
|
L['n_drops'] = 0
|
|
L['big_drops'] = drops
|
|
L['near0_frac'] = float((L['near'] == 0).mean()) if len(L['near']) else 0.0
|
|
L['n_pos'] = len(L['pos'])
|
|
L['n_distinct'] = len(set(zip(L['X'].tolist(), L['Y'].tolist()))) if n else 0
|
|
L['Cmin'] = float(L['CMa'].min()) if n else 0
|
|
L['Cmaxv'] = float(L['CMa'].max()) if n else 0
|
|
L['Pmax'] = float(L['PW'].max()) if n else 0
|
|
L['home_idx'] = np.where(L['X'] == 0)[0]
|
|
L['n_home'] = int(len(L['home_idx']))
|
|
L['valid'] = (L['X'] != 0)
|
|
return L
|
|
|
|
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)
|
|
|
|
def event_marks(ax, L, kinds, yref):
|
|
for (ts, kind) in L['events_c']:
|
|
if kinds is not None and kind not in kinds:
|
|
continue
|
|
ax.axvline(mdates.date2num(ts), color='gray', ls=':', lw=0.6, alpha=0.5, zorder=0)
|
|
ax.text(mdates.date2num(ts), yref, kind, rotation=90, fontsize=6, ha='center', va='top', color='dimgray')
|
|
|
|
def savefig(fig, base):
|
|
# Render to an in-memory buffer, then delete+create the destination.
|
|
# (Never overwrite an existing OneDrive-synced file in place: it raises
|
|
# OSError EINVAL due to stale placeholder state. Create-after-delete works.)
|
|
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 p_trajectory(ax, L, equal=True, colorbar=False):
|
|
X = L['X'][L['valid']]; Y = L['Y'][L['valid']]; hour = L['hour'][L['valid']]
|
|
if not len(X):
|
|
ax.text(0.5, 0.5, 'no data', ha='center'); return
|
|
ax.plot(X, Y, color='0.82', lw=0.5, zorder=1)
|
|
sc = ax.scatter(X, Y, c=hour, cmap='coolwarm', s=3, alpha=0.8, zorder=2, edgecolors='none')
|
|
ax.plot([X[0]], [Y[0]], 'o', ms=10, mfc='green', mec='k', zorder=4)
|
|
ax.plot([X[-1]], [Y[-1]], '*', ms=18, mfc='red', mec='k', zorder=4)
|
|
ax.annotate('start', (X[0], Y[0]), textcoords='offset points', xytext=(6, 4), fontsize=8, color='green')
|
|
ax.annotate('end', (X[-1], Y[-1]), textcoords='offset points', xytext=(6, 4), fontsize=8, color='red')
|
|
ax.set_xlabel('Position X'); ax.set_ylabel('Position Y')
|
|
if equal:
|
|
ax.set_aspect('equal', adjustable='box')
|
|
if colorbar:
|
|
ax.figure.colorbar(sc, ax=ax, label='hours since start [h]')
|
|
|
|
def p_ts(ax, L, field, color, ylabel, ev_kinds=None, show_drops=False, fill=False):
|
|
x = mdates.date2num(L['t'].tolist()); y = L[field]
|
|
if fill:
|
|
ax.fill_between(x, 0, y, color=color, alpha=0.12)
|
|
ax.plot(x, y, color=color, lw=0.7)
|
|
ax.set_xlabel('Time'); ax.set_ylabel(ylabel)
|
|
fmt_ax(ax)
|
|
top = ax.get_ylim()[1]
|
|
if ev_kinds is not None:
|
|
event_marks(ax, L, ev_kinds, top)
|
|
if show_drops:
|
|
for (ts, drop) in L['big_drops']:
|
|
ax.axvline(mdates.date2num(ts), color='red', ls='--', lw=0.9, alpha=0.7)
|
|
ax.annotate('-%.0f mA' % (drop/6144.0), (mdates.date2num(ts), top), textcoords='offset points', xytext=(2, -2), fontsize=6, color='red')
|
|
|
|
def p_er(ax, L):
|
|
x = mdates.date2num(L['t'].tolist()); er = L['er']
|
|
ax.step(x, er, where='post', color='tab:purple', lw=1.3)
|
|
ax.set_xlabel('Time'); ax.set_ylabel('er (threshold)')
|
|
fmt_ax(ax)
|
|
for (ts, val) in L['er_trans']:
|
|
ax.plot([mdates.date2num(ts)], [val], 'o', color='red', ms=5, zorder=5)
|
|
ax.annotate('er=%d' % val, (mdates.date2num(ts), val), textcoords='offset points', xytext=(5, 5), fontsize=7, color='red')
|
|
|
|
def p_ft(ax, L):
|
|
x = mdates.date2num(L['ft_t'].tolist())
|
|
ax.plot(x, L['dCMa'], color='tab:blue', lw=0.6, label='dC [mA]')
|
|
ax.set_xlabel('Time'); ax.set_ylabel('dC [mA]', color='tab:blue')
|
|
ax.tick_params(axis='y', labelcolor='tab:blue')
|
|
fmt_ax(ax)
|
|
ax2 = ax.twinx()
|
|
ax2.plot(x, L['near'], color='0.55', lw=0.5, ls=':', alpha=0.7, label='near')
|
|
ax2.set_ylim(-0.3, 1.3); ax2.set_ylabel('near', color='0.55'); ax2.tick_params(axis='y', labelcolor='0.55')
|
|
h1, l1 = ax.get_legend_handles_labels(); h2, l2 = ax2.get_legend_handles_labels()
|
|
ax.legend(h1+h2, l1+l2, fontsize=7, loc='upper right')
|
|
|
|
def stats_text(L, label):
|
|
erseq = ' -> '.join(str(v) for v in L['er_seq'])
|
|
maxdrop = max((d[1] for d in L['big_drops']), default=0)
|
|
lines = [
|
|
label, '',
|
|
'Window: %s -> %s (%.0f min)' % (L['t0'].strftime('%H:%M:%S'), L['tend'].strftime('%H:%M:%S'), L['minutes']),
|
|
'Samples: X/Y=%d FineTune=%d POS=%d' % (L['n'], len(L['ft_t']), L['n_pos']),
|
|
'Distinct pos.: %d' % L['n_distinct'],
|
|
'Gaps > 15 s: %d (max %d s)' % (len(L['gaps']), (max(L['gaps']) if L['gaps'] else 0)),
|
|
'er (sequence): %s' % erseq,
|
|
'C: %.1f - %.1f mA | P max %.0f W/m2' % (L['Cmin'], L['Cmaxv'], L['Pmax']),
|
|
'Large C drops (>16 mA): %d (max %.0f mA)' % (L['n_drops'], maxdrop/6144.0),
|
|
'Home (0,0) samples excluded from trajectory: %d' % L['n_home'],
|
|
'near=0: %.1f%%' % (100*L['near0_frac']),
|
|
]
|
|
return '\n'.join(lines)
|
|
|
|
def build_overview(L, tag, label):
|
|
fig, axes = plt.subplots(2, 3, figsize=(19, 9), dpi=200)
|
|
fig.suptitle('Microtracking - %s' % label, fontsize=15, fontweight='bold')
|
|
p_trajectory(axes[0,0], L, equal=False)
|
|
axes[0,0].set_title('Trajectory (color = time of day)')
|
|
p_ts(axes[0,1], L, 'PW', '#ff7f0e', 'Irradiance P [W/m2]', ev_kinds={'auto-thresh','delay','sun-pos','ws-close','boot'}, fill=True)
|
|
axes[0,1].set_title('Irradiance P [W/m2]')
|
|
p_ts(axes[0,2], L, 'CMa', '#d62728', 'Photocurrent C [mA]', ev_kinds={'auto-thresh','delay','sun-pos','ws-close','boot'}, show_drops=True)
|
|
axes[0,2].set_title('Photocurrent C [mA] (red lines = drops)')
|
|
p_er(axes[1,0], L)
|
|
axes[1,0].set_title('Adaptive threshold er [u]')
|
|
p_ft(axes[1,1], L)
|
|
axes[1,1].set_title('FineTune: dC [mA] and near')
|
|
axes[1,2].axis('off')
|
|
axes[1,2].text(0.0, 1.0, stats_text(L, label), family='monospace', fontsize=9, va='top',
|
|
bbox=dict(boxstyle='round,pad=0.6', fc='#f7f7f7', ec='gray'))
|
|
fig.tight_layout(rect=[0, 0, 1, 0.95])
|
|
savefig(fig, tag + '_overview')
|
|
plt.close(fig)
|
|
|
|
def build_trajectory(L, tag, label):
|
|
fig, ax = plt.subplots(1, 1, figsize=(9, 9), dpi=200)
|
|
fig.suptitle('Microtracking - %s' % label, fontsize=14, fontweight='bold')
|
|
p_trajectory(ax, L, equal=True, colorbar=True)
|
|
ax.set_title('Tracker position trajectory (color = time of day)', fontsize=11)
|
|
if L['n_home'] > 0:
|
|
h0 = int(L['home_idx'][0]); a = max(0, h0 - 12); b = min(len(L['X']), h0 + 12)
|
|
rx = L['X'][a:b]; ry = L['Y'][a:b]; rh = L['hour'][a:b]
|
|
axins = ax.inset_axes([0.02, 0.02, 0.40, 0.40])
|
|
axins.plot(rx, ry, color='0.6', lw=0.6, zorder=1)
|
|
axins.scatter(rx, ry, c=rh, cmap='coolwarm', s=6, edgecolors='none', zorder=2)
|
|
hm = (rx == 0)
|
|
axins.scatter(rx[hm], ry[hm], s=18, facecolors='none', edgecolors='red', linewidths=1.2, zorder=5)
|
|
axins.set_title('startup: lost-lock to home (0,0), re-locked in ~10 s', fontsize=7)
|
|
axins.set_xticklabels([]); axins.set_yticklabels([]); axins.tick_params(labelsize=0)
|
|
for s in ('top', 'right'):
|
|
axins.spines[s].set_visible(False)
|
|
fig.tight_layout(rect=[0, 0, 1, 0.94])
|
|
savefig(fig, tag + '_trajectory')
|
|
plt.close(fig)
|
|
|
|
def build_all_traj(parsed):
|
|
fig, axes = plt.subplots(1, len(parsed), figsize=(6.2*len(parsed), 6.2), dpi=200)
|
|
fig.suptitle('Microtracking - position trajectory per session (independent scale per panel)', fontsize=15, fontweight='bold')
|
|
for ax, (tag, label, L) in zip(axes, parsed):
|
|
p_trajectory(ax, L, equal=True, colorbar=True)
|
|
ax.set_title('%s\n%d distinct positions' % (label, L['n_distinct']), fontsize=11)
|
|
fig.tight_layout(rect=[0, 0, 1, 0.93])
|
|
savefig(fig, 'trajectories_all')
|
|
plt.close(fig)
|
|
|
|
def build_crosscheck(parsed):
|
|
d = {tag: L for (tag, label, L) in parsed}
|
|
A = d['14sep_A']; B = d['14sep_B']
|
|
lo = max(A['t0'], B['t0']); hi = min(A['tend'], B['tend'])
|
|
def sel(L):
|
|
idx = np.array([i for i, x in enumerate(L['t']) if lo <= x <= hi])
|
|
return L['X'][idx], L['Y'][idx], L['PW'][idx], mdates.date2num([L['t'][i] for i in idx.tolist()])
|
|
xA, yA, PA, txA = sel(A)
|
|
xB, yB, PB, txB = sel(B)
|
|
fig, axes = plt.subplots(1, 2, figsize=(16, 7), dpi=200)
|
|
fig.suptitle('Cross-check 14/09: log(3) vs gyuggyhoy - same session (common window %s to %s)' % (lo.strftime('%H:%M:%S'), hi.strftime('%H:%M:%S')), fontsize=14, fontweight='bold')
|
|
ax = axes[0]
|
|
ax.plot(xA, yA, color='0.8', lw=0.4, zorder=1); ax.plot(xB, yB, color='0.8', lw=0.4, zorder=1)
|
|
ax.scatter(xA, yA, s=6, color=LOG_COLORS['14sep_A'], alpha=0.85, zorder=2, label='log(3)')
|
|
ax.scatter(xB, yB, s=6, color=LOG_COLORS['14sep_B'], alpha=0.85, zorder=2, label='gyuggyhoy')
|
|
ax.set_xlabel('X'); ax.set_ylabel('Y'); ax.set_title('Trajectory (overlay: 0 X/Y mismatches)'); ax.legend()
|
|
ax.set_aspect('equal', adjustable='box')
|
|
ax = axes[1]
|
|
ax.plot(txA, PA, color=LOG_COLORS['14sep_A'], lw=0.7, label='log(3)')
|
|
ax.plot(txB, PB, color=LOG_COLORS['14sep_B'], lw=0.7, label='gyuggyhoy')
|
|
ax.set_xlabel('Time'); ax.set_ylabel('P [W/m2]'); ax.set_title('Irradiance P (exact overlay)'); fmt_ax(ax); ax.legend()
|
|
fig.tight_layout(rect=[0, 0, 1, 0.93])
|
|
savefig(fig, 'crosscheck_14sep')
|
|
plt.close(fig)
|
|
|
|
def main():
|
|
parsed = []
|
|
for tag, path, label in LOGS:
|
|
print('Parsing', tag, '...')
|
|
L = derive(parse_log(path))
|
|
parsed.append((tag, label, L))
|
|
t0s = L['t0'].strftime('%H:%M:%S') if L['t0'] else 'n/a'
|
|
tens = L['tend'].strftime('%H:%M:%S') if L['tend'] else 'n/a'
|
|
print(' %s: n=%d %s -> %s (%.0f min) gaps>15s=%d drops=%d er=%s' % (
|
|
tag, L['n'], t0s, tens, L['minutes'], len(L['gaps']), L['n_drops'], L['er_seq']))
|
|
for tag, label, L in parsed:
|
|
build_overview(L, tag, label)
|
|
build_trajectory(L, tag, label)
|
|
print(' saved %s_overview.{pdf,png} , %s_trajectory.{pdf,png}' % (tag, tag))
|
|
build_all_traj(parsed)
|
|
print(' saved trajectories_all.{pdf,png}')
|
|
try:
|
|
build_crosscheck(parsed)
|
|
print(' saved crosscheck_14sep.{pdf,png}')
|
|
except Exception as ex:
|
|
print(' crosscheck skipped:', repr(ex))
|
|
print('DONE ->', OUT)
|
|
|
|
if __name__ == '__main__':
|
|
main() |