aboutsummaryrefslogtreecommitdiff
path: root/scripts/rqtl2/bundleutils.py
blob: 17faa7c30306373938be1b020214e5de5a47128c (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
"""Common utilities to operate in R/qtl2 bundles."""
from typing import Union, Callable

def build_line_splitter(cdata: dict) -> Callable[[str], tuple[Union[str, None], ...]]:
    """Build and return a function to use to split data in the files.

    Parameters
    ----------
    cdata: A dict holding the control information included with the R/qtl2
        bundle.

    Returns
    -------
    A function that takes a string and return a tuple of strings.
    """
    separator = cdata["sep"]
    na_strings = cdata["na.strings"]
    def __splitter__(line: str) -> tuple[Union[str, None], ...]:
        return tuple(
            item if item not in na_strings else None
            for item in
            (field.strip() for field in line.strip().split(separator)))
    return __splitter__


def build_line_joiner(cdata: dict) -> Callable[[tuple[Union[str, None], ...]], str]:
    """Build and return a function to use to split data in the files.

    Parameters
    ----------
    cdata: A dict holding the control information included with the R/qtl2
        bundle.

    Returns
    -------
    A function that takes a string and return a tuple of strings.
    """
    separator = cdata["sep"]
    na_strings = cdata["na.strings"]
    def __joiner__(row: tuple[Union[str, None], ...]) -> str:
        return separator.join(
            (na_strings[0] if item is None else item)
            for item in row)
    return __joiner__