pme

PME Instruments

https://www.pme.com/

minidot_cat(path, read_csv_kwargs=None)

cat reads PME MiniDot concatenated CAT files

Source code in ocean_data_parser/parsers/pme.py

def minidot_cat(path: str, read_csv_kwargs: dict = None) -> xr.Dataset:
    """
    cat reads PME MiniDot concatenated CAT files
    """
    if read_csv_kwargs is None:
        read_csv_kwargs = {}
    with open(path, "r", encoding=read_csv_kwargs.get("encoding", "utf8")) as f:
        header = f.readline()

        if header != "MiniDOT Logger Concatenated Data File\n":
            raise RuntimeError(
                "Can't recognize the CAT file! \nCAT File should start with ''MiniDOT Logger Concatenated Data File'"
            )
        # Read header and column names and units
        header = [f.readline() for _ in range(6)]
        columns = [f.readline() for _ in range(2)]

        names = columns[0].replace(r"\n", "").split(",")
        units = columns[1].replace(r"\n", "")

        ds = pd.read_csv(f, names=names, **read_csv_kwargs).to_xarray()

    # Include units
    for name, units in zip(names, units):
        if units:
            ds[name].attrs[units] = units

    # Extract metadata from header
    ds.attrs = re.search(
        (
            r"Sensor:\s*(?P<instrument_sn>.*)\n"
            + r"Concatenation Date:\s*(?P<concatenation_date>.*)\n\n"
            + r"DO concentration compensated for salinity:\s*(?P<reference_salinity>.*)\n"
            + r"Saturation computed at elevation:\s*(?P<elevation>.*)\n"
        ),
        "".join(header),
    ).groupdict()

    return ds

minidot_txt(path, read_csv_kwargs=None, rename_variables=True)

minidot_txt parses the txt format provided by the PME Minidot instruments.

Source code in ocean_data_parser/parsers/pme.py

def minidot_txt(
    path: str, read_csv_kwargs: dict = None, rename_variables: bool = True
) -> xr.Dataset:
    """
    minidot_txt parses the txt format provided by the PME Minidot instruments.
    """

    def _append_to_history(msg):
        ds.attrs["history"] += f"{pd.Timestamp.utcnow():%Y-%m-%dT%H:%M:%SZ} {msg}"

    # Default read_csv_kwargs
    if read_csv_kwargs is None:
        read_csv_kwargs = {}

    # Read MiniDot
    with open(
        path,
        "r",
        encoding=read_csv_kwargs.get("encoding", "utf-8"),
        errors=read_csv_kwargs.get("encoding_errors"),
    ) as f:
        # Read the headre
        serial_number = f.readline().replace("\n", "")
        logger.debug("Parse file from serial number: %s", serial_number)
        metadata = re.search(
            (
                r"OS REV: (?P<software_version>\d+\.\d+)\s"
                r"Sensor Cal: (?P<instrument_calibration>\d*)"
            ),
            f.readline(),
        )

        # If metadata is null than it's likely not a minidot file
        if metadata is None:
            warnings.warn("Failed to read: {path}", RuntimeWarning)
            return pd.DataFrame(), None

        # Read the data with pandas
        ds = pd.read_csv(
            f,
            converters={0: lambda x: pd.Timestamp(int(x), unit="s", tz="UTC")},
            **read_csv_kwargs,
        ).to_xarray()

    # Strip whitespaces from variables names
    ds = ds.rename({var: var.strip() for var in ds.keys()})

    # Global attributes
    ds.attrs = {
        **global_attributes,
        **metadata.groupdict(),
        "instrument_manufacturer": "PME",
        "instrument_model": "MiniDot",
        "instrument_sn": serial_number,
        "history": "",
    }

    # Retrieve raw saturation values from minidot
    #  assume:
    #   - 0 salinity
    #   - surface (pressure=0).
    if "DO (mg/l)" in ds:
        ds["DO (perc)"] = O2ctoO2s(31.2512 * ds["DO (mg/l)"], ds["T (deg C)"], S=0, P=0)
        _append_to_history(
            "Derive DO (perc) from = "
            "o2Conversion.O2ctoO2s( 31.2512*'DO (mg/l)', 'T (deg C)', S=0, P=0)",
        )

        ds["pO2 (mbar)"] = O2ctoO2p(
            31.2512 * ds["DO (mg/l)"], ds["T (deg C)"], S=0, P=0
        )
        _append_to_history(
            "Derive pO2 (mbar) from = "
            "o2Conversion.O2ctoO2s(31.2512*'DO (mg/l)', 'T (deg C)', S=0, P=0)",
        )

    # Add attributes to the dataset and rename variables to mapped names.
    for var in ds.variables:
        if var not in variable_attributes:
            logger.warning("Unknown variable: %s", var)
            continue
        ds[var].attrs = variable_attributes[var]

    if rename_variables:
        ds = ds.rename_vars(vars_rename)
    ds.attrs[
        "history"
    ] += f"\n{pd.Timestamp.now().isoformat()} Rename variables: {vars_rename}"

    ds = standardize_dataset(ds)
    return ds

minidot_txts(paths)

Parse PME Minidots txt files

Parameters:

Name	Type	Description	Default
paths	listorstr	List of file paths to read.	required

Returns:

Type	Description
xr.Dataset	xr.Dataset: xarray dataset which is compliant with CF-1.6

Source code in ocean_data_parser/parsers/pme.py

def minidot_txts(paths: Union[list, str]) -> xr.Dataset:
    """Parse PME Minidots txt files

    Args:
        paths (listorstr): List of file paths to read.

    Returns:
        xr.Dataset: xarray dataset which is compliant with CF-1.6
    """
    # If a single string is givien, assume only one path
    if type(paths) is str:
        paths = [paths]

    datasets = []
    for path in paths:
        # Ignore concatenated Cat.TXT files or not TXT file
        if path.endswith("Cat.TXT") or not path.endswith(("TXT", "txt")):
            print(f"Ignore {path}")
            continue
        # Read txt file
        datasets += minidot_txt(path)

    return xr.merge(datasets)