Initialise scripts for genotype uploads files' processing.HEADmain

3 files changed, 228 insertions, 0 deletions

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)