File indexing completed on 2026-08-12 09:36:11
0001
0002
0003 import os
0004 import sys
0005 import yaml
0006
0007 import matplotlib.pyplot as plt
0008 import numpy as np
0009 from matplotlib.backends.backend_pdf import PdfPages
0010 from matplotlib.offsetbox import AnchoredOffsetbox, HPacker, TextArea, VPacker
0011
0012 from argparsing import submission_args
0013 from sphenixdbutils import dbQuery, cnxn_string_map, list_to_condition
0014 from sphenixprodrules import RuleConfig
0015 from sphenixmisc import setup_rot_handler
0016 from simpleLogger import slogger, CustomFormatter, CHATTY, DEBUG, INFO, WARN, ERROR, CRITICAL
0017
0018
0019 def get_memory_values(run_condition, dsttype, tag, dataset):
0020 query = f"""
0021 SELECT MemoryProvisioned, MemoryUsage
0022 FROM production_jobs
0023 WHERE {run_condition}
0024 AND tag = '{tag}'
0025 AND dataset = '{dataset}'
0026 AND status IN ('finished', 'failed')
0027 AND dsttype LIKE '{dsttype}%'
0028 AND MemoryProvisioned IS NOT NULL
0029 AND MemoryUsage IS NOT NULL
0030 AND MemoryProvisioned > 0
0031 AND MemoryUsage > 0
0032 """
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 memory_values = []
0046 for provisioned_mb, usage_mb in results:
0047 memory_values.append((float(provisioned_mb) / 1024.0, float(usage_mb) / 1024.0))
0048
0049 return memory_values
0050
0051
0052 def get_status_counts(run_condition, dsttype, tag, dataset):
0053 query = f"""
0054 SELECT
0055 SUM(CASE WHEN status = 'finished' THEN 1 ELSE 0 END) AS finished,
0056 SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS failed
0057 FROM production_jobs
0058 WHERE {run_condition}
0059 AND tag = '{tag}'
0060 AND dataset = '{dataset}'
0061 AND status IN ('finished', 'failed')
0062 AND dsttype LIKE '{dsttype}%'
0063 """
0064
0065 DEBUG(f"Executing query:\n{query}")
0066
0067 cursor = dbQuery(cnxn_string_map['statr'], query)
0068 if not cursor:
0069 ERROR("Failed to query production database.")
0070 return None
0071
0072 row = cursor.fetchone()
0073 finished = int(row.finished or 0)
0074 failed = int(row.failed or 0)
0075 return {
0076 "finished": finished,
0077 "failed": failed,
0078 "total": finished + failed,
0079 }
0080
0081
0082 def _axis_limits(*arrays):
0083 values = np.concatenate([np.asarray(a, dtype=float) for a in arrays if len(a)])
0084 if values.size == 0:
0085 return 0, 15
0086
0087 low = max(0, float(np.min(values)))
0088 high = float(np.max(values))
0089 span = high - low
0090 pad = max(0.5, span * 0.08)
0091
0092 axis_min = min(max(0, low - pad), 14.5)
0093 return axis_min, 15
0094
0095
0096 def plot_memory(ax_hist, ax_scatter, memory_values, title):
0097 provisioned_gb = np.array([v[0] for v in memory_values])
0098 usage_gb = np.array([v[1] for v in memory_values])
0099
0100 min_axis, max_axis = _axis_limits(provisioned_gb, usage_gb)
0101 bin_width = 0.5
0102 bins = np.arange(0, max_axis + bin_width, bin_width)
0103 if len(bins) < 2:
0104 bins = np.array([0, max_axis + bin_width])
0105
0106 ax_hist.hist(
0107 provisioned_gb,
0108 bins=bins,
0109 alpha=0.65,
0110 label=f'Provisioned (avg: {np.mean(provisioned_gb):.1f} GB)',
0111 )
0112 ax_hist.hist(
0113 usage_gb,
0114 bins=bins,
0115 alpha=0.65,
0116 label=f'Actual usage (avg: {np.mean(usage_gb):.1f} GB)',
0117 )
0118 ax_hist.set_title('Memory distribution')
0119 ax_hist.set_xlabel('Memory (GB)')
0120 ax_hist.set_ylabel('Number of jobs')
0121 ax_hist.set_xlim(min_axis, max_axis)
0122 ax_hist.legend(loc='upper right')
0123 ax_hist.grid(True, which='both', linestyle='--', linewidth=0.5)
0124
0125 ax_scatter.scatter(provisioned_gb, usage_gb, alpha=0.45, s=18, edgecolors='none')
0126 ax_scatter.plot([min_axis, max_axis], [min_axis, max_axis], 'r--', linewidth=1.5, label='usage = provisioned')
0127 ax_scatter.set_title('Actual vs provisioned memory')
0128 ax_scatter.set_xlabel('Memory provisioned (GB)')
0129 ax_scatter.set_ylabel('Actual memory usage (GB)')
0130 ax_scatter.set_xlim(min_axis, max_axis)
0131 ax_scatter.set_ylim(min_axis, max_axis)
0132 ax_scatter.set_aspect('equal', adjustable='box')
0133 ax_scatter.legend(loc='upper left')
0134 ax_scatter.grid(True, which='both', linestyle='--', linewidth=0.5)
0135
0136 ax_hist.figure.suptitle(title)
0137
0138
0139 def add_status_box(ax, status_counts, plotted_count):
0140 if status_counts is None:
0141 return
0142
0143 rows = [
0144 ("plotted:", plotted_count, "black"),
0145 ("finished:", status_counts["finished"], "black"),
0146 ("failed:", status_counts["failed"], "red"),
0147 ("total:", status_counts["total"], "black"),
0148 ]
0149
0150 label_column = VPacker(
0151 children=[
0152 TextArea(label, textprops={"color": color, "ha": "left"})
0153 for label, _, color in rows
0154 ],
0155 align="left",
0156 pad=0,
0157 sep=2,
0158 )
0159 value_column = VPacker(
0160 children=[
0161 TextArea(f"{value}", textprops={"color": color, "ha": "right"})
0162 for _, value, color in rows
0163 ],
0164 align="right",
0165 pad=0,
0166 sep=2,
0167 )
0168 status_box = HPacker(
0169 children=[label_column, value_column],
0170 align="baseline",
0171 pad=0,
0172 sep=12,
0173 )
0174
0175 anchored_box = AnchoredOffsetbox(
0176 loc="upper right",
0177 child=status_box,
0178 bbox_to_anchor=(0.98, 0.95),
0179 bbox_transform=ax.transAxes,
0180 frameon=True,
0181 borderpad=0,
0182 pad=0.35,
0183 )
0184 anchored_box.patch.set_boxstyle("round,pad=0.35")
0185 anchored_box.patch.set_facecolor("white")
0186 anchored_box.patch.set_edgecolor("0.5")
0187 anchored_box.patch.set_alpha(0.85)
0188 ax.add_artist(anchored_box)
0189
0190
0191 def run_label(rule):
0192 run_str = f"Run(s): {rule.runlist_int}"
0193 if rule.runlist is not None:
0194 run_str = f"Runs from file: {os.path.basename(rule.runlist)}"
0195 return run_str
0196
0197
0198 def make_plot(memory_values, status_counts, title, output_target):
0199 fig, (ax_hist, ax_scatter) = plt.subplots(1, 2, figsize=(18, 8))
0200 plt.style.use('seaborn-v0_8-deep')
0201
0202 plot_memory(ax_hist, ax_scatter, memory_values, title)
0203 add_status_box(ax_hist, status_counts, len(memory_values))
0204
0205 plt.tight_layout(rect=[0, 0, 1, 0.93])
0206 output_target(fig)
0207 plt.close(fig)
0208
0209
0210 def main():
0211 """
0212 Main function to plot job memory distribution.
0213 """
0214 args = submission_args()
0215
0216 plt.rcParams.update({'font.size': 16})
0217
0218 sublogdir = setup_rot_handler(args)
0219 slogger.setLevel(args.loglevel)
0220 INFO(f"Logging to {sublogdir}, level {args.loglevel}")
0221
0222 param_overrides = {}
0223 param_overrides["runs"] = args.runs
0224 param_overrides["runlist"] = args.runlist
0225 param_overrides["nevents"] = 0
0226
0227 if args.physicsmode is not None:
0228 param_overrides["physicsmode"] = args.physicsmode
0229
0230 param_overrides["prodmode"] = "production"
0231 if args.mangle_dirpath:
0232 param_overrides["prodmode"] = args.mangle_dirpath
0233
0234 try:
0235 rule = RuleConfig.from_yaml_file(
0236 yaml_file=args.config,
0237 rule_name=args.rulename,
0238 param_overrides=param_overrides
0239 )
0240 INFO(f"Successfully loaded rule configuration: {args.rulename}")
0241 except (ValueError, FileNotFoundError) as e:
0242 ERROR(f"Error: {e}")
0243 sys.exit(1)
0244
0245 if args.runs and 1 < len(rule.runlist_int) <= 5:
0246 output_pdf_path = f'job_memory_distribution_{args.rulename}.pdf'
0247 with PdfPages(output_pdf_path) as pdf:
0248 for run in rule.runlist_int:
0249 INFO(f"Processing run: {run}")
0250 run_condition = list_to_condition([run], name="runnumber")
0251 memory_values = get_memory_values(run_condition, rule.dsttype, rule.outtriplet, rule.dataset)
0252 status_counts = get_status_counts(run_condition, rule.dsttype, rule.outtriplet, rule.dataset)
0253
0254 if memory_values is None:
0255 sys.exit(1)
0256 if not memory_values:
0257 INFO(f"No jobs with memory values found for run {run}.")
0258 continue
0259
0260 title = f'Job Memory Distribution for {args.rulename}\nRun: {run}'
0261 make_plot(memory_values, status_counts, title, pdf.savefig)
0262
0263 INFO(f"Saved multi-page PDF to {output_pdf_path}")
0264
0265 else:
0266 run_condition = list_to_condition(rule.runlist_int, name="runnumber")
0267 memory_values = get_memory_values(run_condition, rule.dsttype, rule.outtriplet, rule.dataset)
0268 status_counts = get_status_counts(run_condition, rule.dsttype, rule.outtriplet, rule.dataset)
0269 if memory_values is None:
0270 sys.exit(1)
0271 if not memory_values:
0272 INFO("No jobs with memory values found for the specified runs.")
0273 sys.exit(0)
0274
0275 title = f'Job Memory Distribution for {args.rulename}\n{run_label(rule)}'
0276 output_file = f'job_memory_distribution_{args.rulename}.png'
0277 make_plot(memory_values, status_counts, title, lambda fig: fig.savefig(output_file))
0278 INFO(f"Saved plot to {output_file}")
0279
0280
0281 if __name__ == '__main__':
0282 main()
