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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
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)
|
