Source code for dclab.rtdc_dataset.fmt_tdms

"""RT-DC .tdms file format"""
import copy
import pathlib
import time

from ...external.packaging import parse as parse_version

from .exc import ContourIndexingError  # noqa:F401
from .exc import InvalidTDMSFileFormatError  # noqa:F401
from .exc import IncompleteTDMSFileFormatError  # noqa:F401
from .exc import InvalidVideoFileError  # noqa:F401

    import nptdms
except ModuleNotFoundError:
    if parse_version(nptdms.__version__) < parse_version("0.23.0"):
        raise ValueError("Please install nptdms>=0.23.0")
    from .event_contour import ContourColumn
    from .event_image import ImageColumn
    from .event_mask import MaskColumn
    from .event_trace import TraceColumn

import numpy as np

from ... import definitions as dfn
from ...util import hashobj, hashfile

from ..config import Configuration
from ..core import RTDCBase

from . import naming

[docs] class RTDC_TDMS(RTDCBase): def __init__(self, tdms_path, *args, **kwargs): """TDMS file format for RT-DC measurements Parameters ---------- tdms_path: str or pathlib.Path Path to a '.tdms' measurement file. *args: Arguments for `RTDCBase` **kwargs: Keyword arguments for `RTDCBase` Attributes ---------- path: pathlib.Path Path to the experimental dataset (main .tdms file) """ if not NPTDMS_AVAILABLE: raise ModuleNotFoundError( "Package `nptdms` required for TDMS format!") # Initialize RTDCBase super(RTDC_TDMS, self).__init__(*args, **kwargs) tdms_path = pathlib.Path(tdms_path) # Events is a simple dictionary self._events = {} self._hash = None self.path = tdms_path self.title = get_project_name_from_path(tdms_path, append_mx=True) # tdms-related convenience properties self._fdir = tdms_path.parent self._mid ="_")[0] self._init_data_with_tdms(tdms_path) # Add additional features # event images self._events["image"] = ImageColumn(self) # event contours (requires image) self._events["contour"] = ContourColumn(self) # event masks (requires contour) self._events["mask"] = MaskColumn(self) # event traces self._events["trace"] = TraceColumn(self) def __contains__(self, key): ct = False if key in ["contour", "image", "mask", "trace"]: # Take into account special cases of the tdms file format: # tdms features "image", "trace", "contour", and "mask" # evaluate to True (len()!=0) if the data exist on disk if key in self._events and self._events[key]: ct = True else: ct = super(RTDC_TDMS, self).__contains__(key) return ct @staticmethod def extract_tdms_config(path, features_available=None, ret_source_files=False, ignore_missing=False): """Extract as much metadata as possible for a .tdms dataset Parameters ---------- path: str or pathlib.Path A path representing the dataset. This could be either a .tdms file or an .avi file. The only thing important here is the prefix (before the underscore "_") which determines the location of the camera.ini and para.ini files. features_available: list of str List of features known to be available for this dataset. Used for defining e.g. fluorescnence metadata. ret_source_files: bool Return the list of files used to extract metadata from. ignore_missing: bool Nevermind when para.ini is missing. Returns ------- config: .Configuration The metadata Configuration instance source_paths: str List of metadata file paths, only returned when ret_source_files is True """ if features_available is None: features_available = [] mid ="_")[0] # Set up configuration config_paths = [] para_ini = path.with_name(mid + "_para.ini") if not para_ini.exists() and not ignore_missing: raise IncompleteTDMSFileFormatError(f"Could not find {para_ini}!") for pp in [para_ini, path.with_name(mid + "_camera.ini"), path.with_name(mid + "_SoftwareSettings.ini")]: if pp.exists(): config_paths.append(pp) tdms_config = Configuration(files=config_paths, disable_checks=True) dclab_config = Configuration() source_files = copy.copy(config_paths) for cfgii in [naming.configmap, naming.config_map_set]: for section in cfgii: for pname in cfgii[section]: meta = cfgii[section][pname] convfunc = dfn.get_config_value_func(section, pname) if isinstance(meta, tuple): osec, opar = meta if osec in tdms_config and opar in tdms_config[osec]: val = tdms_config[osec].pop(opar) dclab_config[section][pname] = convfunc(val) else: dclab_config[section][pname] = convfunc(meta) # Additional information from log file rtfdc_log = path.with_name(mid + "_log.ini") if rtfdc_log.exists(): source_files.append(rtfdc_log) with"r", errors="replace") as fd: loglines = fd.readlines() for line in loglines: if line.startswith("[EVENT LOG]"): sv = line.split("]")[1].strip() if sv: dclab_config["setup"]["software version"] = sv rtfdc_parm = path.with_name("parameters.txt") if rtfdc_parm.exists(): source_files.append(rtfdc_parm) with"r", errors="replace") as fd: parlines = fd.readlines() p1 = None p2 = None p3 = None for line in parlines: if line.startswith("pulse_led"): fdur = float(line.split()[1]) dclab_config["imaging"]["flash duration"] = fdur elif line.startswith("numberofchannels"): nc = int(line.split()[1]) dclab_config["fluorescence"]["channel count"] = nc elif line.startswith("laser488"): p1 = float(line.split()[1]) dclab_config["fluorescence"]["laser 1 lambda"] = 488 dclab_config["fluorescence"]["laser 1 power"] = p1 elif line.startswith("laser561"): p2 = float(line.split()[1]) dclab_config["fluorescence"]["laser 2 lambda"] = 561 dclab_config["fluorescence"]["laser 2 power"] = p2 elif line.startswith("laser640"): p3 = float(line.split()[1]) dclab_config["fluorescence"]["laser 3 lambda"] = 640 dclab_config["fluorescence"]["laser 3 power"] = p3 elif line.startswith("samplerate"): sr = int(float(line.split()[1])) dclab_config["fluorescence"]["sample rate"] = sr elif line.startswith("samplesperframe"): spe = int(line.split()[1]) dclab_config["fluorescence"]["samples per event"] = spe elif line.startswith("Vmin"): vmin = float(line.split()[1]) dclab_config["fluorescence"]["signal min"] = vmin elif line.startswith("Vmax"): vmax = float(line.split()[1]) dclab_config["fluorescence"]["signal max"] = vmax elif line.startswith("median_pmt"): mfs = int(line.split()[1]) dclab_config["fluorescence"]["trace median"] = mfs # Add generic channel names (independent of lasers) for ii in range(1, 4): chn = "channel {} name".format(ii) fln = "fl{}_max".format(ii) if (fln in features_available and chn not in dclab_config["fluorescence"]): dclab_config["fluorescence"][chn] = "FL{}".format(ii) lc = bool(p1) + bool(p2) + bool(p3) dclab_config["fluorescence"]["laser count"] = lc li = (p1 is not None) + (p2 is not None) + (p3 is not None) dclab_config["fluorescence"]["lasers installed"] = li dclab_config["fluorescence"]["channels installed"] = 3 # fluorescence if ("fluorescence" in dclab_config or "fl1_max" in features_available or "fl2_max" in features_available or "fl3_max" in features_available): # hardware-defined (always the same) dclab_config["fluorescence"].setdefault("bit depth", 16) dclab_config["fluorescence"].setdefault("laser 1 lambda", 488.) dclab_config["fluorescence"].setdefault("laser 2 lambda", 561.) dclab_config["fluorescence"].setdefault("laser 3 lambda", 640.) # Additional information from commented-out log-file (manual) if para_ini.exists(): text = para_ini.read_text(errors="replace").split("\n") lns = [s[1:].strip() for s in text if s.startswith("#")] if lns and lns[0] == "[FLUOR]": if ("software version" not in dclab_config["setup"] and lns[1].startswith("fRTDC")): dclab_config["setup"]["software version"] = lns[1] for ll in lns[2:]: if ("sample rate" not in dclab_config["fluorescence"] and ll.startswith("Samplerate")): val = int(float(ll.split("=")[1])) dclab_config["fluorescence"]["sample rate"] = val elif ("signal min" not in dclab_config["fluorescence"] and ll.startswith("ADCmin")): val = float(ll.split("=")[1]) dclab_config["fluorescence"]["signal min"] = val elif ("signal max" not in dclab_config["fluorescence"] and ll.startswith("ADCmax")): val = float(ll.split("=")[1]) dclab_config["fluorescence"]["signal max"] = val if dclab_config["imaging"].get("frame rate") == 0: dclab_config["imaging"].pop("frame rate") if dclab_config["setup"].get("flow rate") == 0: dclab_config["setup"].pop("flow rate") if "channel width" not in dclab_config["setup"]: if "channel width" in tdms_config.get("general", {}): channel_width = tdms_config["general"]["channel width"] elif dclab_config["setup"].get("flow rate", 0) < 0.16: channel_width = 20. else: channel_width = 30. dclab_config["setup"]["channel width"] = channel_width if "sample" not in dclab_config["experiment"]: # Measured sample or user-defined reference sample = get_project_name_from_path(path) dclab_config["experiment"]["sample"] = sample # imaging dclab_config["imaging"].setdefault("pixel size", 0.34) # medium convention for CellCarrierB if ("medium" in dclab_config["setup"] and dclab_config["setup"]["medium"].lower() == "cellcarrier b"): dclab_config["setup"]["medium"] = "CellCarrierB" # replace "+" with "," if "module composition" in dclab_config["setup"]: mc = dclab_config["setup"]["module composition"] if mc.count("+"): mc2 = ", ".join([m.strip() for m in mc.split("+")]) dclab_config["setup"]["module composition"] = mc2 dclab_config["imaging"].setdefault("flash device", "LED") dclab_config["imaging"].setdefault("flash duration", 2.0) dclab_config["imaging"].setdefault("roi position x", 0) dclab_config["imaging"].setdefault("roi position y", 0) if mid.startswith("m") and mid[1] in "0123456789": run_index = int(mid.strip("mM")) else: run_index = 1 dclab_config["experiment"].setdefault("run index", run_index) if ret_source_files: return dclab_config, source_files else: return dclab_config def _init_data_with_tdms(self, tdms_filename): """Initializes the current RT-DC dataset with a tdms file. """ tdms_file = nptdms.TdmsFile(str(tdms_filename)) # time is always there table = "Cell Track" # Edit naming.dclab2tdms to add features for arg in naming.tdms2dclab: try: data = tdms_file[table][arg].data except KeyError: pass else: if data is None or len(data) == 0: # Ignore empty features. npTDMS treats empty # features in the following way: # - in nptdms 0.8.2, `data` is `None` # - in nptdms 0.9.0, `data` is an array of length 0 continue self._events[naming.tdms2dclab[arg]] = data if len(self._events) == 0: raise IncompleteTDMSFileFormatError( "No usable feature data found in '{}'!".format(tdms_filename)) self.config, config_paths = self.extract_tdms_config( self.path, features_available=sorted(self._events.keys()), ret_source_files=True) self._complete_config_with_data() # Load log files log_files = config_paths for name in [self._mid + "_events.txt", self._mid + "_log.ini", self._mid + "_SoftwareSettings.ini", "FG_Config.mcf", "parameters.txt"]: pl = self.path.with_name(name) if pl.exists(): log_files.append(pl) for pp in sorted(set(log_files)): # avoid duplicates with"r", errors="replace") as f: cfg = [s.strip() for s in f.readlines()] self.logs[] = cfg def _complete_config_with_data(self): # measurement start time tse = self.path.stat().st_mtime if "time" in self: # correct for duration of experiment tse -= self["time"][-1] loct = time.localtime(tse) # Start time of measurement ('HH:MM:SS') timestr = time.strftime("%H:%M:%S", loct) self.config["experiment"].setdefault("time", timestr) # Date of measurement ('YYYY-MM-DD') datestr = time.strftime("%Y-%m-%d", loct) self.config["experiment"].setdefault("date", datestr) # Number of recorded events self.config["experiment"].setdefault("event count", len(self)) # fmt_tdms self.config["fmt_tdms"].setdefault("video frame offset", 1) # setup (compatibility to old tdms formats) self.config["setup"].setdefault("flow rate", np.nan) @staticmethod def can_open(h5path): """Check whether a given file is in the .tdms file format""" return pathlib.Path(h5path).suffix == ".tdms" @property def hash(self): """Hash value based on file name and .ini file content""" if self._hash is None: # Only hash _camera.ini and _para.ini fsh = [self.path.with_name(self._mid + "_camera.ini"), self.path.with_name(self._mid + "_para.ini")] tohash = [hashfile(f) for f in fsh] tohash.append( # Hash a maximum of ~1MB of the tdms file tohash.append(hashfile(self.path, blocksize=65536, count=20)) self._hash = hashobj(tohash) return self._hash def close(self): super(RTDC_TDMS, self).close() del self._events["image"]._image_data
[docs] def get_project_name_from_path(path, append_mx=False): """Get the project name from a path. For a path "/home/peter/hans/HLC12398/online/M1_13.tdms" or For a path "/home/peter/hans/HLC12398/online/data/M1_13.tdms" or without the ".tdms" file, this will return always "HLC12398". Parameters ---------- path: str or pathlib.Path path to tdms file append_mx: bool append measurement number, e.g. "M1" """ path = pathlib.Path(path) if path.suffix == ".tdms": dirn = path.parent mx ="_")[0] elif path.is_dir(): dirn = path mx = "" else: dirn = path.parent mx = "" project = "" if mx: # check para.ini para = dirn / (mx + "_para.ini") if para.exists(): with"r", errors="replace") as fd: lines = fd.readlines() for line in lines: if line.startswith("Sample Name ="): project = line.split("=")[1].strip() break if not project: # check if the directory contains data or is online root1, trail1 = dirn.parent, root2, trail2 = root1.parent, trail3 = if trail1.lower() in ["online", "offline"]: # /home/peter/hans/HLC12398/online/ project = trail2 elif (trail1.lower() == "data" and trail2.lower() in ["online", "offline"]): # this is olis new folder sctructure # /home/peter/hans/HLC12398/online/data/ project = trail3 else: project = trail1 if append_mx: project += " - " + mx return project
[docs] def get_tdms_files(directory): """Recursively find projects based on '.tdms' file endings Searches the `directory` recursively and return a sorted list of all found '.tdms' project files, except fluorescence data trace files which end with `_traces.tdms`. """ path = pathlib.Path(directory).resolve() # get all tdms files tdmslist = [r for r in path.rglob("*.tdms") if r.is_file()] # exclude traces files tdmslist = [r for r in tdmslist if not"_traces.tdms")] return sorted(tdmslist)