swf-monitor/scripts/evgen_job_dispatcher.py

0001 #!/usr/bin/env python3
0002 """
0003 In-job per-row dispatcher for the client-API EVGEN production path.
0004
0005 This runs *inside* the PanDA job (in the eic_xl container), shipped in the
0006 submission sandbox by the submit-evgen-task doer. PanDA expands the task's
0007 ``%RNDM`` into a per-job ``${SEQNUMBER}`` (1-based); this reads the matching
0008 row of the one sandbox'd CSV and hands it to the ePIC production payload.
0009
0010 It is our owned equivalent of job_submission_condor/scripts/submit_panda.py
0011 (spec only — never run from that repo); the CSV row shape
0012 (``file,ext,nevents,ichunk``) and the payload entry point
0013 (``/opt/campaigns/hepmc3/scripts/run.sh``) are the fixed contract with the
0014 payload. The payload's run.sh sources ``environment*.sh`` from the unpacked
0015 sandbox itself, so no env is set here.
0016
0017 Usage (as PanDA expands it):  evgen_job_dispatcher.py <SEQNUMBER> <csv_base>
0018 """
0019 import csv
0020 import subprocess
0021 import sys
0022 from itertools import islice
0023
0024 # Payload entry point inside the eic_xl container (campaigns checkout at
0025 # /opt/campaigns/hepmc3). Fixed by the container layout.
0026 PAYLOAD_RUN = "/opt/campaigns/hepmc3/scripts/run.sh"
0027
0028
0029 def main():
0030     if len(sys.argv) < 3:
0031         print(f"Usage: {sys.argv[0]} <SEQNUMBER> <csv_base>", file=sys.stderr)
0032         return 2
0033     n = int(sys.argv[1])
0034     csv_base = sys.argv[2]
0035
0036     # PanDA SEQNUMBER is 1-based; the CSV is 0-indexed.
0037     csv_index = n - 1
0038
0039     # Read only the requested row — no need to load the whole manifest.
0040     with open(f"{csv_base}.csv") as f:
0041         reader = csv.reader(f)
0042         row = next(islice(reader, csv_index, csv_index + 1), None)
0043     if row is None:
0044         print(f"Error: row {n} not found in {csv_base}.csv", file=sys.stderr)
0045         return 1
0046     if len(row) < 4:
0047         print(f"Error: malformed CSV row {n}: {row!r}", file=sys.stderr)
0048         return 1
0049
0050     file_path, ext, nevents, ichunk = row[0], row[1], row[2], row[3]
0051     # The payload prepends the JLab xrootd path to EVGEN/<file>; pass the
0052     # EVGEN-relative path, extension, event count and chunk index through.
0053     result = subprocess.run(
0054         [PAYLOAD_RUN, f"EVGEN/{file_path}", ext, nevents, ichunk],
0055         text=True,
0056     )
0057     return result.returncode
0058
0059
0060 if __name__ == "__main__":
0061     sys.exit(main())