Skip to main content
`Ctrl`+`K`

#

# 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