feat: Add a CLI command and utility function to combine two qmap HDF5 files into a single output file.

Author: AZjk

pyproject.toml

@@ -45,6 +45,7 @@ dev = [
 
 [project.scripts]
 pysimplemask = "pysimplemask.cli:main"
+pysimplemask-combine-qmaps = "pysimplemask.cli:combine_qmaps"
 
 [project.urls]
 bugs = "https://github.com/AdvancedPhotonSource/pySimpleMask/issues"

src/pysimplemask/cli.py

@@ -1,8 +1,10 @@
 """Console script for pysimplemask."""
 import argparse
+import logging
 import sys
 import os
 from pysimplemask import main_gui, __version__
+from pysimplemask.utils import combine_qmap_files
 
 
 def main():
@@ -15,5 +17,30 @@ def main():
     sys.exit(main_gui(args.path))
 
 
+def combine_qmaps():
+    """CLI entry point: combine two qmap HDF5 files into one."""
+    parser = argparse.ArgumentParser(
+        prog="pysimplemask-combine-qmaps",
+        description="Combine two pySimpleMask qmap HDF5 files into a single output file.",
+    )
+    parser.add_argument("qmap_file1", help="Path to the first qmap HDF5 file.")
+    parser.add_argument("qmap_file2", help="Path to the second qmap HDF5 file.")
+    parser.add_argument("output_file", help="Path for the combined output qmap HDF5 file.")
+    parser.add_argument(
+        "-v", "--verbose",
+        action="store_true",
+        help="Enable verbose (DEBUG-level) logging.",
+    )
+    args = parser.parse_args()
+
+    logging.basicConfig(
+        level=logging.DEBUG if args.verbose else logging.INFO,
+        format="%(asctime)s [%(levelname)s] %(message)s",
+        datefmt="%H:%M:%S",
+    )
+
+    combine_qmap_files(args.qmap_file1, args.qmap_file2, args.output_file)
+
+
 if __name__ == "__main__":
     sys.exit(main())  # pragma: no cover

src/pysimplemask/utils.py

@@ -1,7 +1,13 @@
 import hashlib
+import json
+import logging
+import shutil
+
+import h5py
 import numpy as np
 from typing import Dict, Union
-import json
+
+logger = logging.getLogger(__name__)
 
 
 def hash_numpy_dict(input_dictionary):
@@ -292,3 +298,119 @@ def check_consistency(dqmap: np.ndarray, sqmap: np.ndarray, mask: np.ndarray) ->
             sq_to_dq[sq_value] = dq_value
 
     return True
+
+
+def combine_qmap_files(qmap_file1, qmap_file2, output_file):
+    """
+    Combine two qmap files into a single qmap file.
+
+    Parameters
+    ----------
+    qmap_file1 : str
+        Path to the first qmap file.
+    qmap_file2 : str
+        Path to the second qmap file.
+    output_file : str
+        Path to the output qmap file.
+    """
+    logger.info("Combining qmap files:")
+    logger.info("  file1 : %s", qmap_file1)
+    logger.info("  file2 : %s", qmap_file2)
+    logger.info("  output: %s", output_file)
+
+    with h5py.File(qmap_file1, "r") as f1, h5py.File(qmap_file2, "r") as f2:
+        map_names1 = tuple(f1["/qmap/map_names"][()])
+        map_names2 = tuple(f2["/qmap/map_names"][()])
+        assert map_names1 == map_names2, (
+            f"map_names must be the same: {map_names1!r} != {map_names2!r}"
+        )
+        logger.info("map_names validated: %s", map_names1)
+
+        logger.info("Copying file1 -> output file as base ...")
+        shutil.copy(qmap_file1, output_file)
+
+        with h5py.File(output_file, "r+") as fo:
+            # Combine masks
+            mask1 = f1["/qmap/mask"][()]
+            mask2 = f2["/qmap/mask"][()]
+            combined_mask = np.logical_or(mask1, mask2)
+            logger.info(
+                "Mask: file1 valid=%d  file2 valid=%d  combined valid=%d",
+                mask1.sum(), mask2.sum(), combined_mask.sum(),
+            )
+            del fo["/qmap/mask"]
+            fo["/qmap/mask"] = combined_mask
+
+            for prefix in ["static", "dynamic"]:
+                logger.info("--- Processing '%s' partition ---", prefix)
+
+                f1_num_pts = f1[f"/qmap/{prefix}_num_pts"][()]
+                f2_num_pts = f2[f"/qmap/{prefix}_num_pts"][()]
+                logger.debug(
+                    "  num_pts: file1=%s  file2=%s",
+                    f1_num_pts.tolist(), f2_num_pts.tolist(),
+                )
+
+                dim0_num_pts = f1_num_pts[0] + f2_num_pts[0]
+                dim1_num_pts = max(f1_num_pts[1], f2_num_pts[1])
+                logger.info(
+                    "  Combined num_pts: dim0=%d (file1 %d + file2 %d)  dim1=%d",
+                    dim0_num_pts, f1_num_pts[0], f2_num_pts[0], dim1_num_pts,
+                )
+                del fo[f"/qmap/{prefix}_num_pts"]
+                fo[f"/qmap/{prefix}_num_pts"] = np.array([dim0_num_pts, dim1_num_pts])
+
+                # Combine dim0 value list (concatenate both ranges)
+                v_list_dim0 = np.concatenate(
+                    [
+                        f1[f"/qmap/{prefix}_v_list_dim0"][()],
+                        f2[f"/qmap/{prefix}_v_list_dim0"][()],
+                    ]
+                )
+                logger.debug(
+                    "  v_list_dim0: range [%.6g, %.6g], %d entries",
+                    v_list_dim0.min(), v_list_dim0.max(), len(v_list_dim0),
+                )
+                del fo[f"/qmap/{prefix}_v_list_dim0"]
+                fo[f"/qmap/{prefix}_v_list_dim0"] = v_list_dim0
+
+                # Keep the longer dim1 value list
+                if f1_num_pts[1] < f2_num_pts[1]:
+                    logger.debug(
+                        "  v_list_dim1: using file2's list (%d > %d entries)",
+                        f2_num_pts[1], f1_num_pts[1],
+                    )
+                    del fo[f"/qmap/{prefix}_v_list_dim1"]
+                    fo[f"/qmap/{prefix}_v_list_dim1"] = f2[f"/qmap/{prefix}_v_list_dim1"][()]
+
+                # Merge roi maps: offset file2's non-zero indices so they don't
+                # collide with file1's, then add the two maps together.
+                roi_map1 = f1[f"/qmap/{prefix}_roi_map"][()]
+                roi_map2 = f2[f"/qmap/{prefix}_roi_map"][()].copy()
+                logger.debug(
+                    "  roi_map: file1 max=%d  file2 max=%d",
+                    roi_map1.max(), roi_map2.max(),
+                )
+                roi_map2[roi_map2 > 0] += np.max(roi_map1[roi_map1 > 0])
+                roi_map = roi_map1 + roi_map2
+
+                start_index = np.min(roi_map)
+
+                # Re-index to natural order (0 = masked, 1-based = valid)
+                unique_idx, inverse = np.unique(roi_map, return_inverse=True)
+                partition_natural_order = inverse.reshape(roi_map.shape).astype(np.uint32)
+
+                # If no masked pixels exist, shift up so index 0 stays reserved
+                if start_index > 0:
+                    partition_natural_order += 1
+
+                num_partitions = int((unique_idx > 0).sum())
+                logger.info("  Combined roi_map: %d valid partitions", num_partitions)
+
+                del fo[f"/qmap/{prefix}_roi_map"]
+                fo[f"/qmap/{prefix}_roi_map"] = partition_natural_order
+
+                del fo[f"/qmap/{prefix}_index_mapping"]
+                fo[f"/qmap/{prefix}_index_mapping"] = unique_idx[unique_idx > 0] - 1
+
+    logger.info("Done. Output written to: %s", output_file)