Aluode/PerceptionLabPortable

1# Authors: The MNE-Python contributors.2# License: BSD-3-Clause3# Copyright the MNE-Python contributors.4 5import re6 7import numpy as np8 9from ..._fiff.pick import _picks_to_idx, pick_types10from ...utils import _check_option, _validate_type, fill_doc11 12# Standardized fNIRS channel name regexs13_S_D_F_RE = re.compile(r"S(\d+)_D(\d+) (\d+\.?\d*)")14_S_D_H_RE = re.compile(r"S(\d+)_D(\d+) (\w+)")15 16 17@fill_doc18def source_detector_distances(info, picks=None):19    r"""Determine the distance between NIRS source and detectors.20 21    Parameters22    ----------23    %(info_not_none)s24    %(picks_all_data)s25 26    Returns27    -------28    dists : array of float29        Array containing distances in meters.30        Of shape equal to number of channels, or shape of picks if supplied.31    """32    return np.array(33        [34            np.linalg.norm(35                np.diff(info["chs"][pick]["loc"][3:9].reshape(2, 3), axis=0)[0]36            )37            for pick in _picks_to_idx(info, picks, exclude=[])38        ],39        float,40    )41 42 43@fill_doc44def short_channels(info, threshold=0.01):45    r"""Determine which NIRS channels are short.46 47    Channels with a source to detector distance of less than48    ``threshold`` are reported as short. The default threshold is 0.01 m.49 50    Parameters51    ----------52    %(info_not_none)s53    threshold : float54        The threshold distance for what is considered short in meters.55 56    Returns57    -------58    short : array of bool59        Array indicating which channels are short.60        Of shape equal to number of channels.61    """62    return source_detector_distances(info) < threshold63 64 65def _channel_frequencies(info):66    """Return the light frequency for each channel."""67    # Only valid for fNIRS data before conversion to haemoglobin68    picks = _picks_to_idx(69        info, ["fnirs_cw_amplitude", "fnirs_od"], exclude=[], allow_empty=True70    )71    freqs = list()72    for pick in picks:73        freqs.append(round(float(_S_D_F_RE.match(info["ch_names"][pick]).groups()[2])))74    return np.array(freqs, int)75 76 77def _channel_chromophore(info):78    """Return the chromophore of each channel."""79    # Only valid for fNIRS data after conversion to haemoglobin80    picks = _picks_to_idx(info, ["hbo", "hbr"], exclude=[], allow_empty=True)81    chroma = []82    for ii in picks:83        chroma.append(info["ch_names"][ii].split(" ")[1])84    return chroma85 86 87def _check_channels_ordered(info, pair_vals, *, throw_errors=True, check_bads=True):88    """Check channels follow expected fNIRS format.89 90    If the channels are correctly ordered then an array of valid picks91    will be returned.92 93    If throw_errors is True then any errors in fNIRS formatting will be94    thrown to inform the user. If throw_errors is False then an empty array95    will be returned if the channels are not sufficiently formatted.96    """97    # Every second channel should be same SD pair98    # and have the specified light frequencies.99 100    # All wavelength based fNIRS data.101    picks_wave = _picks_to_idx(102        info, ["fnirs_cw_amplitude", "fnirs_od"], exclude=[], allow_empty=True103    )104    # All chromophore fNIRS data105    picks_chroma = _picks_to_idx(info, ["hbo", "hbr"], exclude=[], allow_empty=True)106 107    if (len(picks_wave) > 0) & (len(picks_chroma) > 0):108        picks = _throw_or_return_empty(109            "MNE does not support a combination of amplitude, optical "110            "density, and haemoglobin data in the same raw structure.",111            throw_errors,112        )113 114    # All continuous wave fNIRS data115    if len(picks_wave):116        error_word = "frequencies"117        use_RE = _S_D_F_RE118        picks = picks_wave119    else:120        error_word = "chromophore"121        use_RE = _S_D_H_RE122        picks = picks_chroma123 124    pair_vals = np.array(pair_vals)125    if pair_vals.shape != (2,):126        raise ValueError(127            f"Exactly two {error_word} must exist in info, got {list(pair_vals)}"128        )129    # In principle we do not need to require that these be sorted --130    # all we need to do is change our sorted() below to make use of a131    # pair_vals.index(...) in a sort key -- but in practice we always want132    # (hbo, hbr) or (lower_freq, upper_freq) pairings, both of which will133    # work with a naive string sort, so let's just enforce sorted-ness here134    is_str = pair_vals.dtype.kind == "U"135    pair_vals = list(pair_vals)136    if is_str:137        if pair_vals != ["hbo", "hbr"]:138            raise ValueError(139                f'The {error_word} in info must be ["hbo", "hbr"], but got '140                f"{pair_vals} instead"141            )142    elif not np.array_equal(np.unique(pair_vals), pair_vals):143        raise ValueError(144            f"The {error_word} in info must be unique and sorted, but got "145            f"got {pair_vals} instead"146        )147 148    if len(picks) % 2 != 0:149        picks = _throw_or_return_empty(150            "NIRS channels not ordered correctly. An even number of NIRS "151            f"channels is required. {len(info.ch_names)} channels were"152            f"provided",153            throw_errors,154        )155 156    # Ensure wavelength info exists for waveform data157    all_freqs = [info["chs"][ii]["loc"][9] for ii in picks_wave]158    if np.any(np.isnan(all_freqs)):159        picks = _throw_or_return_empty(160            f"NIRS channels is missing wavelength information in the "161            f'info["chs"] structure. The encoded wavelengths are {all_freqs}.',162            throw_errors,163        )164 165    # Validate the channel naming scheme166    for pick in picks:167        ch_name_info = use_RE.match(info["chs"][pick]["ch_name"])168        if not bool(ch_name_info):169            picks = _throw_or_return_empty(170                "NIRS channels have specified naming conventions. "171                "The provided channel name can not be parsed: "172                f"{repr(info.ch_names[pick])}",173                throw_errors,174            )175            break176        value = ch_name_info.groups()[2]177        if len(picks_wave):178            value = value179        else:  # picks_chroma180            if value not in ["hbo", "hbr"]:181                picks = _throw_or_return_empty(182                    "NIRS channels have specified naming conventions."183                    "Chromophore data must be labeled either hbo or hbr. "184                    f"The failing channel is {info['chs'][pick]['ch_name']}",185                    throw_errors,186                )187                break188 189    # Reorder to be paired (naive sort okay here given validation above)190    picks = picks[np.argsort([info["ch_names"][pick] for pick in picks])]191 192    # Validate our paired ordering193    for ii, jj in zip(picks[::2], picks[1::2]):194        ch1_name = info["chs"][ii]["ch_name"]195        ch2_name = info["chs"][jj]["ch_name"]196        ch1_re = use_RE.match(ch1_name)197        ch2_re = use_RE.match(ch2_name)198        ch1_S, ch1_D, ch1_value = ch1_re.groups()[:3]199        ch2_S, ch2_D, ch2_value = ch2_re.groups()[:3]200        if len(picks_wave):201            ch1_value, ch2_value = float(ch1_value), float(ch2_value)202        if (203            (ch1_S != ch2_S)204            or (ch1_D != ch2_D)205            or (ch1_value != pair_vals[0])206            or (ch2_value != pair_vals[1])207        ):208            picks = _throw_or_return_empty(209                "NIRS channels not ordered correctly. Channels must be "210                "ordered as source detector pairs with alternating"211                f" {error_word} {pair_vals[0]} & {pair_vals[1]}, but got "212                f"S{ch1_S}_D{ch1_D} pair "213                f"{repr(ch1_name)} and {repr(ch2_name)}",214                throw_errors,215            )216            break217 218    if check_bads:219        for ii, jj in zip(picks[::2], picks[1::2]):220            want = [info.ch_names[ii], info.ch_names[jj]]221            got = list(set(info["bads"]).intersection(want))222            if len(got) == 1:223                raise RuntimeError(224                    f"NIRS bad labelling is not consistent, found {got} but "225                    f"needed {want}"226                )227    return picks228 229 230def _throw_or_return_empty(msg, throw_errors):231    if throw_errors:232        raise ValueError(msg)233    else:234        return []235 236 237def _validate_nirs_info(238    info,239    *,240    throw_errors=True,241    fnirs=None,242    which=None,243    check_bads=True,244    allow_empty=True,245):246    """Apply all checks to fNIRS info. Works on all continuous wave types."""247    _validate_type(fnirs, (None, str), "fnirs")248    kinds = dict(249        od="optical density",250        cw_amplitude="continuous wave",251        hb="chromophore",252    )253    _check_option("fnirs", fnirs, (None,) + tuple(kinds))254    if fnirs is not None:255        kind = kinds[fnirs]256        fnirs = ["hbo", "hbr"] if fnirs == "hb" else f"fnirs_{fnirs}"257        if not len(pick_types(info, fnirs=fnirs)):258            raise RuntimeError(259                f"{which} must operate on {kind} data, but none was found."260            )261    freqs = np.unique(_channel_frequencies(info))262    if freqs.size > 0:263        pair_vals = freqs264    else:265        pair_vals = np.unique(_channel_chromophore(info))266    out = _check_channels_ordered(267        info, pair_vals, throw_errors=throw_errors, check_bads=check_bads268    )269    return out270 271 272def _fnirs_spread_bads(info):273    """Spread bad labeling across fnirs channels."""274    # For an optode pair if any component (light frequency or chroma) is marked275    # as bad, then they all should be. This function will find any pairs marked276    # as bad and spread the bad marking to all components of the optode pair.277    picks = _validate_nirs_info(info, check_bads=False)278    new_bads = set(info["bads"])279    for ii, jj in zip(picks[::2], picks[1::2]):280        ch1_name, ch2_name = info.ch_names[ii], info.ch_names[jj]281        if ch1_name in new_bads:282            new_bads.add(ch2_name)283        elif ch2_name in new_bads:284            new_bads.add(ch1_name)285    info["bads"] = sorted(new_bads)286 287    return info288 289 290def _fnirs_optode_names(info):291    """Return list of unique optode names."""292    picks_wave = _picks_to_idx(293        info, ["fnirs_cw_amplitude", "fnirs_od"], exclude=[], allow_empty=True294    )295    picks_chroma = _picks_to_idx(info, ["hbo", "hbr"], exclude=[], allow_empty=True)296 297    if len(picks_wave) > 0:298        regex = _S_D_F_RE299    elif len(picks_chroma) > 0:300        regex = _S_D_H_RE301    else:302        return [], []303 304    sources = np.unique([int(regex.match(ch).groups()[0]) for ch in info.ch_names])305    detectors = np.unique([int(regex.match(ch).groups()[1]) for ch in info.ch_names])306 307    src_names = [f"S{s}" for s in sources]308    det_names = [f"D{d}" for d in detectors]309 310    return src_names, det_names311 312 313def _optode_position(info, optode):314    """Find the position of an optode."""315    idx = [optode in a for a in info.ch_names].index(True)316 317    if "S" in optode:318        loc_idx = range(3, 6)319    elif "D" in optode:320        loc_idx = range(6, 9)321 322    return info["chs"][idx]["loc"][loc_idx]323 324 325def _reorder_nirx(raw):326    # Maybe someday we should make this public like327    # mne.preprocessing.nirs.reorder_standard(raw, order='nirx')328    info = raw.info329    picks = pick_types(info, fnirs=True, exclude=[])330    prefixes = [info["ch_names"][pick].split()[0] for pick in picks]331    nirs_names = [info["ch_names"][pick] for pick in picks]332    nirs_sorted = sorted(333        nirs_names,334        key=lambda name: (prefixes.index(name.split()[0]), name.split(maxsplit=1)[1]),335    )336    raw.reorder_channels(nirs_sorted)337