Added document logging & unit tests

doc/docs_logging_callback.md

@@ -0,0 +1,5 @@
+# Document Logging Callback
+
+The document logger is a callback that the BlueSky RunEngine subscribes to unconditionally. After receiving each document, if they share the same start document (in the same run) then it will write them to the same file. These logs are stored under `C:/instrument/var/logs/bluesky/raw_documents` and are handled by the log rotation.
+
+Each document is stored in a JSON format so can be both machine and human readable. It is in the format `{"type": name, "document": document}` whereby `name` is the type of the document, e.g start, stop, event, descriptor and the `document` is the [document from BlueSky in JSON format](https://blueskyproject.io/bluesky/main/documents.html). As these files are produced per BlueSky run, these will be useful for debugging.

src/ibex_bluesky_core/callbacks/__init__.py

@@ -0,0 +1 @@
+"""For callbacks that relate to the BlueSky run engine."""

src/ibex_bluesky_core/callbacks/document_logger.py

@@ -0,0 +1,37 @@
+"""Logs all documents that the BlueSky run engine creates via a callback."""
+
+import json
+from pathlib import Path
+
+log_location = Path("C:\\") / "instrument" / "var" / "logs" / "bluesky" / "raw_documents"
+
+
+class DocLoggingCallback:
+    """Logs all documents under log_location, with the file name of their UID (.log)."""
+
+    def __init__(self) -> None:
+        """Initialise current_start_document and filename."""
+        self.current_start_document = None
+        self.filename = None
+
+    def __call__(self, name: str, document: dict) -> None:
+        """Is called when a new document needs to be processed. Writes document to a file.
+
+        Args:
+            name: The type of the document e.g start, event, stop
+            document: The contents of the docuement as a dictionary
+
+        """
+        if name == "start":
+            log_location.mkdir(parents=True, exist_ok=True)
+
+            self.current_start_document = document["uid"]
+            self.filename = log_location / f"{self.current_start_document}.log"
+
+        assert self.filename is not None, "Could not create filename."
+        assert self.current_start_document is not None, "Saw a non-start document before a start."
+
+        to_write = {"type": name, "document": document}
+
+        with open(self.filename, "a") as outfile:
+            outfile.write(f"{json.dumps(to_write)}\n")

src/ibex_bluesky_core/run_engine.py

@@ -7,6 +7,8 @@
 from bluesky.run_engine import RunEngine
 from bluesky.utils import DuringTask
 
+from ibex_bluesky_core.callbacks.document_logger import DocLoggingCallback
+
 __all__ = ["get_run_engine"]
 
 
@@ -65,4 +67,7 @@ def get_run_engine() -> RunEngine:
     )
     RE.record_interruptions = True
 
+    log_callback = DocLoggingCallback()
+    RE.subscribe(log_callback)
+
     return RE

tests/conftest.py

@@ -0,0 +1,9 @@
+import pytest
+from bluesky.run_engine import RunEngine
+from ibex_bluesky_core.run_engine import get_run_engine
+
+
+@pytest.fixture
+def RE() -> RunEngine:
+    get_run_engine.cache_clear()
+    return get_run_engine()

tests/test_document_logging_callback.py

@@ -0,0 +1,32 @@
+import json
+from pathlib import Path
+from typing import Generator
+from unittest.mock import mock_open, patch
+
+import bluesky.plan_stubs as bps
+from bluesky.run_engine import RunEngineResult
+from bluesky.utils import Msg
+
+
+def test_run_engine_logs_all_documents(RE):
+    m = mock_open()
+    log_location = Path("C:\\") / "instrument" / "var" / "logs" / "bluesky" / "raw_documents"
+
+    def basic_plan() -> Generator[Msg, None, None]:
+        yield from bps.open_run()
+        yield from bps.close_run()
+
+    with patch("ibex_bluesky_core.callbacks.document_logger.open", m):
+        result: RunEngineResult = RE(basic_plan())
+        filepath = log_location / f"{result.run_start_uids[0]}.log"
+
+    for i in range(0, 3):
+        assert m.call_args_list[i].args == (filepath, "a")
+        # Checks that the file is opened 3 times, for open, descriptor then stop
+
+    handle = m()
+    document = json.loads(handle.write.mock_calls[-1].args[0])
+
+    assert document["document"]["exit_status"] == "success"
+    # In the stop document to be written, check that the run is successful with no interruptions
+    assert document["document"]["num_events"]["interruptions"] == 0

tests/test_run_engine.py

@@ -4,17 +4,11 @@
 
 import bluesky.plan_stubs as bps
 import pytest
-from bluesky.run_engine import RunEngine, RunEngineResult
+from bluesky.run_engine import RunEngineResult
 from bluesky.utils import Msg, RequestAbort, RequestStop, RunEngineInterrupted
 from ibex_bluesky_core.run_engine import _DuringTask, get_run_engine
 
 
-@pytest.fixture
-def RE() -> RunEngine:
-    get_run_engine.cache_clear()
-    return get_run_engine()
-
-
 def test_run_engine_is_singleton():
     re1 = get_run_engine()
     re2 = get_run_engine()