Support dynamic enable/disable of VENUS image cataloging

README.md

@@ -99,6 +99,25 @@ If not specified, this defaults to the VENUS instrument metadata path. For other
 this parameter to match the appropriate metadata path(s) in your NeXus files. Multiple paths can be specified
 as an array.
 
+###### Enabling and disabling image cataloging
+
+Image cataloging can be dynamically enabled or disabled per instrument, without a code change,
+configuration edit, or service restart. This mirrors the `reduce_<INSTRUMENT>.py` convention used to
+toggle autoreduction: the agent catalogs image files for a run **only if** the script
+
+    /<facility>/<instrument>/shared/autoreduce/catalog_<INSTRUMENT>.py
+
+exists (for VENUS: `/SNS/VENUS/shared/autoreduce/catalog_VENUS.py`). The agent checks for the file's
+presence only; its contents are not executed. See `configuration/catalog_VENUS.py.example` for a
+ready-to-deploy placeholder.
+
+  - **To enable** image cataloging: place the script at the path above.
+  - **To disable** it (e.g. to relieve a cataloging backlog): move or rename the script, for example
+    `mv catalog_VENUS.py catalog_VENUS.py.disabled`.
+
+Disabling image cataloging affects only that image substep. Cataloging of the raw data file and its
+related files continues normally.
+
 
 Installation
 ------------

README_developer.md

@@ -134,6 +134,14 @@ Example message with custom parameter values:
 
 Queue name: `CATALOG.ONCAT.DATA_READY`.
 
+Cataloging ingests the raw data file and any related files. For instruments that produce image files
+(currently VENUS), an additional image-cataloging substep batch-ingests the associated FITS/TIFF files.
+This substep is gated per instrument: `ONCatProcessor.catalog_images()` runs only when the script
+`/<facility>/<instrument>/shared/autoreduce/catalog_<INSTRUMENT>.py` exists, mirroring the
+`reduce_<INSTRUMENT>.py` toggle used for autoreduction. The `dev_instrument_shared` configuration
+parameter overrides the shared directory for local development and testing. See the README for the
+operator-facing enable/disable instructions.
+
 #### Related configuration parameters
 
 | Configuration parameter | Description | Default value |

configuration/catalog_VENUS.py.example

@@ -0,0 +1,15 @@
+# Image cataloging toggle for VENUS
+#
+# The post-processing agent catalogs image files (a VENUS-specific substep of
+# ONCat cataloging) ONLY when this file is present at:
+#
+#     /SNS/VENUS/shared/autoreduce/catalog_VENUS.py
+#
+# This mirrors the reduce_<INSTRUMENT>.py convention used to toggle autoreduction.
+# The agent checks for the file's PRESENCE only; its contents are not executed.
+#
+# To ENABLE image cataloging:  place this file at the path above.
+# To DISABLE image cataloging: move or rename the file (e.g. catalog_VENUS.py.disabled).
+#     Raw and reduced data cataloging are unaffected either way.
+#
+# No code change or service restart is required; the check happens per job.

postprocessing/processors/oncat_processor.py

@@ -65,7 +65,36 @@ def ingest(self, location):
             logging.info("Calling ONCat for %s", related_file)
             oncat.Datafile.ingest(related_file)
 
-        # Catalog image files using batch API for efficiency
+        # Catalog image files (a VENUS-specific substep), if enabled for this instrument
+        self.catalog_images(oncat, datafile)
+
+    def catalog_images(self, oncat, datafile):
+        """Catalog image files using the batch API, if enabled for this instrument.
+
+        Image cataloging is a special substep currently used only by VENUS. It can be
+        dynamically enabled or disabled per instrument by adding or removing the script
+        ``catalog_<INSTRUMENT>.py`` in the instrument's shared autoreduce directory. This
+        mirrors the ``reduce_<INSTRUMENT>.py`` convention used to toggle autoreduction, so
+        an instrument scientist can turn image cataloging off (e.g. to relieve a backlog)
+        by moving that file, with no code change or service restart.
+
+        @param oncat: an authenticated pyoncat.ONCat client
+        @param datafile: the ONCat datafile object returned by ingesting the main file
+        """
+        instrument_shared_dir = os.path.join("/", self.facility, self.instrument, "shared", "autoreduce")
+        if len(self.configuration.dev_instrument_shared) > 0:
+            instrument_shared_dir = self.configuration.dev_instrument_shared
+
+        catalog_script = os.path.join(instrument_shared_dir, f"catalog_{self.instrument}.py")
+        if not os.path.isfile(catalog_script):
+            logging.info(
+                "Image cataloging disabled for %s (no %s)",
+                self.instrument,
+                catalog_script,
+            )
+            return
+
+        logging.info("Image cataloging enabled for %s (found %s)", self.instrument, catalog_script)
         images = image_files(datafile, self.configuration.image_filepath_metadata_paths)
         for batch in batches(images, IMAGE_BATCH_SIZE):
             logging.info("Batch ingesting %d image files", len(batch))

tests/integration/Dockerfile

@@ -50,6 +50,15 @@ RUN mkdir -p /opt/postprocessing/log && \
     touch /SNS/VENUS/IPTS-99999/images/image_001.fits && \
     touch /SNS/VENUS/IPTS-99999/images/image_002.fits && \
     touch /SNS/VENUS/IPTS-99999/images/image_003.tiff && \
+    mkdir -p /SNS/VENUS/shared/autoreduce && \
+    echo "# image cataloging enabled for VENUS" > /SNS/VENUS/shared/autoreduce/catalog_VENUS.py && \
+    \
+    mkdir -p /SNS/IMAGING/IPTS-99999/nexus && \
+    mkdir -p /SNS/IMAGING/IPTS-99999/images && \
+    mkdir -p /SNS/IMAGING/shared/autoreduce && \
+    touch /SNS/IMAGING/IPTS-99999/nexus/IMAGING_12346.nxs.h5 && \
+    touch /SNS/IMAGING/IPTS-99999/images/image_001.fits && \
+    touch /SNS/IMAGING/IPTS-99999/images/image_002.fits && \
     \
     chown -R root:root /opt/postprocessing /SNS
 

tests/integration/Dockerfile.oncat

@@ -13,5 +13,11 @@ RUN mkdir -p /SNS/CORELLI/IPTS-15526/nexus && \
     touch /SNS/VENUS/IPTS-99999/nexus/VENUS_12345.nxs.h5 && \
     touch /SNS/VENUS/IPTS-99999/images/image_001.fits && \
     touch /SNS/VENUS/IPTS-99999/images/image_002.fits && \
-    touch /SNS/VENUS/IPTS-99999/images/image_003.tiff
+    touch /SNS/VENUS/IPTS-99999/images/image_003.tiff && \
+    \
+    mkdir -p /SNS/IMAGING/IPTS-99999/nexus && \
+    mkdir -p /SNS/IMAGING/IPTS-99999/images && \
+    touch /SNS/IMAGING/IPTS-99999/nexus/IMAGING_12346.nxs.h5 && \
+    touch /SNS/IMAGING/IPTS-99999/images/image_001.fits && \
+    touch /SNS/IMAGING/IPTS-99999/images/image_002.fits
 CMD ["python", "oncat_server.py"]

tests/integration/oncat_server.py

@@ -102,8 +102,10 @@ def do_POST(self):
             "indexed": {"run_number": run_number},
         }
 
-        # Add metadata for VENUS instrument to support image cataloging
-        if instrument == "VENUS":
+        # Add metadata for imaging instruments to support image cataloging.
+        # IMAGING is a test-only instrument used to exercise the disabled path
+        # (it has image metadata and image files, but no catalog_IMAGING.py script).
+        if instrument in ("VENUS", "IMAGING"):
             response["metadata"] = {"entry": {"daslogs": {"bl10:exp:im:imagefilepath": {"value": "images"}}}}
 
         self.wfile.write(json.dumps(response).encode("utf-8"))

tests/integration/test_cataloging.py

@@ -170,6 +170,65 @@ def test_oncat_catalog_venus_images():
     assert any("INFO   - /SNS/VENUS/IPTS-99999/images/image_003.tiff" in line for line in log)
 
 
+def test_oncat_catalog_images_disabled():
+    """Image cataloging is skipped when the instrument's catalog_<INSTRUMENT>.py is
+    absent, even though image files and image metadata exist. The main file is still
+    cataloged (COMPLETE), but no image batch request is made for this run.
+
+    IMAGING is a test-only imaging instrument with no catalog_IMAGING.py script.
+    """
+    message = {
+        "run_number": "12346",
+        "instrument": "IMAGING",
+        "ipts": "IPTS-99999",
+        "facility": "SNS",
+        "data_file": "/SNS/IMAGING/IPTS-99999/nexus/IMAGING_12346.nxs.h5",
+    }
+
+    conn = stomp.Connection(host_and_ports=[("localhost", 61613)])
+
+    listener = stomp.listener.TestListener(10)  # 10 second timeout
+    conn.set_listener("", listener)
+
+    try:
+        conn.connect("icat", "icat")
+    except stomp.exception.ConnectFailedException:
+        pytest.skip("Requires activemq running")
+
+    # expect a message on CATALOG.ONCAT.COMPLETE
+    conn.subscribe("/queue/CATALOG.ONCAT.COMPLETE", id="imaging123", ack="auto")
+
+    # send data ready
+    conn.send("/queue/CATALOG.ONCAT.DATA_READY", json.dumps(message).encode())
+
+    # Wait for messages until we get the one for this run
+    max_attempts = 10
+    for _ in range(max_attempts):
+        listener.wait_for_message()
+        header, body = listener.get_latest_message()
+        msg = json.loads(body)
+        if msg["run_number"] == message["run_number"]:
+            break
+    else:
+        pytest.fail(f"Did not receive COMPLETE message for IMAGING run {message['run_number']}")
+
+    conn.disconnect()
+
+    assert msg["run_number"] == message["run_number"]
+
+    time.sleep(1)  # give oncat_server time to write its log
+    log = docker_exec_and_cat("/oncat_server.log", "oncat").splitlines()
+
+    # The main NeXus file is still cataloged
+    assert any(
+        "INFO Received datafile ingest request for /SNS/IMAGING/IPTS-99999/nexus/IMAGING_12346.nxs.h5" in line
+        for line in log
+    )
+
+    # But no IMAGING image file is ever batch-ingested (image cataloging was disabled)
+    assert not any("/SNS/IMAGING/IPTS-99999/images/" in line for line in log)
+
+
 def test_oncat_reduction_catalog():
     """This should run reduction ONCatProcessor"""
     message = {

tests/unit/postprocessing/processors/test_oncat_processor.py

@@ -175,6 +175,7 @@ def test_oncat_processor_ingest_with_images():
     mock_conf.oncat_url = "http://oncat:8000"
     mock_conf.oncat_api_token = "test-token"
     mock_conf.image_filepath_metadata_paths = ["metadata.entry.daslogs.bl10:exp:im:imagefilepath.value"]
+    mock_conf.dev_instrument_shared = ""
 
     mock_send_function = Mock()
 
@@ -183,6 +184,8 @@ def test_oncat_processor_ingest_with_images():
         patch("postprocessing.processors.oncat_processor.pyoncat.ONCat") as mock_oncat_class,
         patch("postprocessing.processors.oncat_processor.related_files") as mock_related,
         patch("postprocessing.processors.oncat_processor.image_files") as mock_images,
+        # Image cataloging is enabled for VENUS: the catalog_VENUS.py script is present
+        patch("postprocessing.processors.oncat_processor.os.path.isfile", return_value=True),
     ):
         # Setup mocks
         mock_oncat = Mock()
@@ -235,6 +238,7 @@ def test_oncat_processor_ingest_with_many_images():
     mock_conf.oncat_url = "http://oncat:8000"
     mock_conf.oncat_api_token = "test-token"
     mock_conf.image_filepath_metadata_paths = ["metadata.entry.daslogs.bl10:exp:im:imagefilepath.value"]
+    mock_conf.dev_instrument_shared = ""
 
     mock_send_function = Mock()
 
@@ -246,6 +250,8 @@ def test_oncat_processor_ingest_with_many_images():
         patch("postprocessing.processors.oncat_processor.pyoncat.ONCat") as mock_oncat_class,
         patch("postprocessing.processors.oncat_processor.related_files") as mock_related,
         patch("postprocessing.processors.oncat_processor.image_files") as mock_images,
+        # Image cataloging is enabled for VENUS: the catalog_VENUS.py script is present
+        patch("postprocessing.processors.oncat_processor.os.path.isfile", return_value=True),
     ):
         # Setup mocks
         mock_oncat = Mock()
@@ -269,3 +275,94 @@ def test_oncat_processor_ingest_with_many_images():
         assert len(calls[0][0][0]) == 50
         assert len(calls[1][0][0]) == 50
         assert len(calls[2][0][0]) == 25
+
+
+def test_oncat_processor_image_cataloging_disabled():
+    """When catalog_<INSTRUMENT>.py is absent, image cataloging is skipped but the
+    main file and related files are still cataloged."""
+    test_message = {
+        "run_number": "12345",
+        "instrument": "VENUS",
+        "ipts": "IPTS-99999",
+        "facility": "SNS",
+        "data_file": "/SNS/VENUS/IPTS-99999/nexus/VENUS_12345.nxs.h5",
+    }
+
+    mock_conf = Mock()
+    mock_conf.oncat_url = "http://oncat:8000"
+    mock_conf.oncat_api_token = "test-token"
+    mock_conf.image_filepath_metadata_paths = ["metadata.entry.daslogs.bl10:exp:im:imagefilepath.value"]
+    mock_conf.dev_instrument_shared = ""
+
+    mock_send_function = Mock()
+
+    with (
+        patch("postprocessing.processors.base_processor.open", create=True),
+        patch("postprocessing.processors.oncat_processor.pyoncat.ONCat") as mock_oncat_class,
+        patch("postprocessing.processors.oncat_processor.related_files") as mock_related,
+        patch("postprocessing.processors.oncat_processor.image_files") as mock_images,
+        # Image cataloging is disabled: the catalog_VENUS.py script is absent
+        patch("postprocessing.processors.oncat_processor.os.path.isfile", return_value=False),
+    ):
+        mock_oncat = Mock()
+        mock_oncat_class.return_value = mock_oncat
+        mock_oncat.Datafile.ingest.return_value = Mock()
+        mock_related.return_value = ["/SNS/VENUS/IPTS-99999/images/det_1/VENUS_12345_det_1.tiff"]
+
+        processor = ONCatProcessor(test_message, mock_conf, mock_send_function)
+        processor.ingest(test_message["data_file"])
+
+        # Main file (1) + related file (1) are still ingested
+        assert mock_oncat.Datafile.ingest.call_count == 2
+        # But the image batch API is never called, and we don't even scan for images
+        mock_oncat.Datafile.batch.assert_not_called()
+        mock_images.assert_not_called()
+
+
+def test_catalog_images_uses_instrument_shared_path():
+    """The gate looks for catalog_<INSTRUMENT>.py under the instrument shared dir."""
+    test_message = {
+        "run_number": "12345",
+        "instrument": "VENUS",
+        "ipts": "IPTS-99999",
+        "facility": "SNS",
+        "data_file": "/SNS/VENUS/IPTS-99999/nexus/VENUS_12345.nxs.h5",
+    }
+
+    mock_conf = Mock()
+    mock_conf.image_filepath_metadata_paths = []
+    mock_conf.dev_instrument_shared = ""
+
+    with (
+        patch("postprocessing.processors.base_processor.open", create=True),
+        patch("postprocessing.processors.oncat_processor.os.path.isfile", return_value=False) as mock_isfile,
+    ):
+        processor = ONCatProcessor(test_message, mock_conf, Mock())
+        processor.catalog_images(Mock(), Mock())
+
+        mock_isfile.assert_called_once_with("/SNS/VENUS/shared/autoreduce/catalog_VENUS.py")
+
+
+def test_catalog_images_honors_dev_instrument_shared():
+    """A configured dev_instrument_shared overrides the standard shared path,
+    so the toggle can be exercised locally and in integration tests."""
+    test_message = {
+        "run_number": "12345",
+        "instrument": "VENUS",
+        "ipts": "IPTS-99999",
+        "facility": "SNS",
+        "data_file": "/SNS/VENUS/IPTS-99999/nexus/VENUS_12345.nxs.h5",
+    }
+
+    mock_conf = Mock()
+    mock_conf.image_filepath_metadata_paths = []
+    mock_conf.dev_instrument_shared = "/tmp/dev_shared"
+
+    with (
+        patch("postprocessing.processors.base_processor.open", create=True),
+        patch("postprocessing.processors.oncat_processor.os.path.isfile", return_value=False) as mock_isfile,
+    ):
+        processor = ONCatProcessor(test_message, mock_conf, Mock())
+        processor.catalog_images(Mock(), Mock())
+
+        mock_isfile.assert_called_once_with("/tmp/dev_shared/catalog_VENUS.py")