diff options
Diffstat (limited to 'scripts')
| -rw-r--r-- | scripts/cli/validators.py | 8 | ||||
| -rw-r--r-- | scripts/genotypes/__init__.py | 1 | ||||
| -rw-r--r-- | scripts/genotypes/preprocess_csv_collect_info.py | 169 | ||||
| -rw-r--r-- | scripts/genotypes/types.py | 58 |
4 files changed, 236 insertions, 0 deletions
diff --git a/scripts/cli/validators.py b/scripts/cli/validators.py index 6d16e4c..dc198f5 100644 --- a/scripts/cli/validators.py +++ b/scripts/cli/validators.py @@ -2,6 +2,14 @@ from pathlib import Path +def file_exists(val: str) -> Path: + """Ensure the file path specified actually exists.""" + _file = Path(val) + if _file.exists() and _file.is_file(): + return _file + raise FileNotFoundError(f"Could not find the file '{_file}'.") + + def directory_exists(val: str) -> Path: """Check that directory path specified actually exists.""" _dir = Path(val).absolute() diff --git a/scripts/genotypes/__init__.py b/scripts/genotypes/__init__.py new file mode 100644 index 0000000..74fa7a5 --- /dev/null +++ b/scripts/genotypes/__init__.py @@ -0,0 +1 @@ +"""Scripts handling genotype-specific tasks.""" diff --git a/scripts/genotypes/preprocess_csv_collect_info.py b/scripts/genotypes/preprocess_csv_collect_info.py new file mode 100644 index 0000000..f5d3418 --- /dev/null +++ b/scripts/genotypes/preprocess_csv_collect_info.py @@ -0,0 +1,169 @@ +"""Preprocess the genotypes CSV files: Collect basic info.""" +import sys +import logging +from pathlib import Path +from typing import Iterator +from argparse import Namespace, ArgumentParser + +from gn_libs.cli import SysExit + +from scripts.cli.options import add_logging +from scripts.cli.logging import setup_logging +from scripts.cli.validators import file_exists + +from .types import FileStats, LineDetails, DotGenoFields, FieldsIdentity + + +logger = logging.getLogger(__name__) + + +def identify_columns( + fields: tuple[str, ...], + settings: dict[str, str] +) -> FieldsIdentity: + """Identify columns using settings provided by user""" + _id = {} + for field in fields: + if field.lower() in settings["markers"].lower(): + _id["marker_field"] = field + continue + if field.lower() in settings["chromosome"].lower(): + _id["chromosome_field"] = field + continue + if field.lower() in settings["cm"].lower(): + # linkage map: genetic distance + _id["centimorgan_field"] = field + continue + if field.lower() in settings["mb"].lower(): + # physical map: physical distance + _id["megabases_field"] = field + continue + + _id["samples_list"] = _id.get("samples_list") + (field,) + + return FieldsIdentity(**_id) + + +def process_transposed(line: tuple[str, ...]) -> LineDetails: + """Process a line in a transposed file.""" + raise NotImplementedError("Please implement this!") + + +def process_untransposed(line: tuple[str, ...]) -> LineDetails: + """Processed line in a non-transposed file.""" + raise NotImplementedError("Please implement this!") + + +def file_statistics(lines: Iterator[lines], fileconfigs: dict) -> FileStats: + """Compute file statistics from the lines in the file.""" + stats = {} + col_ids = tuple() + dotgeno = {} + headers: tuple[str, ...] = tuple() + genosymbols: set[str] = set() + errors: tuple() + for line in lines: + stats["total_lines"] = stats.get("total_lines", 0) + 1 + _line = line.strip() + if _line == "": + stats["blank_lines"] = stats.get("blank_lines", 0) + 1 + continue + + if _line.startswith(fileconfigs["comment.char"]): + stats["comment_lines"] = stats.get("comment_lines", 0) + 1 + continue + + if _line.startswith("@"): # custom GeneNetwork Settings + stats["dot_geno_lines"] = stats.get("dot_geno_lines", 0) + 1 + _fld = tuple(item.strip() for item in line.strip("@").split(":")) + dot_geno[_fld[0]] = dot_geno[_fld[1]] + continue + + stats["data_lines"] = stats.get("data_lines", 0) + 1 + fields: tuple[str, ...] = tuple( + fld.strip() for fld in line.split(fileconfigs["separator"])) + + # Basic QC: + # * Header row (normal) or column (transposed): + # - **MUST** have values: no missing values + # - **MUST** be strings + # * Mandatory Fields -- Fields that MUST have values: + # - Marker/Locus + # - Chromosome + # - cM: Genetic distance + # - Mb: Physical distance + if stats["data_lines"] == 1: # This is the first data line + if not fileconfigs["transposed"]: # File **IS NOT** transposed. + headers = fields # First line ***IS** headers' line! + continue + # ELSE: File **IS** transposed. + headers = headers + (fields[0]) + continue + + line_dets = ( + process_transposed( + fields, + na_strings=fileconfigs["transposed"], + ...) + if fileconfigs["transposed"] + else process_untransposed( + fields, + na_strings=fileconfigs["transposed"], + headers, + ...)) + + if fileconfigs["transposed"]: + headers = headers + (line_dets.header) + + genosymbols.update(line_dets.geno_symbols) + # TODO: do more processing + continue # end of line processing + + col_ids = identify_columns(fields) + return FileStats( + **stats, + **({"dot_geno_fields": DotGenoFields(**dotgeno)} + if bool(dotgeno) + else {}), + samples_list=col_ids.samples, + geno_symbols=tuple(genosymbols)) + + + +if __name__ == "__main__": + + def parse_args() -> Namespace: + """Parse the command-line arguments.""" + parser = add_logging( + ArgumentParser( + prog="preprocess-csv-collect-info", + description=( + "Pre-process the CSV file, with a focus on collecting " + "basic file information and statistics."))) + + parser.add_argument( + "csv-file", + metavar="CSV-FILE", + type=Path, #file_exists, + help="The CSV file to process.") + + parser.add_argument( + "--is-transposed", + action="store_true", + default=False, + help="Are the rows and columns in the CSV file flipped?") + + return parser.parse_args() + + + def main() -> SysExit: + try: + args = parse_args() + setup_logging(logger, args.log_level.upper(), tuple()) + logger.debug("CLI Arguments: %s", args) + return SysExit.OK + except FileNotFoundError as _fnf: + logger.error(", ".join(_fnf.args)) + return SysExit.FILENOTFOUND + + sys.exit(main().value) diff --git a/scripts/genotypes/types.py b/scripts/genotypes/types.py new file mode 100644 index 0000000..0ac335a --- /dev/null +++ b/scripts/genotypes/types.py @@ -0,0 +1,58 @@ +"""The datatypes this package will use.""" +from dataclasses import dataclass +from typing import Sequence, Optional + + +@dataclass +class DotGenoFields: + """Custom .geno file format values""" + name: str, + mat: str, + pat: str, + het: str, + type_: Optional[str] = None + unk: Optional[str] = None + + +@dataclass(frozen=True) +class FieldsIdentity: + """Data type for identity of the fields.""" + marker: str + chromosome: str + genetic_distance: str # cM + physical_distance: str # Mb + samples: Sequence[str] + + +@dataclass(frozen=True) +class LineDetails: + """Data type for processed line details""" + line_content: Sequence[Union[str, None]] + geno_symbols: Sequence[str] + line_header: Optional[str] = None + + +@dataclass(frozen=True) +class LineError: + """An error object for a line.""" + marker: str + field: str + value: Union[str, None] + message: str + + +@dataclass(frozen=True) +class FileStats: + """Custom file statistics.""" + data_lines: int + total_lines: int + geno_symbols: Sequence[str] + + blank_lines: int = 0 + comment_lines: int = 0 + dot_geno_lines: int = 0 + dot_geno_fields: Optional[DotGenoFields] = None + samples: Sequence[str] = field(default_factory=tuple) + samples_list: Sequence[str] = field(default_factory=tuple) + markers_list: Sequence[str] = field(default_factory=tuple) + errors: Sequence[LineError] = field(default_factory=tuple) |