Use proper beamline name in nexus

src/hyperion/external_interaction/nexus/write_nexus.py

@@ -8,6 +8,7 @@
 import math
 from pathlib import Path
 
+from dodal.utils import get_beamline_name
 from nexgen.nxs_utils import Detector, Goniometer, Source
 from nexgen.nxs_write.NXmxWriter import NXmxFileWriter
 
@@ -48,7 +49,7 @@ def __init__(
         self.beam, self.attenuator = create_beam_and_attenuator_parameters(
             parameters.hyperion_params.ispyb_params
         )
-        self.source: Source = Source(parameters.hyperion_params.beamline)
+        self.source: Source = Source(get_beamline_name("S03"))
         self.directory: Path = Path(
             parameters.hyperion_params.detector_params.directory
         )

tests/unit_tests/external_interaction/test_write_nexus.py

@@ -1,6 +1,7 @@
 import os
 from contextlib import contextmanager
 from typing import Literal
+from unittest.mock import patch
 
 import h5py
 import numpy as np
@@ -20,7 +21,9 @@
 def assert_end_data_correct(nexus_writer: NexusWriter):
     for filename in [nexus_writer.nexus_file, nexus_writer.master_file]:
         with h5py.File(filename, "r") as written_nexus_file:
-            assert "end_time_estimated" in written_nexus_file["entry"]
+            entry = written_nexus_file["entry"]
+            assert isinstance(entry, h5py.Group)
+            assert "end_time_estimated" in entry
 
 
 @pytest.fixture
@@ -107,17 +110,25 @@ def test_given_dummy_data_then_datafile_written_correctly(
 
     for filename in [nexus_writer_1.nexus_file, nexus_writer_1.master_file]:
         with h5py.File(filename, "r") as written_nexus_file:
-            data_path = written_nexus_file["/entry/data"]
+            assert isinstance(
+                data_path := written_nexus_file["/entry/data"], h5py.Group
+            )
             assert_x_data_stride_correct(
                 data_path, grid_scan_params, grid_scan_params.y_steps
             )
+            assert isinstance(sam_y := data_path["sam_y"], h5py.Dataset)
             assert_varying_axis_stride_correct(
-                data_path["sam_y"][:], grid_scan_params, grid_scan_params.y_axis
+                sam_y[:], grid_scan_params, grid_scan_params.y_axis
             )
             assert_axis_data_fixed(written_nexus_file, "z", grid_scan_params.z1_start)
-            assert written_nexus_file["/entry/instrument/beam/total_flux"][()] == 9.0
+            assert isinstance(
+                flux := written_nexus_file["/entry/instrument/beam/total_flux"],
+                h5py.Dataset,
+            )
+            assert flux[()] == 9.0
             assert_contains_external_link(data_path, "data_000001", "dummy_0_000001.h5")
-            assert np.all(data_path["omega"][:] == 0.0)
+            assert isinstance(omega := data_path["omega"], h5py.Dataset)
+            assert np.all(omega[:] == 0.0)
 
             assert np.all(
                 written_nexus_file["/entry/data/omega"].attrs.get("vector")
@@ -151,18 +162,26 @@ def test_given_dummy_data_then_datafile_written_correctly(
 
     for filename in [nexus_writer_2.nexus_file, nexus_writer_2.master_file]:
         with h5py.File(filename, "r") as written_nexus_file:
-            data_path = written_nexus_file["/entry/data"]
+            assert isinstance(
+                data_path := written_nexus_file["/entry/data"], h5py.Group
+            )
             assert_x_data_stride_correct(
                 data_path, grid_scan_params, grid_scan_params.z_steps
             )
+            assert isinstance(sam_z := data_path["sam_z"], h5py.Dataset)
             assert_varying_axis_stride_correct(
-                data_path["sam_z"][:], grid_scan_params, grid_scan_params.z_axis
+                sam_z[:], grid_scan_params, grid_scan_params.z_axis
             )
             assert_axis_data_fixed(written_nexus_file, "y", grid_scan_params.y2_start)
-            assert written_nexus_file["/entry/instrument/beam/total_flux"][()] == 9.0
+            assert isinstance(
+                flux := written_nexus_file["/entry/instrument/beam/total_flux"],
+                h5py.Dataset,
+            )
+            assert flux[()] == 9.0
             assert_contains_external_link(data_path, "data_000001", "dummy_0_000001.h5")
             assert_contains_external_link(data_path, "data_000002", "dummy_0_000002.h5")
-            assert np.all(data_path["omega"][:] == 90.0)
+            assert isinstance(omega := data_path["omega"], h5py.Dataset)
+            assert np.all(omega[:] == 90.0)
             assert np.all(
                 written_nexus_file["/entry/data/sam_z"].attrs.get("vector")
                 == [
@@ -199,9 +218,10 @@ def assert_axis_data_fixed(written_nexus_file, axis, expected_value):
 def assert_data_edge_at(nexus_file, expected_edge_index):
     """Asserts that the datafile's last datapoint is at the specified index"""
     with h5py.File(nexus_file) as f:
-        assert f["entry"]["data"]["data"][expected_edge_index, 0, 0] == 0
+        assert isinstance(data := f["entry/data/data"], h5py.Dataset)
+        assert data[expected_edge_index, 0, 0] == 0
         with pytest.raises(IndexError):
-            assert f["entry"]["data"]["data"][expected_edge_index + 1, 0, 0] == 0
+            assert data[expected_edge_index + 1, 0, 0] == 0
 
 
 def assert_contains_external_link(data_path, entry_name, file_name):
@@ -236,6 +256,14 @@ def test_nexus_writer_writes_width_and_height_correctly(
     )
 
 
+@patch.dict(os.environ, {"BEAMLINE": "I03"})
+def test_nexus_writer_writes_beamline_name_correctly(
+    test_fgs_params,
+):
+    nexus_writer = NexusWriter(test_fgs_params, **test_fgs_params.get_nexus_info(1))
+    assert nexus_writer.source.beamline == "I03"
+
+
 def check_validity_through_zocalo(nexus_writers: tuple[NexusWriter, NexusWriter]):
     import dlstbx.swmr.h5check
 
@@ -282,15 +310,17 @@ def test_given_some_datafiles_outside_of_VDS_range_THEN_they_are_not_in_nexus_fi
 
     for filename in [nexus_writer_1.nexus_file, nexus_writer_1.master_file]:
         with h5py.File(filename, "r") as written_nexus_file:
-            assert "data_000001" in written_nexus_file["entry/data"]
-            assert "data_000002" in written_nexus_file["entry/data"]
-            assert "data_000003" not in written_nexus_file["entry/data"]
+            assert isinstance(data := written_nexus_file["entry/data"], h5py.Dataset)
+            assert "data_000001" in data
+            assert "data_000002" in data
+            assert "data_000003" not in data
 
     for filename in [nexus_writer_2.nexus_file, nexus_writer_2.master_file]:
         with h5py.File(filename, "r") as written_nexus_file:
-            assert "data_000001" not in written_nexus_file["entry/data"]
-            assert "data_000002" in written_nexus_file["entry/data"]
-            assert "data_000003" in written_nexus_file["entry/data"]
+            assert isinstance(data := written_nexus_file["entry/data"], h5py.Dataset)
+            assert "data_000001" not in data
+            assert "data_000002" in data
+            assert "data_000003" in data
 
 
 def test_given_data_files_not_yet_written_when_nexus_files_created_then_nexus_files_still_written(