Extend MRC to Nexus converter to work with TVIPS

pyproject.toml

@@ -26,6 +26,7 @@ dependencies = [
     "scanspec",
     "dataclasses-json",
     "pydantic",
+    "mrcfile>=1.5.3",
 ]
 license.file = "LICENSE"
 readme = "README.rst"

src/nexgen/beamlines/ED_params.py

@@ -47,3 +47,8 @@
     Facility("Diamond Light Source", "DLS", "Electron Source", "DIAMOND MICROSCOPE"),
     "electron",
 )
+
+UnknownSource = Source("unknown",
+                       Facility("unknown", "unknown", "Electron Source", None),
+                       "electron")
+

src/nexgen/command_line/ED_mrc_to_nexus.py

@@ -9,42 +9,86 @@
 
 import numpy as np
 
-from nexgen.beamlines.ED_params import ED_coord_system, EDCeta, EDSource
-from nexgen.nxs_utils import Attenuator, Beam, CetaDetector, Detector, Goniometer
+from nexgen.beamlines.ED_params import ED_coord_system, EDCeta, EDSource, UnknownSource
+from nexgen.nxs_utils import (
+    Attenuator,
+    Beam,
+    CetaDetector,
+    Detector,
+    Goniometer,
+    TVIPSDetector,
+)
 from nexgen.nxs_utils.scan_utils import calculate_scan_points
 from nexgen.nxs_write.ed_nxmx_writer import EDNXmxFileWriter
 from nexgen.tools.mrc_tools import collect_data, get_metadata
 
-logger = logging.getLogger("nexgen.ED_mrc_to_nexus")
+logger = logging.getLogger('nexgen.ED_mrc_to_nexus')
 
 
 def main():
 
-    parser = ArgumentParser(add_help=False)
+    msg = "Convert electron diffraction data from an MRC format "
+    msg += "to a NeXus format"
+    parser = ArgumentParser(description=msg, add_help=True)
+
+    parser.add_argument(
+        "--detector",
+        type=str,
+        default="cetad",
+        help="Detector type: use --detector cetad or --detector tvips",
+    )
+
+    parser.add_argument(
+        "--facility",
+        type=str,
+        default=None,
+        help="Name of the facility (e.g. Diamond Light Source)."
+    )
+
+    parser.add_argument(
+        "--facility-short",
+        type=str,
+        default=None,
+        help="Short name of the facility (e.g. DLS)."
+    )
+
+    msg = "Facility ID (e.g. DIAMOND). "
+    msg += "Naming tries to follow the recommended convention for NXmx: "
+    msg += "https://mmcif.wwpdb.org/dictionaries/"
+    msg += "mmcif_pdbx_v50.dic/Items/_diffrn_source.type.html"
+
+    parser.add_argument(
+        "--facility-id",
+        type=str,
+        default=None,
+        help=msg
+    )
+
     parser.add_argument(
         "--det_distance",
         type=float,
         default=None,
         help="The sample-detector distance.",
     )
+
     parser.add_argument(
         "-wl",
         "--wavelength",
         type=float,
         default=None,
-        help="Incident beam wavelength, in A.",
+        help="Incident beam wavelength, in Angstroms.",
     )
     parser.add_argument(
-        "--axis-start",
+        "--angle-start",
         type=float,
         default=None,
-        help="Rotation axis start position.",
+        help="Starting angle of a rotation scan.",
     )
     parser.add_argument(
-        "--axis-inc",
+        "--angle-inc",
         type=float,
         default=None,
-        help="Rotation axis increment.",
+        help="Angle increment of a rotation scan.",
     )
     parser.add_argument(
         "-e",
@@ -59,7 +103,17 @@ def main():
         nargs=2,
         help="Beam center (x,y) positions.",
     )
-    parser.add_argument("input_files", nargs="+", help="List of input files")
+    parser.add_argument(
+        "--dtype",
+        type=str,
+        nargs=1,
+        default=None,
+        help="Data type for the HDF5 output (int32, int64, float64, ...)"
+    )
+    des = "List of input files. Can be a single MRC file containing all the "
+    des += "images, or a list of files containing single images, "
+    des += "usually obtained by global expansion (e.g. *mrc)"
+    parser.add_argument('input_files', nargs='+', help=des)
 
     args = parser.parse_args()
 
@@ -75,7 +129,6 @@ def main():
     tot_imgs, out_file, angles = collect_data(mrc_files)
 
     mdict = get_metadata(mrc_files[0])
-
     attenuator = Attenuator(transmission=None)
 
     if args.det_distance is not None:
@@ -87,51 +140,52 @@ def main():
         logger.info(msg)
     else:
         msg = "No detector distance in the MRC metadata. "
-        msg += "You can set it with -det_distance option."
+        msg += "You can set it with the --det-distance option."
         raise ValueError(msg)
 
     if args.wavelength is not None:
         beam = Beam(args.wavelength)
     elif "wavelength" in mdict:
         beam = Beam(mdict["wavelength"])
-        msg = "Reading wavelength from the MRC files: %f" % mdict["wavelength"]
+        msg = ("Reading wavelength from the MRC files: %f" %
+               mdict['wavelength'])
         logger.info(msg)
     else:
         msg = "No wavelength in the MRC metadata. "
-        msg += "You can set it with --wavelength option."
+        msg += "You can set it with the --wavelength option."
         raise ValueError(msg)
 
-    if args.axis_start is not None:
-        start_angle = args.axis_start
+    if args.angle_start is not None:
+        start_angle = args.angle_start
     elif angles[0] is not None:
         start_angle = angles[0]
-        msg = "Reading starting angle from the MRC files: %.2f" % start_angle
+        msg = ("Reading starting angle from the MRC files: %.2f" % start_angle)
         logger.info(msg)
     else:
         msg = "No starting angle in the MRC metadata. "
-        msg += "You can set it with --axis_start option."
+        msg += "You can set it with the --angle-start option."
         raise ValueError(msg)
 
-    if args.axis_inc is not None:
-        increment = args.axis_inc
+    if args.angle_inc is not None:
+        increment = args.angle_inc
     elif "tiltPerImage" in mdict:
         increment = mdict["tiltPerImage"]
-        msg = "Reading angle increment from the MRC files: %f" % increment
+        msg = ("Reading angle increment from the MRC files: %f" % increment)
         logger.info(msg)
     else:
         msg = "No angle increment in the MRC metadata. "
-        msg += "You can set it with --axis_inc option."
+        msg += "You can set it with the --angle-inc option."
         raise ValueError(msg)
 
     if args.exp_time is not None:
         exposure_time = args.exp_time
     elif "integrationTime" in mdict:
         exposure_time = mdict["integrationTime"]
-        msg = "Reading exposure time from the MRC files: %f" % exposure_time
+        msg = ("Reading exposure time from the MRC files: %f" % exposure_time)
         logger.info(msg)
     else:
         msg = "No exposure time in the MRC metadata. "
-        msg += "You can set it with --exp_time option."
+        msg += "You can set it with --exp-time option."
         raise ValueError(msg)
 
     if args.beam_center is not None:
@@ -142,36 +196,78 @@ def main():
             msg = "Beam center requires two arguments"
             raise ValueError(msg)
     elif ("beamCentreXpx" in mdict) and ("beamCentreYpx" in mdict):
-        beam_center = (mdict["beamCentreXpx"], mdict["beamCentreYpx"])
-        msg = "Reading beam center from the MRC files: (%.1f, %.1f)" % beam_center
+        beam_center = (mdict["beamCentreXpx"],
+                       mdict["beamCentreYpx"])
+        msg = ("Reading beam center from the MRC files: (%.1f, %.1f)" %
+               beam_center)
         logger.info(msg)
     else:
         msg = "No beam center in the MRC metadata. "
-        msg += "You can set it with --beam_center option."
+        msg += "You can set it with --beam-center option."
         raise ValueError(msg)
 
     gonio_axes = EDCeta.gonio
     gonio_axes[0].start_pos = start_angle
     gonio_axes[0].increment = increment
     gonio_axes[0].num_steps = tot_imgs
 
-    scan = calculate_scan_points(gonio_axes[0], rotation=True, tot_num_imgs=tot_imgs)
+    scan = calculate_scan_points(gonio_axes[0],
+                                 rotation=True,
+                                 tot_num_imgs=tot_imgs)
     goniometer = Goniometer(gonio_axes, scan)
 
     nx = mdict["nx"]
     ny = mdict["ny"]
 
-    if (nx == 2048) and (ny == 2048):
-        binning = 2
-    else:
-        binning = 1
+    if args.detector == "tvips":
+
+        logger.info("Detector set to tvips")
+        det_params = TVIPSDetector("TVIPS Detector", [nx, ny])
+        if (nx == 4096) and (ny == 4096):
+            binning = 1
+        elif (nx == 2048) and (ny == 2048):
+            binning = 2
+        elif (nx == 1024) and (ny == 1024):
+            binning = 4
+        elif (nx == 512) and (ny == 512):
+            binning = 8
+        else:
+            msg = f"Non standard detector size ({nx}, {ny})."
+            msg += "Expecting (4096/b, 4096/b) with binning b = 1, 2, 4, 8"
+            raise ValueError(msg)
+
+        pixel_size = 0.015500 * binning
+        source = UnknownSource
+        if args.facility:
+            source.name = args.facility
+        if args.facility_short:
+            source.short_name = args.facility_short
+        if args.facility_id:
+            source.facility_id = args.facility_id
+        source.beamline = "TVIPS"
+
+    elif args.detector == "cetad":
 
-    det_params = CetaDetector("Thermo Fisher Ceta-D", [nx, ny])
+        logger.info("Detector set to cetad")
+        det_params = CetaDetector("Thermo Fisher Ceta-D", [nx, ny])
 
-    pixel_size = 0.014 * binning
-    ps_str = "%5.3fmm" % pixel_size
+        if (nx == 2048) and (ny == 2048):
+            binning = 2
+        else:
+            binning = 1
+
+        pixel_size = 0.014 * binning
+        det_params.overload = 8000 * binning**2
+        source = EDSource
+        source.beamline = "CetaD-eBIC"
+
+    else:
+        msg = f"Unkown detector type {args.detector}."
+        msg += "Use either --detector=cetad or --detector=tvips."
+        raise ValueError(msg)
+
+    ps_str = '%5.7fmm' % pixel_size
     det_params.pixel_size = [ps_str, ps_str]
-    det_params.overload = 8000 * binning**2
 
     extra_params = {}
     mask = np.zeros((nx, ny), dtype=np.int32)
@@ -186,16 +282,10 @@ def main():
 
     det_axes[0].start_pos = det_distance
 
-    detector = Detector(
-        det_params,
-        det_axes,
-        beam_center,
-        exposure_time,
-        [EDCeta.fast_axis, EDCeta.slow_axis],
-    )
-
-    source = EDSource
-    source.beamline = "Ceta"
+    detector = Detector(det_params, det_axes, beam_center,
+                        exposure_time,
+                        [EDCeta.fast_axis,
+                         EDCeta.slow_axis])
 
     script_dir = os.path.dirname(full_mrc_path)
     file_name = out_file.replace("h5", "nxs")
@@ -206,17 +296,17 @@ def main():
     msg = "Writing the Nexus file %s" % full_path
     logger.info(msg)
     writer = EDNXmxFileWriter(
-        full_path,
-        goniometer,
-        detector,
-        source,
-        beam,
-        attenuator,
-        tot_imgs,
-        ED_coord_system,
-    )
+         full_path,
+         goniometer,
+         detector,
+         source,
+         beam,
+         attenuator,
+         tot_imgs,
+         ED_coord_system)
 
     datafiles = [data_path]
     writer.write(datafiles, "/entry/data/data")
-    writer.write_vds(vds_dtype=np.int32, datafiles=datafiles)
+    writer.write_vds(vds_dtype=np.int32,
+                     datafiles=datafiles)
     logger.info("MRC images converted to Nexus.")

src/nexgen/nxs_utils/__init__.py

@@ -10,6 +10,7 @@
     JungfrauDetector,
     SinglaDetector,
     TristanDetector,
+    TVIPSDetector,
 )
 from .goniometer import Goniometer
 from .sample import Sample
@@ -33,4 +34,5 @@
     "TransformationType",
     "Facility",
     "CetaDetector",
+    "TVIPSDetector"
 ]

src/nexgen/nxs_utils/detector.py

@@ -69,6 +69,35 @@ class UnknownDetectorTypeError(Exception):
     "software_version": "0.0.0",
 }
 
+TVIPS_CONST = {
+    "flatfield": None,
+    "flatfield_applied": False,
+    "pixel_mask": None,
+    "pixel_mask_applied": False,
+    "software_version": "0.0.0",
+}
+
+
+@dataclass
+class TVIPSDetector(DataClassJsonMixin):
+    """Define a TVIPS camera """
+
+    description: str
+    image_size: List[float] | Tuple[float]
+    pixel_size: str = "0.0155000mm"
+    sensor_material: str = "Si"
+    sensor_thickness: str = "0.0000000000001mm"
+    detector_type: str = "CMOS"
+    overload: int = 65534
+    underload: int = 0
+
+    @property
+    def constants(self) -> Dict:
+        return TVIPS_CONST
+
+    @property
+    def hasMeta(self) -> bool:
+        return False
 
 @dataclass
 class EigerDetector(DataClassJsonMixin):