pull out some constants into _utils.py

Author: rerpha

src/ibex_bluesky_core/callbacks/_utils.py

@@ -0,0 +1,20 @@
+from pathlib import Path
+from platform import node
+
+INSTRUMENT = node()
+DEFAULT_PATH = Path("//isis.cclrc.ac.uk/inst$") / INSTRUMENT / "user" / "TEST" / "scans"
+
+# Common document metadata
+UID = "uid"
+TIME = "time"
+DATA = "data"
+RB = "rb_number"
+START_TIME = "start_time"
+NAME = "name"
+SEQ_NUM = "seq_num"
+DATA_KEYS = "data_keys"
+DESCRIPTOR = "descriptor"
+UNITS = "units"
+PRECISION = "precision"
+MOTORS = "motors"
+UNKNOWN_RB = "Unknown RB"

src/ibex_bluesky_core/callbacks/file_logger.py

@@ -5,7 +5,6 @@
 import os
 from datetime import datetime
 from pathlib import Path
-from platform import node
 from typing import Optional
 from zoneinfo import ZoneInfo
 
@@ -15,21 +14,25 @@
 from event_model.documents.run_start import RunStart
 from event_model.documents.run_stop import RunStop
 
-logger = logging.getLogger(__name__)
+from ibex_bluesky_core.callbacks._utils import (
+    DATA,
+    DATA_KEYS,
+    DEFAULT_PATH,
+    DESCRIPTOR,
+    INSTRUMENT,
+    MOTORS,
+    NAME,
+    PRECISION,
+    RB,
+    SEQ_NUM,
+    START_TIME,
+    TIME,
+    UID,
+    UNITS,
+    UNKNOWN_RB,
+)
 
-TIME = "time"
-START_TIME = "start_time"
-NAME = "name"
-SEQ_NUM = "seq_num"
-DATA_KEYS = "data_keys"
-DATA = "data"
-DESCRIPTOR = "descriptor"
-UNITS = "units"
-UID = "uid"
-RB = "rb_number"
-PRECISION = "precision"
-INSTRUMENT = node()
-DEFAULT_PATH = Path("//isis.cclrc.ac.uk/inst$") / INSTRUMENT / "user" / "TEST" / "scans"
+logger = logging.getLogger(__name__)
 
 
 class HumanReadableFileCallback(CallbackBase):
@@ -57,20 +60,20 @@ def start(self, doc: RunStart) -> None:
         self.current_start_document = doc[UID]
 
         datetime_obj = datetime.fromtimestamp(doc[TIME])
-        title_format_datetime = datetime_obj.astimezone(ZoneInfo("Europe/London")).strftime(
+        title_format_datetime = datetime_obj.astimezone(ZoneInfo("UTC")).strftime(
             "%Y-%m-%d_%H-%M-%S"
         )
-        rb_num = doc.get("rb_number", "Unknown RB")
+        rb_num = doc.get(RB, UNKNOWN_RB)
 
         # motors is a tuple, we need to convert to a list to join the two below
-        motors = list(doc.get("motors", []))
+        motors = list(doc.get(MOTORS, []))
 
         self.filename = (
             self.output_dir
             / f"{rb_num}"
             / f"{INSTRUMENT}{'_' + '_'.join(motors) if motors else ''}_{title_format_datetime}Z.txt"
         )
-        if rb_num == "Unknown RB":
+        if rb_num == UNKNOWN_RB:
             logger.warning('No RB number found, saving to "Unknown RB"')
         assert self.filename is not None
         logger.info("starting new file %s", self.filename)

src/ibex_bluesky_core/callbacks/fitting/livefit_logger.py

@@ -5,7 +5,6 @@
 import os
 from datetime import datetime
 from pathlib import Path
-from platform import node
 from typing import Optional
 from zoneinfo import ZoneInfo
 
@@ -15,14 +14,17 @@
 from event_model.documents.run_start import RunStart
 from event_model.documents.run_stop import RunStop
 
+from ibex_bluesky_core.callbacks._utils import (
+    DATA,
+    DEFAULT_PATH,
+    INSTRUMENT,
+    RB,
+    TIME,
+    UID,
+    UNKNOWN_RB,
+)
 from ibex_bluesky_core.callbacks.fitting import LiveFit
 
-UID = "uid"
-TIME = "time"
-DATA = "data"
-RB = "rb_number"
-INSTRUMENT = node()
-DEFAULT_PATH = Path("//isis.cclrc.ac.uk/inst$") / INSTRUMENT / "user" / "TEST" / "scans"
 logger = logging.getLogger(__name__)
 
 
@@ -78,8 +80,8 @@ def start(self, doc: RunStart) -> None:
         self.output_dir.mkdir(parents=True, exist_ok=True)
         self.current_start_document = doc[UID]
         file = f"{INSTRUMENT}_{self.x}_{self.y}_{title_format_datetime}Z{self.postfix}.txt"
-        rb_num = doc.get("rb_number", "Unknown RB")
-        if rb_num == "Unknown RB":
+        rb_num = doc.get(RB, UNKNOWN_RB)
+        if rb_num == UNKNOWN_RB:
             logger.warning('No RB number found, will save to "Unknown RB"')
         self.filename = self.output_dir / f"{rb_num}" / file
 

tests/callbacks/test_write_log_callback.py

@@ -33,7 +33,7 @@ def test_header_data_all_available_on_start(cb):
         result = (
             save_path
             / f"{run_start.get('rb_number', None)}"
-            / f"{node()}_block_2024-10-04_14-43-43Z.txt"
+            / f"{node()}_block_2024-10-04_13-43-43Z.txt"
         )
 
     mock_file.assert_called_with(result, "a", newline="\n", encoding="utf-8")
@@ -54,7 +54,7 @@ def test_no_rb_number_folder(cb):
         patch("ibex_bluesky_core.callbacks.file_logger.os.makedirs") as mock_mkdir,
     ):
         cb.start(run_start)
-        result = save_path / "Unknown RB" / f"{node()}_block_2024-10-04_14-43-43Z.txt"
+        result = save_path / "Unknown RB" / f"{node()}_block_2024-10-04_13-43-43Z.txt"
         assert mock_mkdir.called
 
     mock_file.assert_called_with(result, "a", newline="\n", encoding="utf-8")
@@ -75,7 +75,7 @@ def test_no_motors_doesnt_append_to_filename(cb):
         patch("ibex_bluesky_core.callbacks.file_logger.os.makedirs") as mock_mkdir,
     ):
         cb.start(run_start)
-        result = save_path / "Unknown RB" / f"{node()}_2024-10-04_14-43-43Z.txt"
+        result = save_path / "Unknown RB" / f"{node()}_2024-10-04_13-43-43Z.txt"
         assert mock_mkdir.called
 
     mock_file.assert_called_with(result, "a", newline="\n", encoding="utf-8")