#!/usr/bin/env python3
import cgi
import cgitb
import html as _html
import json
import os
import sqlite3
import statistics
import sys
from collections import defaultdict
from datetime import datetime, timedelta
sys.path.insert(0, '/opt/ngon/apps')
sys.path.insert(0, '/var/www/html/ngon')
from managers.auth_manager import AuthManager, generate_login_page_html
from links import generate_dropdown_html, generate_dropdown_css, generate_dropdown_js
# Shared OOS metric definitions -- kept in lockstep with the gen_service query
# builder (api/gen_service_api.py). Both count OOS time the exact same way.
from generators.gen_oos_metrics import compute_oos_metrics
# Swap tracking is a SEPARATE population from OOS -- a swap-tagged gen is usually
# still on a pod making power (it reads 'active' in the daily snapshot), so it is
# invisible to every OOS metric above. Sourced from the gen_notes event log
# rather than the daily snapshot; see the module docstring for what that costs.
from generators.gen_swap_metrics import (
daily_open_counts,
load_current_swap_tags,
load_swap_episodes,
summarize_swaps,
)
cgitb.enable()
_form = cgi.FieldStorage()
_auth = AuthManager('gen_trends')
_auth_required, _should_exit, _headers = _auth.require_auth(_form)
if _should_exit:
print("Content-Type: application/json")
if _headers:
print(_headers)
print("")
if _auth_required:
print('{"success": false, "error": "Authentication required"}')
else:
print('{"success": true}')
sys.exit(0)
if _auth_required:
print("Content-Type: text/html")
print("")
print(generate_login_page_html("Generator Trends"))
sys.exit(0)
_user_access = AuthManager.get_user_access()
"""
Generator Trends Dashboard
Historical analysis of generator status and out-of-service tracking across all
sites. Reads daily gen→site/pod assignments from /opt/ngon/data/gen_status.db
(populated daily by apps/reporting/extract_generator_mapping.py). Earlier
version read from gen_status_saves/*.txt — that path is dead; the legacy text
files were migrated into this SQLite DB.
"""
GEN_STATUS_DB = "/opt/ngon/data/gen_status.db"
def load_all_historical_data():
"""
Pull every daily snapshot from gen_status.db and reshape into the per-day
dict structure the downstream timeline/metrics code expects:
[
{
'file_date': datetime,
'date_str': 'YYYY-MM-DD',
'pod_assignments': {pod_name: [gen_ids]}, # status='active'
'oos_generators': {gen_id: reason_text}, # status='out_of_service'
'all_generators': sorted([all gen_ids seen this day]),
},
...
]
Spare gens (status='spare') are folded into all_generators but appear in
neither pod_assignments nor oos_generators — matches the original "exists
but unassigned" path so they render as 'unknown' in the timeline.
"""
try:
if not os.path.exists(GEN_STATUS_DB):
return [], f"Generator status DB not found at {GEN_STATUS_DB}"
conn = sqlite3.connect(GEN_STATUS_DB)
try:
rows = conn.execute(
"SELECT date, gen_id, pod, status, notes "
"FROM generator_daily_status "
"ORDER BY date"
).fetchall()
finally:
conn.close()
if not rows:
return [], "Generator status DB is empty"
# Group rows by date
per_day = {}
for date_str, gen_id, pod, status, notes in rows:
day = per_day.setdefault(date_str, {
'pod_assignments': defaultdict(list),
'oos_generators': {},
'all_generators': set(),
})
day['all_generators'].add(gen_id)
if status == 'active' and pod:
day['pod_assignments'][pod].append(gen_id)
elif status == 'out_of_service':
day['oos_generators'][gen_id] = (notes or '').strip() or "No reason specified"
# 'spare' (or anything else): present in all_generators only
result = []
for date_str in sorted(per_day.keys()):
try:
file_date = datetime.strptime(date_str, '%Y-%m-%d')
except ValueError:
continue
d = per_day[date_str]
result.append({
'file_date': file_date,
'date_str': date_str,
'pod_assignments': dict(d['pod_assignments']),
'oos_generators': d['oos_generators'],
'all_generators': sorted(d['all_generators']),
})
return result, None
except Exception as e:
return [], f"Error loading historical data: {e}"
def filter_historical_data(all_data, days_back):
"""Filter the pre-loaded historical data by date range."""
if days_back == 'all':
return all_data
cutoff_date = datetime.now() - timedelta(days=days_back)
return [data for data in all_data if data['file_date'] >= cutoff_date]
def track_generator_status(historical_data):
"""Track generator status changes over time."""
if not historical_data:
return {}, {}
# Get all generators across all time periods
all_generators = set()
for data in historical_data:
all_generators.update(data['all_generators'])
# Track each generator's status over time
generator_timeline = defaultdict(list)
# Recorded-rows-only sequences (date_str, status, note) per gen -- only days
# the gen actually appears in the snapshot. Feeds the shared OOS helper so
# this page and the gen_service query builder count OOS identically.
generator_records = defaultdict(list)
generator_stats = defaultdict(lambda: {
'total_days': 0,
'in_service_days': 0,
'out_of_service_periods': [],
'current_status': 'unknown',
'current_pod': None,
'current_reason': None,
'last_seen_date': None,
'first_seen_date': None,
'historical_reasons': [] # Track all reasons over time
})
for data in historical_data:
date = data['file_date']
date_str = data['date_str']
assignments = data['pod_assignments']
oos_generators = data.get('oos_generators', {})
# Create reverse mapping: generator -> pod
gen_to_pod = {}
for pod, generators in assignments.items():
for gen in generators:
gen_to_pod[gen] = pod
# Recorded-rows-only sequence per gen (only gens present in THIS day's
# snapshot), classified the same way the helper expects.
for gen in data['all_generators']:
if gen in gen_to_pod:
rstatus = 'active'
elif gen in oos_generators:
rstatus = 'out_of_service'
else:
rstatus = 'spare'
generator_records[gen].append((date_str, rstatus, oos_generators.get(gen)))
# Track each generator
for gen in all_generators:
is_in_service = gen in gen_to_pod and gen not in oos_generators
is_oos = gen in oos_generators
pod = gen_to_pod.get(gen, None)
reason = oos_generators.get(gen, None)
generator_timeline[gen].append({
'date': date,
'date_str': date_str,
'in_service': is_in_service,
'out_of_service': is_oos,
'pod': pod,
'reason': reason
})
# Update stats
stats = generator_stats[gen]
stats['total_days'] += 1
if is_in_service:
stats['in_service_days'] += 1
stats['current_status'] = 'in_service'
stats['current_pod'] = pod
stats['current_reason'] = None
stats['last_seen_date'] = date
if not stats['first_seen_date']:
stats['first_seen_date'] = date
elif is_oos:
stats['current_status'] = 'out_of_service'
stats['current_pod'] = None
stats['current_reason'] = reason
# Track historical reasons
if reason and reason not in [r['reason'] for r in stats['historical_reasons']]:
stats['historical_reasons'].append({
'reason': reason,
'first_seen': date,
'last_seen': date
})
elif reason:
# Update last seen date for this reason
for r in stats['historical_reasons']:
if r['reason'] == reason:
r['last_seen'] = date
break
else:
# Generator exists but not assigned anywhere (unknown status)
stats['current_status'] = 'unknown'
stats['current_pod'] = None
stats['current_reason'] = None
# Calculate out of service periods and count non-continuous appearances
for gen, timeline in generator_timeline.items():
stats = generator_stats[gen]
periods = []
current_oos_start = None
current_oos_reason = None
oos_incidents = 0 # Count of separate OOS incidents
for entry in timeline:
if entry['out_of_service']:
if current_oos_start is None:
# Start of new OOS period
current_oos_start = entry['date']
current_oos_reason = entry['reason']
oos_incidents += 1
# Continue existing OOS period (reason might change)
else:
if current_oos_start is not None:
# End of OOS period
periods.append({
'start': current_oos_start,
'end': entry['date'],
'duration': (entry['date'] - current_oos_start).days + 1, # Include start day
'reason': current_oos_reason
})
current_oos_start = None
current_oos_reason = None
# If currently OOS, add ongoing period
if current_oos_start is not None:
periods.append({
'start': current_oos_start,
'end': None, # Still ongoing
'duration': (datetime.now() - current_oos_start).days + 1,
'reason': current_oos_reason
})
stats['out_of_service_periods'] = periods
stats['oos_incidents'] = oos_incidents
return dict(generator_timeline), dict(generator_stats), dict(generator_records)
def calculate_generator_metrics(generator_stats, generator_timeline, generator_records):
"""Calculate metrics for generator reliability.
The OOS numbers (total_oos_days, incidents, current_oos_days, avg/max
duration, uptime %) come from the shared apps/generators/gen_oos_metrics
helper -- the SAME code the gen_service query builder uses -- so the two
pages can't disagree. Display-only fields (current_pod, historical_reasons,
first/last seen) still come from the per-day timeline scan above.
"""
metrics = {}
for gen, stats in generator_stats.items():
if stats['total_days'] == 0:
continue
# Canonical OOS metrics, recorded-snapshot-day basis (see helper module).
m = compute_oos_metrics(generator_records.get(gen, []))
# Current status label derived from the gen's last RECORDED status so it
# always agrees with the numbers (not the timeline's unknown-injected tail).
if m['currently_oos']:
current_status = 'out_of_service'
elif m['last_status'] == 'active':
current_status = 'in_service'
else:
current_status = 'unknown'
current_reason = m['current_reason'] if m['currently_oos'] else None
metrics[gen] = {
'uptime_percentage': m['in_service_percent'],
'total_days': m['total_days'],
'in_service_days': m['active_days'],
'out_of_service_days': m['total_oos_days'],
'total_oos_days': m['total_oos_days'],
'avg_oos_duration': m['avg_oos_duration'],
'max_oos_duration': m['max_oos_duration'],
'total_oos_incidents': m['oos_incidents'],
'current_status': current_status,
'current_pod': stats['current_pod'],
'current_reason': current_reason,
'current_oos_days': m['current_oos_days'],
'first_seen': stats['first_seen_date'],
'last_seen': stats['last_seen_date'],
'oos_periods': m['oos_periods'],
'historical_reasons': stats.get('historical_reasons', [])
}
return metrics
def render_fleet_overview(generator_metrics):
"""Render the generator fleet overview table."""
html = """
Generator Fleet Overview
Generator
Current Status
Current Pod / Reason
Uptime %
Total OOS Days
OOS Incidents
Avg OOS Duration
Max OOS Duration
"""
# Sort generators by current status (OOS first, then by uptime)
def sort_key(item):
gen, metrics = item
status_priority = 0 if metrics['current_status'] == 'out_of_service' else 1
return (status_priority, -metrics['uptime_percentage'], gen)
sorted_generators = sorted(generator_metrics.items(), key=sort_key)
for gen, metrics in sorted_generators:
# Status classification
if metrics['current_status'] == 'out_of_service':
if metrics['current_oos_days'] >= 7:
status_class = "critical"
elif metrics['current_oos_days'] >= 3:
status_class = "warning"
else:
status_class = "recent-oos"
elif metrics['uptime_percentage'] >= 95:
status_class = "excellent"
elif metrics['uptime_percentage'] >= 90:
status_class = "good"
elif metrics['uptime_percentage'] >= 80:
status_class = "warning"
else:
status_class = "critical"
# Format current status
if metrics['current_status'] == 'out_of_service':
if metrics['current_oos_days'] > 0:
status_text = f"OOS ({metrics['current_oos_days']}d)"
else:
status_text = "OOS"
else:
status_text = "In Service"
# Format pod/reason column
if metrics['current_status'] == 'out_of_service':
pod_reason = metrics.get('current_reason', 'Unknown reason') or 'Unknown reason'
else:
pod_reason = metrics['current_pod'] or "—"
avg_duration = f"{metrics['avg_oos_duration']:.1f}" if metrics['avg_oos_duration'] > 0 else "—"
max_duration = f"{metrics['max_oos_duration']}" if metrics['max_oos_duration'] > 0 else "—"
html += f"""
{gen}
{status_text}
{pod_reason[:50] + '...' if len(pod_reason) > 50 else pod_reason}
{metrics['uptime_percentage']:.1f}%
{metrics['total_oos_days']}
{metrics['total_oos_incidents']}
{avg_duration}
{max_duration}
"""
html += """
"""
return html
def render_timeline_chart(generator_timeline, days_back=30):
"""Render the interactive generator timeline chart.
Three explicit states are plotted (In Service / Spare-Unknown / Out of
Service) so the chart's "down" area matches the OOS-days count exactly --
spare/unknown days are NOT lumped in with OOS. The chart is windowed by the
page's day selector; a per-gen summary line states the window and the
In-Service / OOS / Spare counts WITHIN that window, so the chart can never be
silently compared against an all-time OOS number from elsewhere.
"""
# Plotted level per day: Out of Service = 0, Spare/Unknown = 1, In Service = 2.
def _level(point):
if point.get('in_service', False):
return 2
if point.get('out_of_service', False):
return 0
return 1 # spare / unknown -- assigned to nothing this day
window_label = 'All Historical Data' if str(days_back) == 'all' else f'Last {days_back} Days'
# Prepare data for Chart.js
chart_data = {}
for gen, timeline in generator_timeline.items():
levels = [_level(p) for p in timeline]
chart_data[gen] = {
'dates': [point['date_str'] for point in timeline],
'status': levels,
'pods': [point.get('pod') or ('Out of Service' if point.get('out_of_service', False) else 'Spare/Unknown') for point in timeline],
'reasons': [point.get('reason', '') for point in timeline],
# Windowed counts -- match exactly what the chart draws.
'counts': {
'in_service': sum(1 for lv in levels if lv == 2),
'oos': sum(1 for lv in levels if lv == 0),
'spare': sum(1 for lv in levels if lv == 1),
}
}
html = f"""
Generator Status Timeline
■ In Service■ Spare/Unknown■ Out of Service
"""
return html
def compute_daily_avg_oos_duration(all_historical_data):
"""
For each day D, walk every gen forward and accumulate a "consecutive days
OOS as of D" counter (reset to 0 whenever the gen has a non-OOS day).
Then average those counters across the gens that are OOS on D.
Returns: {date_str: avg_consecutive_oos_days}
Computed over the full historical_data — NOT the windowed subset — so a
gen that's been OOS for 100 days going into a 30-day chart window reads
as ~100, not 1.
"""
# Per-gen, date-ordered list of (date_str, is_oos)
per_gen = defaultdict(list)
for day in all_historical_data:
oos_set = set(day['oos_generators'].keys())
for gen in day['all_generators']:
per_gen[gen].append((day['date_str'], gen in oos_set))
day_runs = defaultdict(list)
for gen, entries in per_gen.items():
run = 0
for date_str, is_oos in entries:
if is_oos:
run += 1
day_runs[date_str].append(run)
else:
run = 0
return {d: (sum(r) / len(r)) for d, r in day_runs.items()}
def render_oos_count_chart(historical_data, daily_avg_oos):
"""
Daily OOS-count trend across the selected window. One line, one value per
day = number of gens with status='out_of_service' on that day in
gen_status.db. Useful for spotting drift (slowly accumulating OOS gens)
vs. discrete bumps (a site pulled a batch).
Second line (right y-axis): average consecutive OOS days across the gens
that were OOS on that date — answers "of the gens currently OOS, how long
on average have they been OOS?". Computed from full history so durations
reflect real run length, not just within the chart window.
"""
labels = [d['date_str'] for d in historical_data]
oos_counts = [len(d['oos_generators']) for d in historical_data]
avg_durations = [round(daily_avg_oos.get(d, 0), 1) for d in labels]
if not labels:
return ""
# Convenience numbers shown above the chart
latest = oos_counts[-1]
peak = max(oos_counts)
peak_date = labels[oos_counts.index(peak)]
first = oos_counts[0]
delta = latest - first
delta_str = f"+{delta}" if delta > 0 else str(delta)
return f"""
Daily Out-of-Service Count
Count of generators flagged Out of Service each day across the selected window.
Source: gen_status.db (one snapshot per day).
Latest
{latest}
Window Start ({labels[0]})
{first}
Change
{delta_str}
Peak ({peak_date})
{peak}
"""
def render_oos_summary(generator_metrics, generator_timeline):
"""Render out-of-service summary section."""
currently_oos = {gen: metrics for gen, metrics in generator_metrics.items()
if metrics['current_status'] == 'out_of_service'}
# Also show historically OOS generators
historically_oos = {gen: metrics for gen, metrics in generator_metrics.items()
if metrics['total_oos_incidents'] > 0}
html = """
Out of Service Summary
"""
if currently_oos:
html += f"""
Currently Out of Service ({len(currently_oos)} generators)
Generator
Current Days OOS
Total OOS Days
OOS Incidents
Current Reason
Last Pod
"""
# Sort by days OOS (longest first)
sorted_oos = sorted(currently_oos.items(),
key=lambda x: x[1]['current_oos_days'], reverse=True)
for gen, metrics in sorted_oos:
last_pod = "Unknown"
# Scan backwards through timeline to find last time this generator was not OOS
if gen in generator_timeline:
timeline = generator_timeline[gen]
for entry in reversed(timeline):
if entry['in_service'] and entry['pod']:
last_pod = entry['pod']
break
current_reason = metrics.get('current_reason', 'Unknown reason') or 'Unknown reason'
oos_class = "critical" if metrics['current_oos_days'] >= 7 else "warning" if metrics['current_oos_days'] >= 3 else "recent"
html += f"""
{gen}
{metrics['current_oos_days']}
{metrics['total_oos_days']}
{metrics['total_oos_incidents']}
{current_reason[:40] + '...' if len(current_reason) > 40 else current_reason}
{last_pod}
"""
html += """
"""
else:
html += """
✅ All generators are currently in service!
"""
# Add historical OOS summary
if historically_oos:
html += f"""
Historical Out of Service Summary (Top 10 by Total OOS Days)
Generator
Status
Total OOS Days
OOS Incidents
Avg OOS Duration
All Reasons
"""
# Sort by total OOS days (highest first) and take top 10
sorted_historical = sorted(historically_oos.items(),
key=lambda x: x[1]['total_oos_days'], reverse=True)[:10]
for gen, metrics in sorted_historical:
status_text = "In Service" if metrics['current_status'] == 'in_service' else "Out of Service"
status_class = "good" if metrics['current_status'] == 'in_service' else "critical"
# Collect all historical reasons
all_reasons = []
for reason_info in metrics.get('historical_reasons', []):
all_reasons.append(reason_info['reason'])
reasons_display = "; ".join(all_reasons) if all_reasons else "No reasons recorded"
avg_duration = f"{metrics['avg_oos_duration']:.1f}" if metrics['avg_oos_duration'] > 0 else "—"
html += f"""
{gen}
{status_text}
{metrics['total_oos_days']}
{metrics['total_oos_incidents']}
{avg_duration}
{reasons_display[:60] + '...' if len(reasons_display) > 60 else reasons_display}
"""
html += """
"""
html += "
"
return html
def render_swap_section(episodes, summary, window_label, meta,
oos_labels=None, oos_counts=None):
"""Render the 'Awaiting Swap' section: KPI cards + open-swap aging.
Swap is deliberately NOT folded into the OOS charts above. OOS is a location
(the gen sits in an Out of Service group and the daily snapshot says so);
swap is a tag on a gen that is usually still bolted to a pod and making
power. Every gen tagged swap today reads 'active' in gen_status.db, so these
gens contribute nothing to the OOS series -- merging them would silently mix
two different populations measured on two different time bases.
The window selector applies only to the completed-swap stats. Open swaps are
current state: a 52-day-old open swap does not stop being open because
someone asked for the last 7 days.
"""
open_eps = sorted(
[e for e in episodes if e["open"]],
key=lambda e: (e["days"] is not None, e["days"] or 0),
reverse=True,
)
def _fmt_days(value, suffix="d"):
if value is None:
return "--"
return f"{value:.1f}{suffix}"
# Bands drive both the bar colours and the table row tint.
def _band(days):
if days is None:
return "unknown"
if days >= 30:
return "critical"
if days >= 14:
return "warning"
return "recent"
BAND_COLORS = {
"critical": "#ff4444",
"warning": "#ff6600",
"recent": "#4a9eff",
"unknown": "#888888",
}
# Chart data. Bars are labelled "GEN (Site)" so the same bar is identifiable
# without cross-referencing the table below it.
bar_labels = [f'{e["gen_id"]} ({e["site"] or "?"})' for e in open_eps]
bar_values = [round(e["days"], 1) if e["days"] is not None else 0 for e in open_eps]
bar_colors = [BAND_COLORS[_band(e["days"])] for e in open_eps]
bar_meta = [
{
"status": e["status"],
"reason": e["reason"] or "No reason recorded",
"estimated": bool(e["start_estimated"]),
"start": e["start"].strftime("%Y-%m-%d") if e["start"] else "unknown",
}
for e in open_eps
]
# ~24px per bar keeps long lists readable instead of squashing them.
chart_height = max(220, len(open_eps) * 26 + 60)
first_swap = min((e["start"] for e in episodes if e["start"]), default=None)
history_note = (
f"Swap history begins {first_swap.strftime('%Y-%m-%d')}"
if first_swap else "No swap history recorded"
)
net = summary["net_change"]
net_str = f"+{net}" if net > 0 else str(net)
resolved_note = ""
if summary["resolved_count"]:
n = summary["resolved_count"]
resolved_note = (f"; {n} more had the tag cleared without a move "
f"(repaired in place or re-triaged)")
est_open = summary["estimated_open"]
est_closed = summary["estimated_closed"]
caveat_bits = []
if est_open:
caveat_bits.append(
f"{est_open} open swap{'s' if est_open != 1 else ''} had no logged start "
f"(age shown is a lower bound)"
)
if est_closed:
caveat_bits.append(
f"{est_closed} resolved swap{'s' if est_closed != 1 else ''} had no logged "
f"end and are excluded from duration stats"
)
caveat = (" · " + "; ".join(caveat_bits)) if caveat_bits else ""
html = f"""
Awaiting Swap
Generators tagged swap or active-swap — condemned and
queued for replacement, but still assigned to a pod rather than moved
to the Out of Service group, so they appear in no OOS chart above.
active-swap means the gen is still running and carrying load;
plain swap means it is down, waiting to be pulled.
Source: gen_history.db note log, reconciled against the live tag in
master_config.json, with replacement confirmed against the location
record in gen_status.db. {history_note}{caveat}.
Open Swaps
{summary['open_count']}
Down, Awaiting Swap
{summary['open_down']}
Still Running
{summary['open_running']}
Oldest Open
{_fmt_days(summary['oldest_open_days'])}
Median Open Age
{_fmt_days(summary['median_open_days'])}
Replaced ({window_label})
{summary['replaced_count']}
Over {window_label}: {summary['tagged_count']} tagged,
{summary['replaced_count']} pulled to an Out of Service group
(net {net_str}){resolved_note}.
Of the swaps that finished, the tag was on for a median of
{_fmt_days(summary['median_days_to_pull'])} — but that only counts swaps that
completed, so it can't see a backlog. The {summary['open_count']} still open
(median {_fmt_days(summary['median_open_days'])}) are not in it. The chart below is
what shows the backlog.
"""
# --- Backlog over time ------------------------------------------------
# The one chart where swap and OOS belong on the same axis: both are counts
# of gens unavailable to the fleet on a given day, and the interesting thing
# is the gap between them. Drawn on the OOS chart's own date labels so the
# two series are sampled identically -- the swap series is nulled before the
# note log starts rather than drawn as zeros.
backlog_labels = list(oos_labels or [])
if backlog_labels:
backlog = daily_open_counts(episodes, backlog_labels)
plotted = [v for v in backlog if v is not None]
gap_note = ""
if plotted and len(plotted) < len(backlog):
first_real = backlog_labels[backlog.index(plotted[0])]
gap_note = (f" The series starts {first_real} — the swap note log "
f"doesn't reach further back, so earlier days are left blank "
f"rather than drawn as zero.")
peak = max(plotted) if plotted else 0
peak_day = backlog_labels[backlog.index(peak)] if plotted else "--"
html += f"""
Swap Backlog Over Time
Generators carrying an open swap tag on each day, against the Out-of-Service
count for the same day. A rising swap line means gens are being condemned
faster than they're being pulled. Peak in this window: {peak} on {peak_day}.{gap_note}
"""
if not open_eps:
html += """
✅ No generators are currently awaiting swap.
"""
return html
html += f"""
Open Swap Aging ({len(open_eps)} generators)
■ under 14 days
■ 14–30 days
■ 30+ days
Generator
Days Waiting
Status
Site
Group
Tagged
Reason
"""
for ep in open_eps:
reason = ep["reason"] or "No reason recorded"
reason_short = reason if len(reason) <= 45 else reason[:45] + "..."
days_cell = _fmt_days(ep["days"])
if ep["start_estimated"]:
days_cell = "≥ " + days_cell
tagged = ep["start"].strftime("%Y-%m-%d") if ep["start"] else "unknown"
html += f"""
{_html.escape(ep['gen_id'])}
{days_cell}
{_html.escape(ep['status'])}
{_html.escape(ep['site'] or '--')}
{_html.escape(ep['group'] or '--')}
{tagged}
{_html.escape(reason_short)}
"""
html += """
"""
return html
def main():
print("Content-Type: text/html\n")
form = cgi.FieldStorage()
days_param = form.getvalue('days', '30')
days_back = 'all' if days_param == 'all' else int(days_param)
export_format = form.getvalue('export', '')
# Load ALL historical data once for performance
all_historical_data, error = load_all_historical_data()
if error:
print(f"
Error: {error}
")
return
# Filter data based on selected time period
historical_data = filter_historical_data(all_historical_data, days_back)
if not historical_data:
print("
No historical data available for selected time period
")
return
# Process data
generator_timeline, generator_stats, generator_records = track_generator_status(historical_data)
generator_metrics = calculate_generator_metrics(generator_stats, generator_timeline, generator_records)
# Handle CSV export
if export_format == 'csv':
print("""
Generator Trends CSV Export
Generator Trends CSV Data
Instructions: Select all the text in the box below, copy it (Ctrl+C), then paste it into Google Sheets.
Generator,Status,Current_Pod_Reason,Uptime_Percent,Total_OOS_Days,OOS_Incidents,Avg_OOS_Duration,Max_OOS_Duration,Current_OOS_Days,All_Historical_Reasons""")
for gen, metrics in sorted(generator_metrics.items()):
if metrics['current_status'] == 'out_of_service':
pod_reason = metrics.get('current_reason', 'Unknown reason') or 'Unknown reason'
else:
pod_reason = metrics['current_pod'] or ""
avg_duration = f"{metrics['avg_oos_duration']:.1f}" if metrics['avg_oos_duration'] > 0 else "0"
# Collect all historical reasons
all_reasons = []
for reason_info in metrics.get('historical_reasons', []):
all_reasons.append(reason_info['reason'])
reasons_text = "; ".join(all_reasons) if all_reasons else ""
print(f"{gen},{metrics['current_status']},\"{pod_reason}\",{metrics['uptime_percentage']:.1f},{metrics['total_oos_days']},{metrics['total_oos_incidents']},{avg_duration},{metrics['max_oos_duration']},{metrics['current_oos_days']},\"{reasons_text}\"")
print("""
Historical analysis of generator status and out-of-service tracking
Showing {len(historical_data)} data points{" over all historical data" if str(days_back) == "all" else f" over {days_back} days"}
""")
# Summary statistics
# Count each status directly so "Out of Service" matches the daily OOS chart
# (gens explicitly flagged out_of_service). Spares/unassigned gens have
# current_status == 'unknown' and are surfaced as their own stat rather than
# being folded into OOS via subtraction.
total_generators = len(generator_metrics)
in_service = sum(1 for m in generator_metrics.values() if m['current_status'] == 'in_service')
out_of_service = sum(1 for m in generator_metrics.values() if m['current_status'] == 'out_of_service')
spare = total_generators - in_service - out_of_service
avg_uptime = statistics.mean([m['uptime_percentage'] for m in generator_metrics.values()]) if generator_metrics else 0
print(f"""
Total Generators
{total_generators}
In Service
{in_service}
Out of Service
{out_of_service}
Spare
{spare}
Fleet Uptime
{avg_uptime:.1f}%
""")
# Render main sections
# Run-length math is done over the full, unfiltered series so durations
# don't artificially read low at the start of a short window.
daily_avg_oos = compute_daily_avg_oos_duration(all_historical_data)
print(render_oos_count_chart(historical_data, daily_avg_oos))
print(render_oos_summary(generator_metrics, generator_timeline))
# Swap tracking. Separate source (gen_notes event log, not the daily
# snapshot) and a separate population, so it gets its own section rather
# than extra series on the OOS charts -- see render_swap_section().
swap_episodes, swap_meta = load_swap_episodes(current_tags=load_current_swap_tags())
swap_window_start = None if str(days_back) == 'all' else datetime.now() - timedelta(days=days_back)
swap_window_label = 'all time' if str(days_back) == 'all' else f'{days_back}d'
swap_summary = summarize_swaps(swap_episodes, window_start=swap_window_start)
if swap_meta.get('error'):
print(f'
Awaiting Swap
'
f'
Swap history unavailable: '
f'{_html.escape(swap_meta["error"])}
')
else:
print(render_swap_section(swap_episodes, swap_summary,
swap_window_label, swap_meta,
oos_labels=[d['date_str'] for d in historical_data],
oos_counts=[len(d['oos_generators']) for d in historical_data]))
print(render_fleet_overview(generator_metrics))
print(render_timeline_chart(generator_timeline, days_back))
print(f"""
Analysis generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} |
Data range: {historical_data[0]['date_str']} to {historical_data[-1]['date_str']}