File indexing completed on 2026-09-01 09:34:18
0001
0002
0003 import os
0004 import sys
0005 from datetime import datetime
0006 import yaml
0007
0008 import matplotlib.pyplot as plt
0009 import numpy as np
0010 from matplotlib.backends.backend_pdf import PdfPages
0011 from matplotlib.offsetbox import AnchoredOffsetbox, HPacker, TextArea, VPacker
0012
0013 from argparsing import submission_args
0014 from sphenixdbutils import dbQuery, cnxn_string_map, list_to_condition
0015 from sphenixprodrules import RuleConfig
0016 from sphenixmisc import setup_rot_handler
0017 from simpleLogger import slogger, CustomFormatter, CHATTY, DEBUG, INFO, WARN, ERROR, CRITICAL
0018
0019
0020 def get_time_diffs_by_cpus(run_condition, dsttype, tag, dataset):
0021 query = f"""
0022 SELECT started, finished, request_cpus
0023 FROM production_jobs
0024 WHERE {run_condition}
0025 AND tag = '{tag}'
0026 AND dataset = '{dataset}'
0027 AND status = 'finished'
0028 AND dsttype LIKE '{dsttype}%'
0029 AND started IS NOT NULL
0030 AND finished IS NOT NULL
0031 """
0032
0033 DEBUG(f"Executing query:\n{query}")
0034
0035 cursor = dbQuery(cnxn_string_map['statr'], query)
0036 if not cursor:
0037 ERROR("Failed to query production database.")
0038 return None
0039
0040 results = cursor.fetchall()
0041 if not results:
0042 return {}
0043
0044 time_diffs_by_cpus = {}
0045 for started, finished, request_cpus in results:
0046 if isinstance(started, str):
0047 started = datetime.fromisoformat(started)
0048 if isinstance(finished, str):
0049 finished = datetime.fromisoformat(finished)
0050
0051 cpu_key = int(request_cpus) if request_cpus is not None else None
0052 time_diffs_by_cpus.setdefault(cpu_key, []).append((finished - started).total_seconds())
0053
0054 return time_diffs_by_cpus
0055
0056
0057 def flatten_time_diffs(time_diffs_by_cpus):
0058 return [diff for diffs in time_diffs_by_cpus.values() for diff in diffs]
0059
0060
0061 def cpu_sort_key(cpu_count):
0062 return (cpu_count is None, cpu_count if cpu_count is not None else 0)
0063
0064
0065 def cpu_label(cpu_count):
0066 if cpu_count is None:
0067 return "unknown CPUs"
0068 if cpu_count == 1:
0069 return "1 CPU"
0070 return f"{cpu_count} CPUs"
0071
0072
0073 def get_status_counts(run_condition, dsttype, tag, dataset):
0074 query = f"""
0075 SELECT
0076 SUM(CASE WHEN status = 'finished' THEN 1 ELSE 0 END) AS finished,
0077 SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS failed
0078 FROM production_jobs
0079 WHERE {run_condition}
0080 AND tag = '{tag}'
0081 AND dataset = '{dataset}'
0082 AND status IN ('finished', 'failed')
0083 AND dsttype LIKE '{dsttype}%'
0084 """
0085
0086 DEBUG(f"Executing query:\n{query}")
0087
0088 cursor = dbQuery(cnxn_string_map['statr'], query)
0089 if not cursor:
0090 ERROR("Failed to query production database.")
0091 return None
0092
0093 row = cursor.fetchone()
0094 finished = int(row.finished or 0)
0095 failed = int(row.failed or 0)
0096 return {
0097 "finished": finished,
0098 "failed": failed,
0099 "total": finished + failed,
0100 }
0101
0102
0103 def pick_unit(time_diffs_seconds):
0104 median = np.median(time_diffs_seconds)
0105 if median > 1800:
0106 return 'hours'
0107 if median > 30:
0108 return 'minutes'
0109 return 'seconds'
0110
0111
0112 def plot_histogram(ax, time_diffs_by_cpus, title, time_unit='hours'):
0113 """
0114 Plots wall-time histograms grouped by request_cpus with an overflow bin.
0115 """
0116 all_time_diffs_seconds = flatten_time_diffs(time_diffs_by_cpus)
0117 max_time_hours = np.max(all_time_diffs_seconds) / 3600 if all_time_diffs_seconds else 0
0118
0119 config = {
0120 'hours': {'conv': 3600, 'label': 'hours', 'max_val': 30, 'bin_w': 0.25, 'tick_step': 2},
0121 'minutes': {'conv': 60, 'label': 'minutes', 'max_val': 600, 'bin_w': 10, 'tick_step': 60},
0122 'seconds': {'conv': 1, 'label': 'seconds', 'max_val': 1200, 'bin_w': 20, 'tick_step': 120},
0123 }
0124
0125 if time_unit not in config:
0126 raise ValueError("Invalid time_unit. Must be 'hours', 'minutes', or 'seconds'.")
0127
0128 cfg = config[time_unit]
0129 max_val = cfg['max_val']
0130 bin_width = cfg['bin_w']
0131 tick_step = cfg['tick_step']
0132
0133 if time_unit == 'hours' and max_time_hours < 10:
0134 max_val = 10
0135 bin_width = 10 / 60
0136 tick_step = 1
0137
0138 bins = np.arange(0, max_val + bin_width, bin_width)
0139
0140 for cpu_count in sorted(time_diffs_by_cpus, key=cpu_sort_key):
0141 time_diffs = [t / cfg['conv'] for t in time_diffs_by_cpus[cpu_count]]
0142 plot_data = [min(diff, max_val) for diff in time_diffs]
0143 avg_time = np.mean(time_diffs)
0144 ax.hist(
0145 plot_data,
0146 bins=bins,
0147 histtype='step',
0148 linewidth=2,
0149 label=f'{cpu_label(cpu_count)} (n={len(time_diffs)}, avg={avg_time:.2f} {cfg["label"]})',
0150 )
0151
0152 ax.set_title(title)
0153 ax.set_xlabel(f'Wall time (start to finish) ({cfg["label"]})')
0154 ax.set_ylabel('Number of Jobs')
0155 ax.legend(loc='upper left')
0156 ax.grid(True, which='both', linestyle='--', linewidth=0.5)
0157
0158 ax.set_xlim(0, max_val)
0159 xticks = np.arange(0, max_val + bin_width, tick_step)
0160
0161 if max_val not in xticks:
0162 xticks = np.append(xticks, max_val)
0163
0164 xticklabels = [f'{t:g}' for t in xticks]
0165 xticklabels[-1] = f'{int(max_val)}+'
0166
0167 ax.set_xticks(xticks)
0168 ax.set_xticklabels(xticklabels)
0169
0170
0171 def add_status_box(ax, status_counts):
0172 if status_counts is None:
0173 return
0174
0175 rows = [
0176 ("finished:", status_counts["finished"], "black"),
0177 ("failed:", status_counts["failed"], "red"),
0178 ("total:", status_counts["total"], "black"),
0179 ]
0180
0181 label_column = VPacker(
0182 children=[
0183 TextArea(label, textprops={"color": color, "ha": "left"})
0184 for label, _, color in rows
0185 ],
0186 align="left",
0187 pad=0,
0188 sep=2,
0189 )
0190 value_column = VPacker(
0191 children=[
0192 TextArea(f"{value}", textprops={"color": color, "ha": "right"})
0193 for _, value, color in rows
0194 ],
0195 align="right",
0196 pad=0,
0197 sep=2,
0198 )
0199 status_box = HPacker(
0200 children=[label_column, value_column],
0201 align="baseline",
0202 pad=0,
0203 sep=12,
0204 )
0205
0206 anchored_box = AnchoredOffsetbox(
0207 loc="upper right",
0208 child=status_box,
0209 bbox_to_anchor=(0.98, 0.95),
0210 bbox_transform=ax.transAxes,
0211 frameon=True,
0212 borderpad=0,
0213 pad=0.35,
0214 )
0215 anchored_box.patch.set_boxstyle("round,pad=0.35")
0216 anchored_box.patch.set_facecolor("white")
0217 anchored_box.patch.set_edgecolor("0.5")
0218 anchored_box.patch.set_alpha(0.85)
0219 ax.add_artist(anchored_box)
0220
0221
0222 def main():
0223 """
0224 Main function to plot job time distribution.
0225 """
0226 args = submission_args()
0227
0228 plt.rcParams.update({'font.size': 16})
0229
0230 sublogdir = setup_rot_handler(args)
0231 slogger.setLevel(args.loglevel)
0232 INFO(f"Logging to {sublogdir}, level {args.loglevel}")
0233
0234 param_overrides = {}
0235 param_overrides["runs"] = args.runs
0236 param_overrides["runlist"] = args.runlist
0237 param_overrides["nevents"] = 0
0238
0239 if args.physicsmode is not None:
0240 param_overrides["physicsmode"] = args.physicsmode
0241
0242 param_overrides["prodmode"] = "production"
0243 try:
0244 rule = RuleConfig.from_yaml_file(
0245 yaml_file=args.config,
0246 rule_name=args.rulename,
0247 param_overrides=param_overrides
0248 )
0249 INFO(f"Successfully loaded rule configuration: {args.rulename}")
0250 except (ValueError, FileNotFoundError) as e:
0251 ERROR(f"Error: {e}")
0252 sys.exit(1)
0253
0254 if args.runs and 1 < len(rule.runlist_int) <= 5:
0255 output_pdf_path = f'job_time_distribution_{args.rulename}.pdf'
0256 with PdfPages(output_pdf_path) as pdf:
0257 for run in rule.runlist_int:
0258 INFO(f"Processing run: {run}")
0259 run_condition = list_to_condition([run], name="runnumber")
0260 time_diffs_by_cpus = get_time_diffs_by_cpus(run_condition, rule.dsttype, rule.outtriplet, rule.dataset)
0261 status_counts = get_status_counts(run_condition, rule.dsttype, rule.outtriplet, rule.dataset)
0262
0263 if time_diffs_by_cpus is None:
0264 sys.exit(1)
0265 all_time_diffs = flatten_time_diffs(time_diffs_by_cpus)
0266 if not all_time_diffs:
0267 INFO(f"No finished jobs found for run {run}.")
0268 continue
0269
0270 fig, ax = plt.subplots(figsize=(12, 7))
0271 plt.style.use('seaborn-v0_8-deep')
0272
0273 title = f'Job Time Distribution for {args.rulename} (Run: {run})'
0274 plot_histogram(ax, time_diffs_by_cpus, title, time_unit=pick_unit(all_time_diffs))
0275 add_status_box(ax, status_counts)
0276
0277 plt.tight_layout()
0278 pdf.savefig(fig)
0279 plt.close(fig)
0280
0281 INFO(f"Saved multi-page PDF to {output_pdf_path}")
0282
0283 else:
0284 run_condition = list_to_condition(rule.runlist_int, name="runnumber")
0285 time_diffs_by_cpus = get_time_diffs_by_cpus(run_condition, rule.dsttype, rule.outtriplet, rule.dataset)
0286 status_counts = get_status_counts(run_condition, rule.dsttype, rule.outtriplet, rule.dataset)
0287 if time_diffs_by_cpus is None:
0288 sys.exit(1)
0289 all_time_diffs = flatten_time_diffs(time_diffs_by_cpus)
0290 if not all_time_diffs:
0291 INFO("No finished jobs found for the specified runs.")
0292 sys.exit(0)
0293
0294 run_str = f"Run(s): {rule.runlist_int}"
0295 if rule.runlist is not None:
0296 run_str = f"Runs from file: {os.path.basename(rule.runlist)}"
0297
0298 base_title = f'Job Time Distribution for {args.rulename}\n{run_str}'
0299
0300 unit = pick_unit(all_time_diffs)
0301 fig, ax = plt.subplots(figsize=(12, 7))
0302 plt.style.use('seaborn-v0_8-deep')
0303
0304 plot_histogram(ax, time_diffs_by_cpus, base_title, time_unit=unit)
0305 add_status_box(ax, status_counts)
0306
0307 plt.tight_layout()
0308 output_file = f'job_time_distribution_{args.rulename}.png'
0309 plt.savefig(output_file)
0310 INFO(f"Saved plot to {output_file}")
0311 plt.close(fig)
0312
0313
0314 if __name__ == '__main__':
0315 main()
