brewpi/utils/analyze_log.py

#!/usr/bin/env python3
"""Offline analysis of a log (and optional forecast) produced by the BrewPi server.

Usage:
    python analyze_log.py <date_time> <sud_name> [--log-dir <dir>]
    python analyze_log.py <path/to/log_*.json>

The two-argument form loads logs/<date_time>_<sud_name>.json and its paired
forecast file.  The single-argument form accepts a direct path to any
log_*.json (e.g. a server-session log without a Sud name); the forecast panel
is shown only if a matching forecast_*.json exists alongside it.

Renders the 3-panel realtime strip chart and, when available, the
forecast-vs-actual comparison.
"""

import argparse
import json
import sys
from pathlib import Path

import matplotlib
matplotlib.use("TkAgg")
import matplotlib.pyplot as plt
from matplotlib.ticker import AutoMinorLocator


def _style_axis(ax):
    ax.set_facecolor('white')
    ax.spines['top'].set_visible(False)
    ax.spines['right'].set_visible(False)
    ax.xaxis.set_minor_locator(AutoMinorLocator())
    ax.yaxis.set_minor_locator(AutoMinorLocator())
    ax.tick_params(which='major', direction='out', labelsize='small')
    ax.tick_params(which='minor', direction='out', length=2)
    ax.grid(which='major', linestyle='-', linewidth=0.5, alpha=0.3)
    ax.grid(which='minor', linestyle=':', linewidth=0.5, alpha=0.15)


def plot_realtime(samples, name):
    """3-panel strip chart mirroring RealtimePlot in the PyQt GUI."""
    t          = [s['t']         for s in samples]
    temp_ist   = [s['temp_ist']  for s in samples]
    temp_soll  = [s['temp_soll'] for s in samples]
    rate_ist   = [s['rate_ist']  for s in samples]
    rate_soll  = [s['rate_soll'] for s in samples]
    power_set  = [s['power_set'] for s in samples]
    power_eff  = [s['power_eff'] for s in samples]

    fig, (ax_temp, ax_rate, ax_power) = plt.subplots(3, 1, sharex=True, facecolor='white')
    fig.suptitle(name, fontsize='small')

    ax_temp.plot(t, temp_ist,  '-b', linewidth=1, label='theta_ist')
    ax_temp.plot(t, temp_soll, '-r', linewidth=1, label='theta_soll')
    ax_temp.set_ylabel('Temp [°C]')
    ax_temp.legend(loc='upper left', fontsize='small')

    ax_rate.plot(t, rate_ist,  '-b', linewidth=1, label='heatrate_ist')
    ax_rate.plot(t, rate_soll, '-r', linewidth=1, label='heatrate_soll')
    ax_rate.set_ylabel('Rate [K/min]')
    ax_rate.legend(loc='upper left', fontsize='small')

    ax_power.plot(t, power_set, '-r', linewidth=1, label='power_set')
    ax_power.plot(t, power_eff, '-b', linewidth=1, label='power_eff')
    ax_power.set_ylabel('Power [W]')
    ax_power.set_xlabel('t [s]')
    ax_power.legend(loc='upper left', fontsize='small')

    for ax in (ax_temp, ax_rate, ax_power):
        _style_axis(ax)

    ax_temp.label_outer()
    ax_rate.label_outer()

    fig.tight_layout()
    return fig


def plot_forecast(forecast, samples, name):
    """Forecast-vs-actual comparison mirroring SudForecastPlot in the PyQt GUI."""
    t_min_fc = [t / 60.0 for t in forecast['T']]
    theta_fc  = forecast['Theta']
    finished  = forecast['Finished']

    t_min_actual = [s['t'] / 60.0  for s in samples]
    theta_actual  = [s['temp_ist'] for s in samples]

    fig, ax = plt.subplots(1, 1, facecolor='white')

    ax.plot(t_min_actual, theta_actual, '-b', linewidth=1.5, zorder=2, label='actual')
    ax.plot(t_min_fc,    theta_fc,      '-b', linewidth=1.5, alpha=0.5, zorder=2, label='forecast')

    total = t_min_fc[-1] if t_min_fc else 0.0
    suffix = '' if finished else ' so far'
    ax.set_title('{} - {:.0f} min{} total'.format(name, total, suffix), fontsize='small')
    ax.set_xlabel('t [min]')
    ax.set_ylabel('Temp [°C]')
    ax.legend(loc='upper left', fontsize='small')
    _style_axis(ax)

    fig.tight_layout()
    return fig


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('date_time_or_path',
                        help='Timestamp part of the filename (e.g. 20260628T184903) '
                             'or a direct path to a log_*.json file')
    parser.add_argument('sud_name', nargs='?', default=None,
                        help='Sud name part of the filename (e.g. Sud-0010); '
                             'omit when passing a direct file path')
    default_log_dir = Path(__file__).parent.parent / 'logs'
    parser.add_argument('--log-dir', default=str(default_log_dir),
                        help='Directory containing the log files (default: %(default)s)')
    args = parser.parse_args()

    if args.sud_name is None:
        log_path = Path(args.date_time_or_path)
        stem = log_path.stem[len('log_'):] if log_path.stem.startswith('log_') else log_path.stem
        forecast_path = log_path.parent / 'forecast_{}.json'.format(stem)
    else:
        log_dir = Path(args.log_dir)
        stem = '{}_{}'.format(args.date_time_or_path, args.sud_name)
        log_path      = log_dir / 'log_{}.json'.format(stem)
        forecast_path = log_dir / 'forecast_{}.json'.format(stem)

    if not log_path.exists():
        print('Error: file not found: {}'.format(log_path), file=sys.stderr)
        sys.exit(1)

    with open(log_path) as f:
        log = json.load(f)

    name = log.get('Name', stem)
    samples = log['Samples']

    plot_realtime(samples, name)
    if forecast_path.exists():
        with open(forecast_path) as f:
            forecast = json.load(f)
        plot_forecast(forecast, samples, name)

    plt.show()


if __name__ == '__main__':
    main()