prepare merlict blocks

plenoirf/production/__init__.py

@@ -33,6 +33,7 @@
 from . import corsika_and_grid
 from . import split_event_tape_into_blocks
 from . import inspect_particle_pool
+from . import simulate_hardware
 
 
 def make_example_job(
@@ -104,6 +105,13 @@ def run_job_in_dir(job, tmp_dir):
     return job
 
 
+def _run_job_block(job, block_id):
+    with jll.TimeDelta(
+        logger, "simulate_hardware_block{:06d}".format(block_id)
+    ):
+        job = simulate_hardware.run_job(job=job, logger=logger)
+
+
 def compile_job_paths_and_unique_identity(job, tmp_dir):
     """
     Adds static information to the job dict.
@@ -178,7 +186,7 @@ def compile_job_paths(job, tmp_dir):
 
     paths["tmp"]["cherenkov_pools"] = opj(tmp_dir, "cherenkov_pools.tar")
     paths["tmp"]["cherenkov_pools_block_fmt"] = opj(
-        tmp_dir, "cherenkov_pools_{block:06d}.tar"
+        tmp_dir, "cherenkov_pools_block{block_id:06d}.tar"
     )
     paths["tmp"]["particle_pools_dat"] = opj(tmp_dir, "particle_pools.dat")
     paths["tmp"]["particle_pools_tar"] = opj(tmp_dir, "particle_pools.tar.gz")
@@ -190,8 +198,15 @@ def compile_job_paths(job, tmp_dir):
 
     paths["tmp"]["corsika_stdout"] = opj(tmp_dir, "corsika.stdout")
     paths["tmp"]["corsika_stderr"] = opj(tmp_dir, "corsika.stderr")
-    paths["tmp"]["merlict_stdout"] = opj(tmp_dir, "merlict.stdout")
-    paths["tmp"]["merlict_stderr"] = opj(tmp_dir, "merlict.stderr")
+    paths["tmp"]["merlict_stdout_block_fmt"] = opj(
+        tmp_dir, "merlict_block{block_id:06d}.stdout"
+    )
+    paths["tmp"]["merlict_stderr_block_fmt"] = opj(
+        tmp_dir, "merlict_block{block_id:06d}.stderr"
+    )
+    paths["tmp"]["merlict_output_block_fmt"] = opj(
+        tmp_dir, "merlict_block{block_id:06d}"
+    )
 
     # debug output
     # ------------

plenoirf/production/draw_event_ids_for_debug_output.py

@@ -1,20 +1,28 @@
 from .. import debugging
+from .. import bookkeeping
 
 
 def run_job(job, logger):
-    job["run"][
-        "event_ids_for_debug"
-    ] = debugging.draw_event_ids_for_debug_output(
+    event_ids_for_debugging = debugging.draw_event_ids_for_debug_output(
         num_events_in_run=job["num_events"],
         min_num_events=job["config"]["debug_output"]["run"]["min_num_events"],
         fraction_of_events=job["config"]["debug_output"]["run"][
             "fraction_of_events"
         ],
         prng=job["prng"],
     )
+
+    job["run"]["uids_for_debugging"] = []
+    for event_id in event_ids_for_debugging:
+        uid = bookkeeping.uid.make_uid(run_id=job["run_id"], event_id=event_id)
+        job["run"]["uids_for_debugging"].append(uid)
+    job["run"]["uids_for_debugging"] = np.array(
+        job["run"]["uids_for_debugging"]
+    )
+
     logger.info(
-        "event-ids for debugging: {:s}.".format(
-            str(job["run"]["event_ids_for_debug"].tolist())
+        "event uids for debugging: {:s}.".format(
+            str(job["run"]["uids_for_debugging"].tolist())
         )
     )
     return job

plenoirf/production/simulate_hardware.py

@@ -0,0 +1,69 @@
+import os
+from os import path as op
+from os.path import join as opj
+
+import merlict_development_kit_python
+from . import job_io
+
+
+def run_job_block(job, block_id, logger):
+    cache_path = opj(
+        job["paths"]["tmp_dir"],
+        "simulate_hardware_block{block_id:06d}".format(block_id=block_id),
+    )
+
+    if os.path.exists(cache_path) and job["cache"]:
+        logger.info(
+            "simulate_hardware block{:06d}, read cache".format(block_id)
+        )
+        return job_io.read(path=cache_path)
+    else:
+        job = simulate_hardware(job=job, block_id=block_id)
+        make_debug_output(job=job, block_id=block_id)
+
+        if job["cache"]:
+            logger.info(
+                "simulate_hardware block{:06d}, write cache".format(block_id)
+            )
+            job_io.write(path=cache_path, job=job)
+
+    return job
+
+
+def simulate_hardware(job, block_id):
+    detector_responses_path = op.join(tmp_dir, "detector_responses")
+
+    rc = merlict_development_kit_python.plenoscope_propagator.plenoscope_propagator(
+        corsika_run_path=job["paths"][
+            "cherenkov_pools_block_fmt".format(block_id=block_id)
+        ],
+        output_path=job["paths"][
+            "merlict_output_block_fmt".format(block_id=block_id)
+        ],
+        light_field_geometry_path=job["paths"]["light_field_calibration"],
+        merlict_plenoscope_propagator_path=job["config"]["executables"][
+            "merlict_plenoscope_propagator_path"
+        ],
+        merlict_plenoscope_propagator_config_path=job[
+            "merlict_plenoscope_propagator_config_path"
+        ],
+        random_seed=job["run_id"],
+        photon_origins=True,
+        stdout_path=job["paths"][
+            "merlict_stdout_block_fmt".format(block_id=block_id)
+        ],
+        stderr_path=job["paths"][
+            "merlict_stderr_block_fmt".format(block_id=block_id)
+        ],
+    )
+    assert rc == 0, "Expected merlict's return code to be zero."
+
+    return job
+
+
+def make_debug_output(job, block_id):
+    uids_in_block = job["run"]["uids_in_cherenkov_pool_blocks"][str(block_id)]
+    for event_id in job["run"]["event_ids_for_debug"]:
+        uid = bookkeeping.uid.make_uid(run_id=job["run_id"], event_id=event_id)
+        if uid in uids_in_block:
+            print("Do some debug I guess?", uid, block_id)

plenoirf/production/split_event_tape_into_blocks.py

@@ -5,11 +5,12 @@
 
 import corsika_primary as cpw
 from .corsika_and_grid import read_all_cherenkov_bunches
+from .. import bookkeeping
 
 
 def run_job(job, logger):
     logger.info("split_event_tape_into_blocks, split")
-    job["run"]["cherenkov_pools"] = split_event_tape_into_blocks(
+    job["run"]["uids_in_cherenkov_pool_blocks"] = split_event_tape_into_blocks(
         inpath=job["paths"]["tmp"]["cherenkov_pools"],
         outpath_block_fmt=job["paths"]["tmp"]["cherenkov_pools_block_fmt"],
         num_events=job["max_num_events_in_merlict_run"],
@@ -21,31 +22,40 @@ def run_job(job, logger):
 def split_event_tape_into_blocks(inpath, outpath_block_fmt, num_events):
     Writer = cpw.cherenkov.CherenkovEventTapeWriter
     Reader = cpw.cherenkov.CherenkovEventTapeReader
-    outpaths = {}
+    block_ids = []
+
+    uid_map = {}
 
     orun = None
-    block = 0
+    block_id = 0
     event_counter = 0
     with Reader(inpath) as irun:
         runh = copy.deepcopy(irun.runh)
 
         for event in irun:
             evth, cherenkov_reader = event
             cherenkov_bunches = read_all_cherenkov_bunches(cherenkov_reader)
+            uid = bookkeeping.uid.make_uid(
+                run_id=int(evth[cpw.I.EVTH.RUN_NUMBER]),
+                event_id=int(evth[cpw.I.EVTH.EVENT_NUMBER]),
+            )
 
             if event_counter % num_events == 0:
-                block += 1
+                block_id += 1
+                block_id_str = "{:06d}".format(block_id)
                 if orun is not None:
                     orun.close()
-                outpaths[block] = opj(outpath_block_fmt.format(block=block))
-                orun = Writer(outpaths[block])
+                outpath = opj(outpath_block_fmt.format(block_id=block_id))
+                orun = Writer(outpath)
                 orun.write_runh(runh)
+                uid_map[block_id_str] = []
 
+            uid_map[block_id_str].append(uid)
             orun.write_evth(evth=evth)
             orun.write_payload(payload=cherenkov_bunches)
             event_counter += 1
 
         if orun is not None:
             orun.close()
 
-    return outpaths
+    return uid_map