diff options
| -rw-r--r-- | scripts/qc_on_rqtl2_bundle2.py | 195 |
1 files changed, 138 insertions, 57 deletions
diff --git a/scripts/qc_on_rqtl2_bundle2.py b/scripts/qc_on_rqtl2_bundle2.py index f39689e..9136243 100644 --- a/scripts/qc_on_rqtl2_bundle2.py +++ b/scripts/qc_on_rqtl2_bundle2.py @@ -6,6 +6,7 @@ from time import sleep from pathlib import Path from zipfile import ZipFile from argparse import Namespace +from datetime import timedelta import multiprocessing as mproc from functools import reduce, partial from logging import Logger, getLogger, StreamHandler @@ -99,25 +100,30 @@ def open_file(file_: Path) -> Iterator: yield line -def check_markers(filename: str, row: tuple[str, ...]) -> tuple[rqfe.InvalidValue]: +def check_markers( + filename: str, + row: tuple[str, ...], + save_error: lambda val: val +) -> tuple[rqfe.InvalidValue]: """Check that the markers are okay""" errors = tuple() counts = {} for marker in row: counts = {**counts, marker: counts.get(marker, 0) + 1} if marker is None or marker == "": - errors = errors + (rqfe.InvalidValue( + errors = errors + (save_error(rqfe.InvalidValue( filename, "markers" "-", marker, - "A marker MUST be a valid value."),) + "A marker MUST be a valid value.")),) return errors + tuple( - rqfe.InvalidValue(filename, - "markers", - key, - f"Marker '{key}' was repeated {value} times") + save_error(rqfe.InvalidValue( + filename, + "markers", + key, + f"Marker '{key}' was repeated {value} times")) for key,value in counts.items() if value > 1) @@ -125,27 +131,28 @@ def check_geno_line( filename: str, headers: tuple[str, ...], row: tuple[Union[str, None]], - cdata: dict + cdata: dict, + save_error: lambda val: val ) -> tuple[rqfe.InvalidValue]: """Check that the geno line is correct.""" errors = tuple() # Verify that line has same number of columns as headers if len(headers) != len(row): - errors = errors + (rqfe.InvalidValue( + errors = errors + (save_error(rqfe.InvalidValue( filename, headers[0], row[0], row[0], - "Every line MUST have the same number of columns."),) + "Every line MUST have the same number of columns.")),) # First column is the individuals/cases/samples if not bool(row[0]): - errors = errors + (rqfe.InvalidValue( + errors = errors + (save_error(rqfe.InvalidValue( filename, headers[0], row[0], row[0], - "The sample/case MUST be a valid value."),) + "The sample/case MUST be a valid value.")),) def __process_value__(val): if val in cdata["na.strings"]: @@ -160,15 +167,34 @@ def check_geno_line( cellvalue is not None and cellvalue not in genocode.keys() ): - errors = errors + (rqfe.InvalidValue( + errors = errors + (save_error(rqfe.InvalidValue( filename, row[0], coltitle, cellvalue, f"Value '{cellvalue}' is invalid. Expected one of " - f"'{', '.join(genocode.keys())}'.")) + f"'{', '.join(genocode.keys())}'.")),) return errors +def push_file_error_to_redis(rconn: Redis, key: str, error: InvalidValue) -> InvalidValue: + """Push the file error to redis a json string + + Parameters + ---------- + rconn: Connection to redis + key: The name of the list where we push the errors + error: The file error to save + + Returns + ------- + Returns the file error it saved + """ + if bool(error): + rconn.rpush(key, json.dumps(error._asdict())) + return error + + def file_errors_and_details( + redisargs: dict[str, str], file_: Path, filetype: str, cdata: dict, @@ -182,55 +208,105 @@ def file_errors_and_details( if cdata[f"{filetype}_transposed"]: rqtl2.transpose_csv_with_rename(file_, linesplitterfn, linejoinerfn) - for lineno, line in enumerate(open_file(file_), start=1): - row = linesplitterfn(line) - if lineno == 1: - headers = tuple(row) + with Redis.from_url(redisargs["redisuri"], decode_responses=True) as rconn: + save_error_fn = partial(push_file_error_to_redis, + rconn, + error_list_name(filetype, file_.name)) + for lineno, line in enumerate(open_file(file_), start=1): + row = linesplitterfn(line) + if lineno == 1: + headers = tuple(row) + errors = errors + reduce( + lambda errs, fnct: errs + fnct( + file_.name, row[1:], save_error_fn), + headercheckers, + tuple()) + continue + errors = errors + reduce( - lambda errs, fnct: errs + fnct(file_.name, row[1:]), - headercheckers, + lambda errs, fnct: errs + fnct( + file_.name, headers, row, cdata, save_error_fn), + bodycheckers, tuple()) - continue - - errors = errors + reduce( - lambda errs, fnct: errs + fnct(file_.name, headers, row, cdata), - bodycheckers, - tuple()) - - return { - "filename": file_.name, - "filesize": os.stat(file_).st_size, - "linecount": lineno, - # TOD0: Maybe put the errors in a redis list and return the name of the - # redis list rather than the errors. Maybe also replace the errors - # key with a flag e.g. "errors-found": True/False - "errors": errors - } + + filedetails = { + "filename": file_.name, + "filesize": os.stat(file_).st_size, + "linecount": lineno + } + rconn.hset(redisargs["fqjobid"], + f"file-details:{filetype}:{file_.name}", + json.dumps(filedetails)) + return {**filedetails, "errors": errors} + + +def error_list_name(filetype: str, filename: str): + """Compute the name of the list where the errors will be pushed. + + Parameters + ---------- + filetype: The type of file. One of `r_qtl.r_qtl2.FILE_TYPES` + filename: The name of the file. + """ + return f"errors:{filetype}:{filename}" def check_for_geno_errors( + redisargs: dict[str, str], extractdir: Path, cdata: dict, linesplitterfn: Callable[[str], tuple[Union[str, None]]], linejoinerfn: Callable[[tuple[Union[str, None], ...]], str], - logger: Logger) -> bool: + logger: Logger +) -> bool: """Check for errors in genotype files.""" - if "geno" in cdata: - genofiles = tuple(extractdir.joinpath(fname) for fname in cdata["geno"]) - ## Run code below in multiprocessing once you verify it works. - gerrs = tuple(file_errors_and_details( - file_, - filetype="geno", - cdata=cdata, - linesplitterfn=linesplitterfn, - linejoinerfn=linejoinerfn, - headercheckers=(check_markers,), - bodycheckers=(check_geno_line,)) for file_ in genofiles) - # TOD0: Add the errors to redis - if len(gerrs) > 0: - logger.error("At least one of the 'geno' files has (an) error(s).") - return True - logger.info("No error(s) found in any of the 'geno' files.") + if "geno" in cdata or "founder_geno" in cdata: + genofiles = tuple( + extractdir.joinpath(fname) for fname in cdata.get("geno", [])) + fgenofiles = tuple( + extractdir.joinpath(fname) for fname in cdata.get("founder_geno", [])) + allgenofiles = genofiles + fgenofiles + with Redis.from_url(redisargs["redisuri"], decode_responses=True) as rconn: + error_list_names = [ + error_list_name("geno", file_.name) for file_ in allgenofiles] + for list_name in error_list_names: + rconn.delete(list_name) + rconn.hset( + redisargs["fqjobid"], + "geno-errors-lists", + json.dumps(error_list_names)) + processes = [ + mproc.Process(target=file_errors_and_details, + args=( + redisargs, + file_, + ftype, + cdata, + linesplitterfn, + linejoinerfn, + (check_markers,), + (check_geno_line,)) + ) + for ftype, file_ in ( + tuple(("geno", file_) for file_ in genofiles) + + tuple(("founder_geno", file_) for file_ in fgenofiles)) + ] + for process in processes: + process.start() + # Set expiry for any created error lists + for key in error_list_names: + rconn.expire(name=key, + time=timedelta(seconds=redisargs["redisexpiry"])) + + # TOD0: Add the errors to redis + if any(rconn.llen(errlst) > 0 for errlst in error_list_names): + logger.error("At least one of the 'geno' files has (an) error(s).") + return True + logger.info("No error(s) found in any of the 'geno' files.") + + else: + logger.info("No 'geno' files to check.") + return False @@ -249,9 +325,8 @@ def check_for_geno_errors( # pass -def run_qc(rconn: Redis, args: Namespace, logger: Logger) -> int: +def run_qc(rconn: Redis, args: Namespace, fqjobid: str, logger: Logger) -> int: """Run quality control checks on R/qtl2 bundles.""" - fqjobid = jobs.job_key(args.redisprefix, args.jobid) thejob = parse_job(rconn, args.redisprefix, args.jobid) print(f"THE JOB =================> {thejob}") jobmeta = thejob["job-metadata"] @@ -264,9 +339,15 @@ def run_qc(rconn: Redis, args: Namespace, logger: Logger) -> int: cdata = rqtl2.control_data(extractdir) splitter = build_line_splitter(cdata) joiner = build_line_joiner(cdata) + + redisargs = { + "fqjobid": fqjobid, + "redisuri": args.redisuri, + "redisexpiry": args.redisexpiry + } check_for_missing_files(rconn, fqjobid, extractdir, logger) - check_for_geno_errors(extractdir, cdata, splitter, joiner, logger) # check_for_pheno_errors(...) + check_for_geno_errors(redisargs, extractdir, cdata, splitter, joiner, logger) # check_for_phenose_errors(...) # check_for_phenocovar_errors(...) ### END: The quality control checks ### @@ -299,7 +380,7 @@ if __name__ == "__main__": rconn, fqjobid, f"{fqjobid}:log-messages", args.redisexpiry)) - exitcode = run_qc(rconn, args, logger) + exitcode = run_qc(rconn, args, fqjobid, logger) rconn.hset( jobs.job_key(args.redisprefix, args.jobid), "exitcode", exitcode) return exitcode |