# -*- coding: utf-8 -*- """PlanetiQ Level-1A SpWx TAR, Level-1B TEC and Level-2 EDP monitoring. Input logs: UTC, ET, GN04, GN05, Total (30-minute counts). Usage: python Plot_count_planetiq_TEC_EDP_updated.py planetiq_nccf current python Plot_count_planetiq_TEC_EDP_updated.py planetiq_nccf 2026-10-13_16:10:00 """ import argparse import os from dataclasses import dataclass from datetime import datetime, timezone from zoneinfo import ZoneInfo import matplotlib as mpl mpl.use('Agg') import matplotlib.dates as mdates import matplotlib.pyplot as plt import numpy as np import pandas as pd ET = ZoneInfo('America/New_York') COLORS = {'GN04': 'red', 'GN05': 'green', 'Total': 'black'} def load_count_log(filename): """Read 30-minute count CSV; tolerate header spaces and older 'total' casing.""" df = pd.read_csv(filename, skipinitialspace=True) df.columns = df.columns.str.strip() df = df.rename(columns={c: 'Total' for c in df.columns if c.lower() == 'total'}) required = {'UTC', 'GN04', 'GN05', 'Total'} missing = required.difference(df.columns) if missing: raise ValueError(f'{filename}: missing columns {sorted(missing)}') # UTC is authoritative. Derive Eastern Time with DST rules rather than trusting # a local-time string lacking a timezone offset. df['UTC'] = pd.to_datetime(df['UTC'], utc=True, errors='coerce') for col in ('GN04', 'GN05', 'Total'): df[col] = pd.to_numeric(df[col], errors='coerce') df = df.dropna(subset=['UTC']).set_index('UTC') df = df[~df.index.duplicated(keep='first')].sort_index() df['ET'] = df.index.tz_convert(ET) return df def subset(df, start, end): return df.loc[(df.index >= start) & (df.index <= end)] def aggregate(df, start, end, rule): # Only aggregate count fields; min_count=1 avoids inventing zero counts for # empty bins in an incomplete input log. data = subset(df, start, end)[['GN04', 'GN05', 'Total']] return data.resample(rule).sum(min_count=1) def plot_lines(ax, df, prefix, marker=False): # Draw Total first and underneath the individual spacecraft curves. for name, z, width in [('Total', 2, 1.5), ('GN04', 4, 2.0), ('GN05', 5, 2.0)]: ax.plot(df.index, df[name], color=COLORS[name], linewidth=width, marker='o' if marker else None, markersize=2.5, zorder=z, label=f'{prefix} {name}' if prefix else name) def padded_ylim(df, name='Total'): value = df[name].max() if not df.empty else np.nan return 0, max(1, float(value) * 1.12 + 0.5) if pd.notna(value) else 1 @dataclass class plot_monitor: task: str now_utc: datetime fname_L1B: str path_L1B: str fname_L1A: str fname_L2_EDP: str path_L1A: str def ingest(self): self.df_L2_EDP = load_count_log(os.path.join(self.path_L1B, self.fname_L2_EDP)) self.df_L1B = load_count_log(os.path.join(self.path_L1B, self.fname_L1B)) self.df_L1A = load_count_log(os.path.join(self.path_L1A, self.fname_L1A)) def plot_monitor(self): self.plot_L2_EDP() self.plot_L1B_TEC() self.plot_L1A_TAR() def plot_three_panels(self, data, product, title, filename, path, show_tar=False): mpl.rcParams.update({'font.size': 11}) fig, axs = plt.subplots(3, 1, figsize=(13, 10.5)) fig.subplots_adjust(left=0.16, right=0.86 if show_tar else 0.96, bottom=0.075, top=0.86, hspace=0.51) now = pd.Timestamp(self.now_utc) starts = [now - pd.Timedelta(hours=30), now - pd.Timedelta(days=7), now - pd.Timedelta(days=30)] labels = ['Last 30 hours\n(30-min)', 'Last 7 days\n(6-Hourly)', 'Last 30 days\n(Daily)'] rules = [None, '6h', '1D'] for i, (ax, start, label, rule) in enumerate(zip(axs, starts, labels, rules)): # Resampling the exact requested window includes a partial first # interval. Fix the visible limits at exactly 30h/7d/30d. df = subset(data, start, now) if rule is None else aggregate(data, start, now, rule) plot_lines(ax, df, product if show_tar else '', marker=(i == 0)) ax.set_xlim(start, now) ax.set_ylim(*padded_ylim(df)) ax.set_ylabel('Number of ' + ('L1A' if product == 'L1A' else product) + ' Files') ax.grid(axis='x', which='major', linestyle='-', color='0.8', linewidth=0.75, zorder=0) ax.text(-0.15, 0.5, label, transform=ax.transAxes, ha='right', va='center', fontsize=11) ax.tick_params(axis='x', direction='out', length=4, pad=5) if i == 0: ax.xaxis.set_major_locator(mdates.HourLocator(interval=3, tz=timezone.utc)) ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M', tz=timezone.utc)) ax.xaxis.set_minor_locator(mdates.HourLocator(interval=1, tz=timezone.utc)) ax.grid(axis='x', which='minor', linestyle=':', color='0.85', zorder=0) ax.set_xlabel('UTC') # Use a single secondary time axis. Both x axes refer to the # same instants: mdates handles the timezone conversion. top = ax.twiny() top.set_xlim(ax.get_xlim()) top.xaxis.set_major_locator(mdates.HourLocator(interval=3, tz=timezone.utc)) top.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M', tz=ET)) top.set_xlabel('Eastern Time', labelpad=11) top.tick_params(axis='x', pad=4) elif i == 1: # One tick per UTC day, only within the 7-day window. ax.xaxis.set_major_locator(mdates.DayLocator(interval=1, tz=timezone.utc)) ax.xaxis.set_major_formatter(mdates.DateFormatter('%m-%d', tz=timezone.utc)) ax.xaxis.set_minor_locator(mdates.HourLocator(interval=6, tz=timezone.utc)) ax.grid(axis='x', which='minor', linestyle=':', color='0.88', zorder=0) ax.set_xlabel('Date (UTC)') # Avoid the visually clipped/overlapping left-edge date label. for tick in ax.get_xticklabels(): tick.set_rotation(0) tick.set_fontsize(10) else: ax.xaxis.set_major_locator(mdates.DayLocator(interval=3, tz=timezone.utc)) ax.xaxis.set_major_formatter(mdates.DateFormatter('%m-%d', tz=timezone.utc)) ax.set_xlabel('Date (UTC)') if show_tar: tar_df = subset(self.df_L1A, start, now) if rule is None else aggregate(self.df_L1A, start, now, rule) right = ax.twinx() right.plot(tar_df.index, tar_df['Total'], color='tab:pink', linewidth=1.6, alpha=0.85, zorder=3, label='L1A TAR') right.set_ylim(*padded_ylim(tar_df)) right.set_ylabel('Number of L1A Files', color='tab:pink') right.tick_params(axis='y', colors='tab:pink') right.grid(False) # Legend uses actual line handles from first panel, with Total last. handles, legends = axs[0].get_legend_handles_labels() if show_tar: # The secondary axis for first panel is followed by its right axis. for axis in fig.axes: if axis is not axs[0] and axis.get_ylabel() == 'Number of L1A Files' and axis.get_shared_x_axes().joined(axis, axs[0]): h, l = axis.get_legend_handles_labels() handles += h legends += l break fig.legend(handles, legends, loc='upper center', bbox_to_anchor=(0.55, 0.935), ncol=4, frameon=False) update_et = now.tz_convert(ET).strftime('%Y-%m-%d %H:%M') fig.suptitle(f'{title} (Last update: {update_et} ET)', fontsize=15, y=0.98) output = os.path.join(path, filename) fig.savefig(output, dpi=180, bbox_inches='tight') plt.close(fig) print(output) def plot_L2_EDP(self): self.plot_three_panels(self.df_L2_EDP, 'EDP', 'PlanetiQ Level-2 EDP File Monitoring', f'Monitor_L2_EDP_30min_{self.task}_nccf.png', self.path_L1B, show_tar=True) def plot_L1B_TEC(self): self.plot_three_panels(self.df_L1B, 'TEC', 'PlanetiQ Level-1B TEC File Monitoring', f'Monitor_L1B_TEC_30min_{self.task}_nccf.png', self.path_L1B, show_tar=True) def plot_L1A_TAR(self): self.plot_three_panels(self.df_L1A, 'L1A', 'PlanetiQ Level-1A SpWx TAR File Monitoring', f'Monitor_L1A_SpWx_30min_{self.task}_nccf.png', self.path_L1A) def main(): parser = argparse.ArgumentParser() parser.add_argument('task', help='task name, e.g. planetiq_nccf') parser.add_argument('now_utc_str', help='current or YYYY-MM-DD_HH:MM:SS (UTC)') args = parser.parse_args() if args.now_utc_str == 'current': now_utc = datetime.now(timezone.utc) else: now_utc = datetime.strptime(args.now_utc_str, '%Y-%m-%d_%H:%M:%S').replace(tzinfo=timezone.utc) mission, env = args.task.split('_', 1) if env != 'nccf': parser.error('only *_nccf tasks are supported') base = os.path.join(os.environ['HOME'], 'gps_pod/L1A_L1B_ExcessPhase_v1.3_20251216_addconfig2/count_statistic') obj = plot_monitor(args.task, now_utc, f'count_L1B_TEC_30min_{mission}_{env}.log', base, f'count_L1A_TAR_30min_{mission}_SpWx_{env}.log', f'count_L2_EDP_30min_{mission}_{env}.log', base) obj.ingest() obj.plot_monitor() if __name__ == '__main__': main()