sphenixprod/plotting/plot_job_throughput.py

0001 #!/usr/bin/env python
0002
0003 import sys
0004 from datetime import datetime, timezone, timedelta
0005
0006 import matplotlib.pyplot as plt
0007 import matplotlib.dates as mdates
0008 import numpy as np
0009
0010 from argparsing import submission_args
0011 from sphenixdbutils import dbQuery, cnxn_string_map
0012 from sphenixprodrules import RuleConfig
0013 from sphenixmisc import setup_rot_handler
0014 from simpleLogger import slogger, CustomFormatter, CHATTY, DEBUG, INFO, WARN, ERROR, CRITICAL  # noqa: F401
0015
0016
0017 # START_DATE    = datetime(2026, 4, 21, tzinfo=timezone.utc)
0018 START_DATE    = datetime(2026, 8, 1, tzinfo=timezone.utc)
0019 BIN_HOURS     = 0.5
0020 ROLLING_BINS  = 3   # rolling average window: 12 bins = 24 hours
0021
0022
0023 def get_start_times(dsttype, tag, dataset, since):
0024     query = f"""
0025     SELECT started
0026     FROM production_jobs
0027     WHERE tag = '{tag}'
0028       AND dataset = '{dataset}'
0029       AND status = 'finished'
0030       AND started >= '{since.isoformat()}'
0031       AND dsttype LIKE '{dsttype}%'
0032       AND started IS NOT NULL
0033     """
0034     DEBUG(f"Executing query:\n{query}")
0035
0036     cursor = dbQuery(cnxn_string_map['statr'], query)
0037     if not cursor:
0038         ERROR("Failed to query production database.")
0039         return None
0040
0041     results = cursor.fetchall()
0042     if not results:
0043         return []
0044
0045     times = []
0046     for (started,) in results:
0047         if isinstance(started, str):
0048             started = datetime.fromisoformat(started)
0049         if started.tzinfo is None:
0050             started = started.replace(tzinfo=timezone.utc)
0051         times.append(started)
0052
0053     return times
0054
0055
0056 def main():
0057     args = submission_args()
0058
0059     plt.rcParams.update({'font.size': 16})
0060
0061     sublogdir = setup_rot_handler(args)
0062     slogger.setLevel(args.loglevel)
0063     INFO(f"Logging to {sublogdir}, level {args.loglevel}")
0064
0065     param_overrides = {}
0066     param_overrides["runs"]     = args.runs
0067     param_overrides["runlist"]  = args.runlist
0068     param_overrides["nevents"]  = 0
0069
0070     if args.physicsmode is not None:
0071         param_overrides["physicsmode"] = args.physicsmode
0072
0073     param_overrides["prodmode"] = "production"
0074     try:
0075         rule = RuleConfig.from_yaml_file(
0076             yaml_file=args.config,
0077             rule_name=args.rulename,
0078             param_overrides=param_overrides
0079         )
0080         INFO(f"Successfully loaded rule configuration: {args.rulename}")
0081     except (ValueError, FileNotFoundError) as e:
0082         ERROR(f"Error: {e}")
0083         sys.exit(1)
0084
0085     start_times = get_start_times(rule.dsttype, rule.outtriplet, rule.dataset, START_DATE)
0086     if start_times is None:
0087         sys.exit(1)
0088     if not start_times:
0089         INFO(f"No finished jobs found since {START_DATE.date()}.")
0090         sys.exit(0)
0091
0092     INFO(f"Found {len(start_times)} finished jobs.")
0093
0094     now       = datetime.now(timezone.utc)
0095     bin_edges = mdates.drange(START_DATE, now + timedelta(hours=BIN_HOURS), timedelta(hours=BIN_HOURS))
0096     start_nums = [mdates.date2num(t) for t in start_times]
0097
0098     counts, _ = np.histogram(start_nums, bins=bin_edges)
0099     bin_centers = 0.5 * (bin_edges[:-1] + bin_edges[1:])
0100
0101     kernel  = np.ones(ROLLING_BINS) / ROLLING_BINS
0102     rolling = np.convolve(counts, kernel, mode='same')
0103
0104     fig, ax = plt.subplots(figsize=(16, 7))
0105     plt.style.use('seaborn-v0_8-deep')
0106
0107     ax.bar(bin_edges[:-1], counts, width=(bin_edges[1] - bin_edges[0]), alpha=0.5, align='edge', label=f'{BIN_HOURS}h bins')
0108     ax.plot(bin_centers, rolling, color='red', linewidth=2, label=f'{ROLLING_BINS * BIN_HOURS}h rolling avg')
0109
0110     ax.xaxis_date()
0111     ax.xaxis.set_major_locator(mdates.DayLocator(interval=1))
0112     ax.xaxis.set_minor_locator(mdates.HourLocator(interval=6))
0113     ax.xaxis.set_major_formatter(mdates.DateFormatter('%b %d'))
0114     plt.xticks(rotation=45, ha='right')
0115
0116     ax.set_xlim(bin_edges[0], bin_edges[-1])
0117     ax.set_title(f'Finished jobs by start time — {args.rulename}\n(since {START_DATE.date()}, {BIN_HOURS}h bins)')
0118     ax.set_xlabel('Job start time')
0119     ax.set_ylabel('Number of jobs finished')
0120     ax.legend()
0121     ax.grid(True, which='both', linestyle='--', linewidth=0.5)
0122
0123     plt.tight_layout()
0124     output_file = f'job_throughput_{args.rulename}.png'
0125     plt.savefig(output_file)
0126     INFO(f"Saved plot to {output_file}")
0127     plt.close(fig)
0128
0129
0130 if __name__ == '__main__':
0131     main()