Readers

The following Readers are available

• DRSReader — Detector Responsivity System file format (MSL Light Standards)
• HDF5Reader — HDF5 file format
• JSONReader — Read a file created by JSONWriter
• RegularTransmittanceReader — Spectrophotometer transmittance file format (MSL Light Standards)

DRSReader

DRSReader(file: ReadLike | str)

Bases: Reader

Reader for the Detector Responsivity System in MSL Light Standards.

Source code in src/msl/io/base.py

def __init__(self, file: ReadLike | str) -> None:
    """Abstract base class for a [Reader][msl-io-readers].

    Args:
        file: The file to read.
    """
    super().__init__(file)
    self._file: ReadLike | str = file

can_read staticmethod

can_read(file: ReadLike | str, **kwargs: Any) -> bool

Checks if the first line starts with DRS and ends with Shindo.

Parameters:

Name	Type	Description	Default
file	ReadLike | str	The file to check.	required
kwargs	Any	All keyword arguments are ignored.	{}

Returns:

Type	Description
bool	Whether file was acquired in the DRS laboratory.

Source code in src/msl/io/readers/detector_responsivity_system.py

@staticmethod
def can_read(file: ReadLike | str, **kwargs: Any) -> bool:  # noqa: ARG004
    """Checks if the first line starts with `DRS` and ends with `Shindo`.

    Args:
        file: The file to check.
        kwargs: All keyword arguments are ignored.

    Returns:
        Whether `file` was acquired in the DRS laboratory.
    """
    if get_extension(file).lower() != ".dat":
        return False

    line = get_lines(file, 1)[0]
    if isinstance(line, bytes):
        line = line.decode()

    return line.startswith(("DRS", '"DRS')) and line.endswith(("Shindo", 'Shindo"'))

read

read(**kwargs: Any) -> None

Reads the .DAT and corresponding .LOG file.

Parameters:

Name	Type	Description	Default
kwargs	Any	All keyword arguments are ignored.	{}

Source code in src/msl/io/readers/detector_responsivity_system.py

def read(self, **kwargs: Any) -> None:  # noqa: ARG002
    """Reads the `.DAT` and corresponding `.LOG` file.

    Args:
        kwargs: All keyword arguments are ignored.
    """
    self._default_alias: dict[str, str] = {"l(nm)": "wavelength"}

    assert isinstance(self.file, str)  # noqa: S101
    self._lines_dat: list[str] = get_lines(self.file, remove_empty_lines=True)
    self._num_lines_dat: int = len(self._lines_dat)

    self._lines_log: list[str] = get_lines(self.file[:-3] + "LOG", remove_empty_lines=True)
    self._num_lines_log: int = len(self._lines_log)

    num_runs = 0
    self._index_dat: int = 0
    self._index_log: int = 0
    while self._index_dat < self._num_lines_dat:
        if "DRS" in self._lines_dat[self._index_dat]:
            num_runs += 1
            group = self.create_group(f"run{num_runs}")
            self._read_run(group)
        else:
            self._index_dat += 1
            self._index_log += 1

HDF5Reader

HDF5Reader(file: ReadLike | str)

Bases: Reader

Reader for the HDF5 file format.

Info

This Reader loads the entire HDF5 file in memory. If you need to use any of the more advanced features of an HDF5 file, it is best to directly load the file using h5py.

Source code in src/msl/io/base.py

def __init__(self, file: ReadLike | str) -> None:
    """Abstract base class for a [Reader][msl-io-readers].

    Args:
        file: The file to read.
    """
    super().__init__(file)
    self._file: ReadLike | str = file

can_read staticmethod

can_read(file: ReadLike | str, **kwargs: Any) -> bool

Checks if the file has the HDF5 signature.

Parameters:

Name	Type	Description	Default
file	ReadLike | str	The file to check.	required
kwargs	Any	All keyword arguments are ignored.	{}

Returns:

Type	Description
bool	Whether the first 8 bytes are \x89HDF\r\n\x1a\n.

Source code in src/msl/io/readers/hdf5.py

@staticmethod
def can_read(file: ReadLike | str, **kwargs: Any) -> bool:  # noqa: ARG004
    r"""Checks if the file has the [HDF5 signature]{:target="_blank"}.

    [HDF5 signature]: https://support.hdfgroup.org/documentation/hdf5/latest/_f_m_t11.html#subsec_fmt11_boot_super

    Args:
        file: The file to check.
        kwargs: All keyword arguments are ignored.

    Returns:
        Whether the first 8 bytes are `\x89HDF\r\n\x1a\n`.
    """
    if isinstance(file, (str, BufferedIOBase)):
        return get_bytes(file, 8) == b"\x89HDF\r\n\x1a\n"
    return False

read

read(**kwargs: Any) -> None

Reads the HDF5 file.

Parameters:

Name	Type	Description	Default
kwargs	Any	All keyword arguments are passed to h5py.File.	{}

Source code in src/msl/io/readers/hdf5.py

def read(self, **kwargs: Any) -> None:
    """Reads the [HDF5](https://www.hdfgroup.org/) file.

    Args:
        kwargs: All keyword arguments are passed to [h5py.File][]{:target="_blank"}.
    """
    if h5py is None:
        msg = "You must install h5py to read HDF5 files, run\n  pip install h5py"
        raise ImportError(msg)

    @no_type_check
    def convert(name: str, obj: object) -> None:
        head, tail = os.path.split(name)
        s = self["/" + head] if head else self
        if isinstance(obj, h5py.Dataset):
            _ = s.create_dataset(tail, data=obj[:], **obj.attrs)
        elif isinstance(obj, h5py.Group):
            _ = s.create_group(tail, **obj.attrs)
        else:
            msg = f"Should never get here, unhandled h5py object {obj}"
            raise TypeError(msg)

    @no_type_check
    def h5_open(f: BufferedIOBase) -> None:
        with h5py.File(f, mode="r", **kwargs) as h5:
            self.add_metadata(**h5.attrs)
            h5.visititems(convert)  # cSpell: ignore visititems

    # Calling h5py.File on a file on a mapped drive could raise
    # an OSError. This occurred when a local folder was shared
    # and then mapped on the same computer. Opening the file
    # using open() and then passing in the file handle to
    # h5py.File is more universal
    if isinstance(self.file, BufferedIOBase):
        h5_open(self.file)
    elif isinstance(self.file, str):
        with open(self.file, mode="rb") as fp:  # noqa: PTH123
            h5_open(fp)
    else:
        msg = f"Should never get here, file type is {type(self.file)}"
        raise TypeError(msg)

JSONReader

JSONReader(file: ReadLike | str)

Bases: Reader

Read a file that was created by JSONWriter.

Source code in src/msl/io/base.py

def __init__(self, file: ReadLike | str) -> None:
    """Abstract base class for a [Reader][msl-io-readers].

    Args:
        file: The file to read.
    """
    super().__init__(file)
    self._file: ReadLike | str = file

can_read staticmethod

can_read(file: ReadLike | str, **kwargs: Any) -> bool

Checks if the file was created by JSONWriter.

Parameters:

Name	Type	Description	Default
file	ReadLike | str	The file to check.	required
kwargs	Any	All keyword arguments are passed to get_lines.	{}

Returns:

Type	Description
bool	Whether the text MSL JSONWriter is in the first line of the file.

Source code in src/msl/io/readers/json_.py

@staticmethod
def can_read(file: ReadLike | str, **kwargs: Any) -> bool:
    """Checks if the file was created by [JSONWriter][msl.io.writers.json_.JSONWriter].

    Args:
        file: The file to check.
        kwargs: All keyword arguments are passed to [get_lines][msl.io.utils.get_lines].

    Returns:
        Whether the text `MSL JSONWriter` is in the first line of the file.
    """
    text: bytes | str
    if isinstance(file, (str, BufferedIOBase)):
        text = get_bytes(file, 21, 34)
    else:
        text = get_lines(file, 1, **kwargs)[0][20:34]

    if isinstance(text, str):
        text = text.encode()
    return text == b"MSL JSONWriter"

read

read(**kwargs: Any) -> None

Read the file that was created by JSONWriter.

Parameters:

Name	Type	Description	Default
kwargs	Any	Accepts encoding and errors keyword arguments which are passed to open. The default encoding value is utf-8 and the default errors value is strict. All additional keyword arguments are passed to json.loads.	{}

Source code in src/msl/io/readers/json_.py

def read(self, **kwargs: Any) -> None:  # noqa: C901
    """Read the file that was created by [JSONWriter][msl.io.writers.json_.JSONWriter].

    Args:
        kwargs:  Accepts `encoding` and `errors` keyword arguments which are passed to
            [open][]. The default `encoding` value is `utf-8` and the default
            `errors` value is `strict`. All additional keyword arguments are passed to
            [json.loads][].
    """
    open_kwargs = {
        "encoding": kwargs.get("encoding", "utf-8"),
        "errors": kwargs.pop("errors", "strict"),
    }

    if isinstance(self.file, str):
        with open(self.file, mode="rt", **open_kwargs) as fp:  # noqa: PTH123, UP015
            _ = fp.readline()  # skip the first line
            dict_ = json.loads(fp.read(), **kwargs)
    else:
        _ = self.file.readline()  # skip the first line
        data = self.file.read()
        if isinstance(data, bytes):
            data = data.decode(**open_kwargs)
        dict_ = json.loads(data, **kwargs)

    def create_group(parent: Group | None, name: str, node: dict[str, Any]) -> None:
        group = self if parent is None else parent.create_group(name)
        for key, value in node.items():
            if not isinstance(value, dict):  # Metadata
                group.metadata[key] = value
            elif "dtype" in value and "data" in value:  # Dataset
                kws: dict[str, Any] = {}
                for d_key, d_val in value.items():  # pyright: ignore[reportUnknownVariableType]
                    if d_key == "data":
                        pass  # handled in the 'dtype' check
                    elif d_key == "dtype":
                        if isinstance(d_val, list):
                            kws["data"] = np.asarray(
                                [tuple(row) for row in value["data"]],  # pyright: ignore[reportUnknownVariableType, reportUnknownArgumentType]
                                dtype=[tuple(item) for item in d_val],  # pyright: ignore[reportUnknownVariableType, reportUnknownArgumentType]
                            )
                        else:
                            kws["data"] = np.asarray(value["data"], dtype=d_val)  # pyright: ignore[reportUnknownArgumentType]
                    else:  # Metadata
                        kws[d_key] = d_val
                _ = group.create_dataset(key, **kws)
            else:  # use recursion to create a sub-Group
                create_group(group, key, value)  # pyright: ignore[reportUnknownArgumentType]

    # create the root group
    create_group(None, "", dict_)

RegularTransmittanceReader

RegularTransmittanceReader(file: ReadLike | str)

Bases: Reader

Reader for Trans files from Light Standards at MSL.

Source code in src/msl/io/base.py

def __init__(self, file: ReadLike | str) -> None:
    """Abstract base class for a [Reader][msl-io-readers].

    Args:
        file: The file to read.
    """
    super().__init__(file)
    self._file: ReadLike | str = file

can_read staticmethod

can_read(file: ReadLike | str, **kwargs: Any) -> bool

Checks if the file extension is .dat and the filename starts with Trans_.

Parameters:

Name	Type	Description	Default
file	ReadLike | str	The file to check.	required
kwargs	Any	All keyword arguments are ignored.	{}

Source code in src/msl/io/readers/spectrophotometer_trans_reader.py

@staticmethod
def can_read(file: ReadLike | str, **kwargs: Any) -> bool:  # noqa: ARG004
    """Checks if the file extension is `.dat` and the filename starts with `Trans_`.

    Args:
        file: The file to check.
        kwargs: All keyword arguments are ignored.
    """
    filename = get_basename(file)
    return filename.startswith("Trans_") and filename.endswith((".dat", ".DAT"))

read

read(**kwargs: Any) -> None

Reads the data in the corresponding .log and .celsius files.

Parameters:

Name	Type	Description	Default
kwargs	Any	All keyword arguments are ignored.	{}

Source code in src/msl/io/readers/spectrophotometer_trans_reader.py

def read(self, **kwargs: Any) -> None:  # noqa: ARG002, C901, PLR0915
    """Reads the data in the corresponding `.log` and `.celsius` files.

    Args:
        kwargs: All keyword arguments are ignored.
    """
    assert isinstance(self.file, str)  # noqa: S101
    lines_log = get_lines(self.file[:-3] + "log", remove_empty_lines=True)
    num_lines_log = len(lines_log)
    group = self.create_group("trans_data")
    meta: dict[str, Any] = {}

    meta["start_time"] = self._convert_time(lines_log[0].split(",")[1], lines_log[1])
    try:
        meta["end_time"] = self._convert_time(lines_log[-2].split(",")[1], lines_log[-1])
    except IndexError:
        meta["note"] = "No end time recorded."
    this_line = 2

    # Read temperature data out of celsius file
    meta["avg_temp"] = self._read_celsius_file()

    # skip lines until we get to the wavelength information and save comment
    comment: list[str] = []
    while not lines_log[this_line].startswith("Wavelengths Settings:"):
        comment.append(lines_log[this_line])
        this_line += 1

    meta["comment"] = "\n".join(comment)
    this_line += 1

    if lines_log[this_line].startswith("Wavelengths read"):
        meta["wavelength input"] = "Read from text file"
    else:
        meta["wavelength input"] = "Manually input"

    while not lines_log[this_line].startswith("TEST MODULE:"):
        this_line += 1

    this_line += 1
    if lines_log[this_line].startswith("DVM INIT"):
        meta["dvm init"] = lines_log[this_line]

    this_line += 2
    if lines_log[this_line].startswith("DELAY"):
        meta["delay"] = float(lines_log[this_line].split()[1])

    this_line += 2
    if lines_log[this_line].startswith("NUM SAMPLES"):
        n_samples = int(lines_log[this_line].split()[2])
        meta["n_samples"] = n_samples

    wavelengths: list[str] = []
    dark_current: list[float] = []
    dark_current_u: list[float] = []
    dark_current_m: list[float] = []
    dark_current_m_u: list[float] = []
    i0: list[float] = []
    i0_u: list[float] = []
    m0: list[float] = []
    m0_u: list[float] = []
    corr: list[float] = []
    is_: list[tuple[float, ...]] = []
    is_u: list[tuple[float, ...]] = []
    is_m: list[tuple[float, ...]] = []
    is_m_u: list[tuple[float, ...]] = []
    is_corr: list[tuple[float, ...]] = []

    n_measurements = 0

    while this_line < num_lines_log - 1:
        if lines_log[this_line].startswith("Wavelength = "):
            n_measurements += 1
            wavelengths.append(lines_log[this_line].split()[2])
            i0.append(float(re.split("=|\t", lines_log[this_line + 1])[1]))
            i0_u.append(float(re.split("=|\t", lines_log[this_line + 1])[3]))
            m0.append(float(re.split("=|\t", lines_log[this_line + 2])[1]))
            m0_u.append(float(re.split("=|\t", lines_log[this_line + 2])[3]))
            corr.append(float(lines_log[this_line + 3].split("=")[1]))
            dark_current.append(float(re.split("=|\t", lines_log[this_line + 4])[1]))
            dark_current_u.append(float(re.split("=|\t", lines_log[this_line + 4])[3]))
            dark_current_m.append(float(re.split("=|\t", lines_log[this_line + 5])[1]))
            dark_current_m_u.append(float(re.split("=|\t", lines_log[this_line + 5])[3]))
            is_.append(tuple([float(item) for item in lines_log[this_line + 7].split("|")[1:] if item]))
            is_u.append(tuple([float(item) for item in lines_log[this_line + 8].split("|")[1:] if item]))
            is_m.append(tuple([float(item) for item in lines_log[this_line + 9].split("|")[1:] if item]))
            is_m_u.append(tuple([float(item) for item in lines_log[this_line + 10].split("|")[1:] if item]))
            is_corr.append(tuple([float(item) for item in lines_log[this_line + 11].split("|")[1:] if item]))

            this_line += 10

        else:
            this_line += 1

    fieldnames = [
        "wavelength",
        "signal",
        "u_signal",
        "mon_signal",
        "u_mon_signal",
        "correlation",
        "dark_current",
        "u_dark_current",
        "mon_dark",
        "u_mon_dark",
    ]
    data = np.array(
        list(
            zip(
                wavelengths,
                i0,
                i0_u,
                m0,
                m0_u,
                corr,
                dark_current,
                dark_current_u,
                dark_current_m,
                dark_current_m_u,
            )
        ),
        dtype=[(name, float) for name in fieldnames],
    )
    _ = group.create_dataset("data", data=data, **meta)

    fieldnames = [
        "Transmitted signal",
        "u(Transmitted signal)",
        "Monitor transmitted",
        "u(Monitor transmitted)",
        "Transmitted correlation",
    ]
    d_names = ["Is", "Is_u", "Is_M", "Is_M_u", "Is_corr"]
    sample_data = np.array(
        (list(zip(*is_)), list(zip(*is_u)), list(zip(*is_m)), list(zip(*is_m_u)), list(zip(*is_corr)))
    )
    ind = 1
    for sample in sample_data:
        _ = group.create_dataset(d_names[ind - 1], data=np.array(sample), **meta)
        ind += 1