File indexing completed on 2026-09-01 09:34:19
0001
0002 """
0003 generate_report.py
0004
0005 Generate a run-level CSV report for sPHENIX production rules.
0006
0007 For now this supports raw/event-combiner rules. Downstream rules are detected
0008 and rejected explicitly so the macro name and CLI can be reused later.
0009 """
0010
0011 import cProfile
0012 import csv
0013 import math
0014 import pstats
0015 import sys
0016 from collections import defaultdict
0017 from pathlib import Path
0018
0019 from argparsing import submission_args
0020 from simpleLogger import CHATTY, DEBUG, INFO, WARN, ERROR, CRITICAL
0021 from sphenixprodrules import RuleConfig
0022 from sphenixmatching import MatchConfig
0023 from sphenixdbutils import cnxn_string_map, dbQuery
0024 from sphenixmisc import human_event_count
0025
0026
0027 EXPECTED_SKIPPED_EVENTS_PER_DAQHOST = 2
0028
0029
0030 CSV_COLUMNS = [
0031 "rule_name",
0032 "runnumber",
0033 "possible_daqhosts",
0034 "total_daqhosts",
0035 "missing_daqhosts",
0036 "possible_segments",
0037 "total_segments",
0038 "missing_segments",
0039 "possible_events",
0040 "total_events",
0041 "missing_events",
0042 "error_codes",
0043 "incomplete_reasons",
0044 "status",
0045 ]
0046
0047
0048 def daqhost_to_dst_leaf(daqhost, match):
0049 if isinstance(match.input_stem, dict):
0050 for leaf, raw_daqhost in match.input_stem.items():
0051 if raw_daqhost == daqhost:
0052 return leaf
0053 return str(daqhost).replace(":", "_")
0054
0055
0056 def sql_literal(value):
0057 return "'" + str(value).replace("'", "''") + "'"
0058
0059
0060 def csv_join(values):
0061 return ";".join(str(value) for value in sorted(values, key=str))
0062
0063
0064
0065
0066 def output_path(args):
0067 return Path(args.output) if args.output else Path(f"{args.rulename}.csv")
0068
0069
0070 def load_rule_and_match(args):
0071 param_overrides = {}
0072 param_overrides["runs"] = args.runs
0073 param_overrides["runlist"] = args.runlist
0074 param_overrides["nevents"] = args.nevents
0075 param_overrides["prodmode"] = "production"
0076 if args.physicsmode:
0077 param_overrides["physicsmode"] = args.physicsmode
0078
0079 try:
0080 rule = RuleConfig.from_yaml_file(
0081 yaml_file=args.config,
0082 rule_name=args.rulename,
0083 param_overrides=param_overrides,
0084 )
0085 except (ValueError, FileNotFoundError) as e:
0086 ERROR(f"Error loading rule configuration: {e}")
0087 sys.exit(2)
0088
0089 return rule, MatchConfig.from_rule_config(rule)
0090
0091
0092
0093 def query_error_codes(args, match, run_condition):
0094
0095 query = f"""
0096 SELECT runnumber, ExitCode
0097 FROM production_jobs
0098 WHERE rulename={sql_literal(args.rulename)}
0099 AND dataset={sql_literal(match.dataset)}
0100 AND tag={sql_literal(match.outtriplet)}
0101 AND dsttype like {sql_literal(match.dst_type_template)}
0102 AND {run_condition}
0103 AND ExitCode IS NOT NULL
0104 AND ExitCode != 0
0105 ORDER BY runnumber, ExitCode
0106 """
0107 rows = dbQuery(cnxn_string_map["statr"], query).fetchall()
0108 codes_by_run = defaultdict(set)
0109 for runnumber, exit_code in rows:
0110 codes_by_run[int(runnumber)].add(int(exit_code))
0111 INFO(f"{sum(len(codes) for codes in codes_by_run.values())} production job exit codes found.")
0112 return codes_by_run
0113
0114
0115 def add_reason(reasons_by_run, runnumber, reason):
0116 reasons_by_run[int(runnumber)].add(reason)
0117
0118
0119 def write_csv_report(
0120 path,
0121 args,
0122 runnumbers,
0123 possible_daqhosts_by_run,
0124 total_daqhosts_by_run,
0125 possible_segments_by_run,
0126 total_segments_by_run,
0127 possible_events_by_run,
0128 total_events_by_run,
0129 error_codes_by_run,
0130 reasons_by_run,
0131 expected_skipped_events_by_run=None,
0132 ):
0133 rows_written = 0
0134 expected_skipped_events_by_run = expected_skipped_events_by_run or {}
0135 path.parent.mkdir(parents=True, exist_ok=True)
0136 with path.open("w", newline="") as handle:
0137 writer = csv.DictWriter(handle, fieldnames=CSV_COLUMNS)
0138 writer.writeheader()
0139 for runnumber in sorted(runnumbers):
0140 possible_daqhosts = int(possible_daqhosts_by_run.get(runnumber, 0))
0141 total_daqhosts = int(total_daqhosts_by_run.get(runnumber, 0))
0142 missing_daqhosts = max(possible_daqhosts - total_daqhosts, 0)
0143 possible_segments = int(possible_segments_by_run.get(runnumber, 0))
0144 total_segments = int(total_segments_by_run.get(runnumber, 0))
0145 missing_segments = max(possible_segments - total_segments, 0)
0146 possible_events = int(possible_events_by_run.get(runnumber, 0))
0147 total_events = int(total_events_by_run.get(runnumber, 0))
0148 expected_skipped_events = int(expected_skipped_events_by_run.get(runnumber, 0))
0149 missing_events = max(possible_events - total_events - expected_skipped_events, 0)
0150 error_codes = error_codes_by_run.get(runnumber, set())
0151
0152 reasons = set(reasons_by_run.get(runnumber, set()))
0153 if missing_daqhosts:
0154 reasons.add("missing_daqhosts")
0155 if missing_segments:
0156 reasons.add("missing_segments")
0157 if error_codes:
0158 reasons.add("error_codes")
0159
0160 writer.writerow({
0161 "rule_name": args.rulename,
0162 "runnumber": runnumber,
0163 "possible_daqhosts": possible_daqhosts,
0164 "total_daqhosts": total_daqhosts,
0165 "missing_daqhosts": missing_daqhosts,
0166 "possible_segments": possible_segments,
0167 "total_segments": total_segments,
0168 "missing_segments": missing_segments,
0169 "possible_events": possible_events,
0170 "total_events": total_events,
0171 "missing_events": missing_events,
0172 "error_codes": csv_join(error_codes),
0173 "incomplete_reasons": csv_join(reasons),
0174 "status": "incomplete" if missing_events else ("questionable" if reasons else "complete"),
0175 })
0176 rows_written += 1
0177
0178 INFO(f"Wrote {rows_written} run-level rows to {path}")
0179
0180
0181 def normalized_neventsper(job_config):
0182 neventsper = getattr(job_config, "neventsper", None)
0183 try:
0184 return int(neventsper) if neventsper is not None else 0
0185 except (TypeError, ValueError):
0186 return 0
0187
0188
0189
0190 def generate_eventcombiner_report(args, rule, match, report_path):
0191 daqhosts_dict, eventsinrun_by_run = match.daqhosts_for_combining()
0192 if not eventsinrun_by_run:
0193 INFO("No runs pass run quality cuts; no report written.")
0194 return False
0195
0196 neventsper = normalized_neventsper(rule.job_config)
0197 if neventsper:
0198 total_expected_outputs = sum(
0199 math.ceil(eventsinrun / neventsper)
0200 for eventsinrun in eventsinrun_by_run.values()
0201 if eventsinrun
0202 )
0203 INFO(f"{total_expected_outputs} expected downstream output files from events/neventsper={neventsper}.")
0204
0205 daqhost_types = [host for host in match.in_types if host != "gl1daq"]
0206 n_ideal = sum(sum(1 for host in hosts if host != "gl1daq") for hosts in daqhosts_dict.values())
0207 INFO(f"{n_ideal} (run, daqhost) combinations have all segments on lustre.")
0208
0209 run_condition = match._run_condition(list(eventsinrun_by_run))
0210
0211 total_query = f"""
0212 SELECT DISTINCT runnumber, daqhost FROM datasets
0213 WHERE {run_condition}
0214 AND daqhost IN {tuple(daqhost_types)}
0215 ORDER BY runnumber, daqhost
0216 """
0217 all_combos = dbQuery(cnxn_string_map["rawr"], total_query).fetchall()
0218 INFO(f"{len(all_combos)} (run, daqhost) combinations found in the raw DB.")
0219
0220 not_on_lustre = [(int(r), h) for r, h in all_combos if h not in daqhosts_dict.get(int(r), set())]
0221 INFO(f"{len(not_on_lustre)} (run, daqhost) combinations are in the DB but not fully on lustre.")
0222 for run, daqhost in not_on_lustre[:args.example_limit]:
0223 DEBUG(f" Not fully on lustre: Run {run} {daqhost}")
0224
0225 runs_without_gl1daq = {run for run, hosts in daqhosts_dict.items() if "gl1daq" not in hosts}
0226 for run in sorted(runs_without_gl1daq):
0227 WARN(f"Run {run}: gl1daq not complete on lustre - run will not be submitted.")
0228
0229 lustre_combos = [
0230 (run, daqhost)
0231 for run, hosts in daqhosts_dict.items()
0232 if "gl1daq" in hosts
0233 for daqhost in hosts
0234 if daqhost != "gl1daq"
0235 ]
0236
0237 lastevent_query = f"""
0238 SELECT runnumber, dsttype, max(lastevent)
0239 FROM datasets
0240 WHERE dataset='{match.dataset}'
0241 AND tag='{match.outtriplet}'
0242 AND dsttype like '{match.dst_type_template}'
0243 AND {run_condition}
0244 GROUP BY runnumber, dsttype
0245 ORDER BY runnumber, dsttype
0246 """
0247 rows = dbQuery(cnxn_string_map["fcr"], lastevent_query).fetchall()
0248 INFO(f"{len(rows)} (run, dsttype) combinations have existing output in the FileCatalog.")
0249
0250 fc_dsttypes_by_run = defaultdict(list)
0251 total_events_by_run = defaultdict(int)
0252 for runnumber, dsttype, lastevent in rows:
0253 runnumber = int(runnumber)
0254 fc_dsttypes_by_run[runnumber].append(dsttype)
0255 total_events_by_run[runnumber] += int(lastevent or 0)
0256
0257 lustre_no_fc = [
0258 (run, daqhost)
0259 for run, daqhost in lustre_combos
0260 if not any(daqhost_to_dst_leaf(daqhost, match) in dsttype for dsttype in fc_dsttypes_by_run.get(run, []))
0261 ]
0262 INFO(f"{len(lustre_no_fc)} lustre combos have no FileCatalog entry.")
0263
0264 all_no_fc = [
0265 (int(r), host)
0266 for r, host in all_combos
0267 if not any(daqhost_to_dst_leaf(host, match) in dsttype for dsttype in fc_dsttypes_by_run.get(int(r), []))
0268 ]
0269 if all_no_fc:
0270 WARN(f"{len(all_no_fc)} raw DB combos (lustre or not) have no FileCatalog entry. Check for corruption?")
0271 for run, daqhost in sorted(all_no_fc)[:args.example_limit]:
0272 DEBUG(f" Run {run} {daqhost}")
0273
0274 INFO(f"Checking for combinations flagged below ratio cut {args.ratio_cut}...")
0275 flagged = []
0276 for runnumber, dsttype, lastevent in rows:
0277 runnumber = int(runnumber)
0278 eventsinrun = eventsinrun_by_run.get(runnumber)
0279 if not eventsinrun:
0280 WARN(f"Run {runnumber} {dsttype}: eventsinrun=0, cannot compute ratio.")
0281 continue
0282
0283 ratio = lastevent / eventsinrun
0284 msg = f"Run {runnumber} {dsttype}: lastevent={lastevent}, eventsinrun={eventsinrun}, ratio={ratio:.3f}"
0285 if ratio < args.ratio_cut:
0286 WARN(msg)
0287 flagged.append((runnumber, dsttype))
0288 elif ratio < 0.999:
0289 CHATTY(msg)
0290
0291 flagged = sorted(set(flagged))
0292 INFO(f"{len(flagged)} (run, dsttype) combinations flagged below ratio cut {args.ratio_cut}.")
0293
0294 possible_daqhost_sets_by_run = defaultdict(set)
0295 possible_daqhosts_by_run = defaultdict(int)
0296 total_daqhosts_by_run = defaultdict(int)
0297 possible_events_by_run = defaultdict(int)
0298 for runnumber, daqhost in all_combos:
0299 runnumber = int(runnumber)
0300 possible_daqhost_sets_by_run[runnumber].add(daqhost)
0301 possible_daqhosts_by_run[runnumber] += 1
0302 possible_events_by_run[runnumber] += int(eventsinrun_by_run.get(runnumber, 0))
0303
0304 missing_daqhost_sets_by_run = defaultdict(set)
0305 for runnumber, daqhost in all_no_fc:
0306 missing_daqhost_sets_by_run[int(runnumber)].add(daqhost)
0307 for runnumber, possible_daqhosts in possible_daqhosts_by_run.items():
0308 total_daqhosts_by_run[runnumber] = max(
0309 possible_daqhosts - len(missing_daqhost_sets_by_run.get(runnumber, set())),
0310 0,
0311 )
0312
0313 possible_segments_by_run = defaultdict(int)
0314 if neventsper:
0315 for runnumber, eventsinrun in eventsinrun_by_run.items():
0316 possible_segments_by_run[int(runnumber)] = math.ceil(int(eventsinrun or 0) / neventsper)
0317
0318 output_events_by_run_host = defaultdict(dict)
0319 for runnumber, dsttype, lastevent in rows:
0320 runnumber = int(runnumber)
0321 for daqhost in possible_daqhost_sets_by_run.get(runnumber, set()):
0322 if daqhost_to_dst_leaf(daqhost, match) in dsttype:
0323 output_events_by_run_host[runnumber][daqhost] = max(
0324 output_events_by_run_host[runnumber].get(daqhost, 0),
0325 int(lastevent or 0),
0326 )
0327
0328 total_segments_by_run = defaultdict(int)
0329 for runnumber, possible_hosts in possible_daqhost_sets_by_run.items():
0330 possible_segments = possible_segments_by_run.get(runnumber, 0)
0331 if not neventsper or not possible_segments or not possible_hosts:
0332 continue
0333 segment_depths = []
0334 for daqhost in possible_hosts:
0335 output_events = output_events_by_run_host.get(runnumber, {}).get(daqhost, 0)
0336 adjusted_events = output_events + EXPECTED_SKIPPED_EVENTS_PER_DAQHOST if output_events else 0
0337 segment_depths.append(min(possible_segments, math.ceil(adjusted_events / neventsper)))
0338 total_segments_by_run[runnumber] = min(segment_depths) if segment_depths else 0
0339
0340 reasons_by_run = defaultdict(set)
0341 for runnumber, _ in flagged:
0342 add_reason(reasons_by_run, runnumber, "low_event_ratio")
0343 for runnumber, _ in not_on_lustre:
0344 add_reason(reasons_by_run, runnumber, "not_on_lustre")
0345 for runnumber in runs_without_gl1daq:
0346 add_reason(reasons_by_run, runnumber, "missing_gl1daq")
0347 for runnumber, possible_events in possible_events_by_run.items():
0348 total_events = total_events_by_run.get(runnumber, 0)
0349 expected_skipped = EXPECTED_SKIPPED_EVENTS_PER_DAQHOST * possible_daqhosts_by_run.get(runnumber, 0)
0350 if possible_events and (total_events + expected_skipped) / possible_events < args.ratio_cut:
0351 add_reason(reasons_by_run, runnumber, "low_run_event_ratio")
0352
0353 expected_skipped_events_by_run = {
0354 runnumber: EXPECTED_SKIPPED_EVENTS_PER_DAQHOST * possible_daqhosts
0355 for runnumber, possible_daqhosts in possible_daqhosts_by_run.items()
0356 }
0357 error_codes_by_run = query_error_codes(args, match, run_condition)
0358
0359 all_report_runs = set(eventsinrun_by_run) | set(possible_events_by_run) | set(total_events_by_run)
0360 write_csv_report(
0361 report_path,
0362 args,
0363 all_report_runs,
0364 possible_daqhosts_by_run,
0365 total_daqhosts_by_run,
0366 possible_segments_by_run,
0367 total_segments_by_run,
0368 possible_events_by_run,
0369 total_events_by_run,
0370 error_codes_by_run,
0371 reasons_by_run,
0372 expected_skipped_events_by_run,
0373 )
0374
0375 files_db_events = sum(total_events_by_run.values())
0376 raw_combo_events = sum(possible_events_by_run.values())
0377 event_pct = 100.0 * files_db_events / raw_combo_events if raw_combo_events else 0.0
0378 INFO(
0379 f"Summary: FileCatalog has {human_event_count(files_db_events)}/"
0380 f"{human_event_count(raw_combo_events)} possible events from raw DB combos "
0381 f"({event_pct:.1f}%)."
0382 )
0383 INFO(f"Available: {raw_combo_events} \t Done {files_db_events}")
0384 return True
0385
0386
0387 def main():
0388 args = submission_args()
0389 args.example_limit = max(0, args.example_limit)
0390
0391 from simpleLogger import slogger, set_log_timestamps_enabled
0392 import logging
0393 set_log_timestamps_enabled(False)
0394 slogger.setLevel(logging.getLevelName(args.loglevel))
0395
0396 profiler = None
0397 if args.profile:
0398 DEBUG("Profiling is ENABLED.")
0399 profiler = cProfile.Profile()
0400 profiler.enable()
0401
0402 rule, match = load_rule_and_match(args)
0403 report_path = output_path(args)
0404
0405 if "raw" not in match.input_config.db:
0406 ERROR(
0407 f"Rule '{args.rulename}' is a downstream rule (db={match.input_config.db}). "
0408 "generate_report.py only supports event-combiner/raw rules for now."
0409 )
0410 sys.exit(2)
0411
0412 wrote_report = generate_eventcombiner_report(args, rule, match, report_path)
0413
0414 if args.report != "none":
0415 WARN("--report is accepted for argument compatibility but ignored by generate_report.py; CSV was written instead.")
0416 if args.delete:
0417 WARN("--delete is accepted for argument compatibility but ignored by generate_report.py.")
0418
0419 if wrote_report:
0420 print(f"Result written to: {report_path}")
0421
0422 if profiler:
0423 profiler.disable()
0424 DEBUG("Profiling finished. Printing stats...")
0425 stats = pstats.Stats(profiler)
0426 stats.strip_dirs().sort_stats("time").print_stats(20)
0427
0428
0429 if __name__ == "__main__":
0430 main()
