Source code for ada.database_utils.fetch_files

import os
from pathlib import Path
from typing import Tuple




def uniqe_files(files: str | list[str]) -> list[str]:
    """Files are duplicated (2 recordings from the same hand for each subject.) Return only one of them.

    Args:
        files (str | list[str]): Unix-style path pattern or list of paths to files.

    Returns:
        list[str]: List of paths.
    """
    if type(files) is str:
        files = [str(e) for e in Path(os.path.dirname(files)).glob(os.path.basename(files))]
    
    used = set()
    out = []
    for file in files:
        name = file[:file.index("wrist")]
        if name not in used:
            used.add(name)
            out.append(file)

    return out





def hand_paths(files: str | list[str], hand: str = 'left') -> list[str]:
    """Return path to recordings from given hand.

    Args:
        files (str | list[str]): Unix-style path pattern or list of paths to files.
        hand (str, optional): Desired hand. Can be 'left' or 'right'. Defaults to 'left'.

    Returns:
        list[str]: List of paths.
    """
    if type(files) is str:
        files = [str(e) for e in Path(os.path.dirname(files)).glob(os.path.basename(files))]
    
    out = []
    for file in files:
        file_hand = file[file.index("wrist") - 3:file.index("wrist") - 2]
        if file_hand == 'h':
            file_hand = 'right'
        elif file_hand == 'f':
            file_hand = 'left'
        else:
            file_hand = 'none'
        
        if file_hand == hand:
            out.append(file)

    return out





def fetch_acti(files: str | list[str], tag_file: str) -> list[str]:
    """Find all actigraphic recordings corresponding to given obci.tag.

    Args:
        files (str | list[str]): Unix-style path pattern or list of paths to files.
        tag_file (str): Path to desired obci.tag file.

    Returns:
        list[str]: List of paths.
    """
    if type(files) is str:
        files = [str(e) for e in Path(os.path.dirname(files)).glob(os.path.basename(files))]
    
    name = os.path.basename(tag_file).split('.')[0]
    out = []
    for file in files:
        acti_name = os.path.basename(file).split('_')[0]
        if name == acti_name:
            out.append(file)

    return out





def fetch_from_tag(tag_file: str, acti_folder: str, xml_folder: str, hand: str = 'left') -> Tuple[str, str]:
    """For given obci.tag file fetch corrersponding actigraphic data and .xml with PSG metadata.

    Args:
        tag_file (str): Path to obci.tag file.
        acti_folder (str): Path to folder with actigraphic data.
        xml_folder (str): Path to folder with .xml metadadata corresponding to PSG recordings.
        hand (str, optional): Desired hand. Can be 'left' or 'right'. Defaults to 'left'.

    Raises:
        RuntimeError: Can't match actigraphic data or .xml to provided obci.tag.

    Returns:
        Tuple[str, str]: Path to .xml and to actigraphic data, respectively.
    """
    xml_path = os.path.join(xml_folder, os.path.basename(tag_file)[:-3] + 'xml')
    if not os.path.exists(xml_path):
        raise RuntimeError("No .xml file found!")
    
    acti_files = [str(e) for e in Path(acti_folder).glob("*.bin")]
    if not acti_files:
        acti_files = [str(e) for e in Path(acti_folder).glob("*.ada")]
    
    acti = fetch_acti(acti_files, tag_file)
    acti = uniqe_files(acti)
    acti = hand_paths(acti, hand)
    try:
        return xml_path, acti[0]
    except IndexError:
        raise RuntimeError("No actigraphic file found!")





def fetch_from_acti(acti_file: str, xml_folder: str, tag_folder: str, tag: bool = True) -> Tuple[str, str]:
    """For given actigraphic file find corresponding .xml and .tag files.

    Args:
        acti_file (str): Path to actigraphic data file.
        xml_folder (str): Path to folder with .xml metadata corresponding to PSG recordings.
        tag_folder (str): Path to folder with .tag tags corresponding to PSG recordings.
        tag (bool): If True, raise error if tag is not found. When False, path to tag might be meaningless. Defaults to True.

    Raises:
        RuntimeError: Can't match .tag or .xml files to provided actigraphic file.

    Returns:
        Tuple[str, str]: Path to .xml and .tag files corresponding to provided actigraphic file.
    """
    name = os.path.basename(acti_file).split('_')[0]
    xml_path = os.path.join(xml_folder, name + '.obci.xml')
    if not os.path.exists(xml_path):
        raise RuntimeError("No .xml file found!")
    
    tag_path = os.path.join(tag_folder, name + '.obci.tag')
    if not os.path.exists(tag_path) and tag:
        raise RuntimeError("No .tag file found!")
    
    return xml_path, tag_path