"""Run Quality Control checks on R/qtl2 bundle.""" import sys import json from time import sleep from pathlib import Path from zipfile import ZipFile from functools import partial from argparse import Namespace import multiprocessing as mproc from logging import Logger, getLogger, StreamHandler from typing import Union, Sequence, Callable, Iterator import MySQLdb as mdb from redis import Redis from gn_libs.mysqldb import database_connection from quality_control.errors import InvalidValue from quality_control.checks import decimal_points_error from uploader import jobs from uploader.check_connections import check_db, check_redis from r_qtl import r_qtl2 as rqtl2 from r_qtl import r_qtl2_qc as rqc from r_qtl import exceptions as rqe from r_qtl import fileerrors as rqfe from scripts.process_rqtl2_bundle import parse_job from scripts.redis_logger import setup_redis_logger from scripts.cli_parser import init_cli_parser, add_global_data_arguments def dict2tuple(dct: dict) -> tuple: """Utility to convert items in dicts to pairs of tuples.""" return tuple((key, val) for key,val in dct.items()) def add_to_errors(rconn: Redis, fqjobid: str, key: str, errors: Sequence[Union[InvalidValue, rqfe.MissingFile]]): """Add `errors` to a given list of errors""" errs = tuple(dict(item) for item in set( [dict2tuple(old) for old in json.loads(rconn.hget(fqjobid, key) or "[]")] + [dict2tuple({"type": type(error).__name__, **error._asdict()}) for error in errors])) rconn.hset(fqjobid, key, json.dumps(errs)) def qc_missing_files(rconn: Redis, fqjobid: str, zfile: ZipFile, logger: Logger) -> bool: """Run QC for files listed in control file that don't exist in bundle.""" logger.info("Checking for missing files…") missing = rqc.missing_files(zfile) add_to_errors(rconn, fqjobid, "errors-generic", tuple( rqfe.MissingFile( mfile[0], mfile[1], ( f"File '{mfile[1]}' is listed in the control file under " f"the '{mfile[0]}' key, but it does not actually exist in " "the bundle.")) for mfile in missing)) if len(missing) > 0: logger.error("Missing files in the bundle!") return True return False def compute_filesize(zfile: ZipFile, filetype: str) -> int: """Compute the total file size.""" cdata = rqtl2.control_data(zfile) if isinstance(cdata[filetype], str): return zfile.getinfo(cdata[filetype]).file_size return sum(zfile.getinfo(afile).file_size for afile in cdata[filetype]) def retrieve_errors_with_progress(rconn: Redis,#pylint: disable=[too-many-locals] fqjobid: str, zfile: ZipFile, filetype: str, checkers: tuple[Callable]) -> Iterator[Union[ InvalidValue, rqfe.MissingFile]]: """Filter the errors while also counting the number of lines in the file.""" assert filetype in rqtl2.FILE_TYPES, f"Invalid file type {filetype}." count = 0 checked = 0 cdata = rqtl2.control_data(zfile) rconn.hset(fqjobid, f"{filetype}-filesize", compute_filesize(zfile, filetype)) def __update_processed__(value): nonlocal checked checked = checked + len(value) rconn.hset(fqjobid, f"{filetype}-checked", checked) try:# pylint: disable=[too-many-nested-blocks] for lineno, row in enumerate( rqtl2.file_data(zfile, filetype, cdata), start=1): count = count + 1 for field, value in row.items(): if field == "id": __update_processed__(value) continue if value is not None: for checker in checkers: error = checker(lineno=lineno, field=field, value=value) if bool(error): yield error __update_processed__(value) rconn.hset(fqjobid, f"{filetype}-linecount", count) except rqe.MissingFileException: fname = cdata.get(filetype) yield rqfe.MissingFile(filetype, fname, ( f"The file '{fname}' does not exist in the bundle despite it being " f"listed under '{filetype}' in the control file.")) def qc_geno_errors(rconn, fqjobid, _dburi, _speciesid, zfile, logger) -> bool: """Check for errors in `geno` file(s).""" cdata = rqtl2.control_data(zfile) if "geno" in cdata: logger.info("Checking for errors in the 'geno' file…") gerrs = tuple(retrieve_errors_with_progress( rconn, fqjobid, zfile, "geno", (rqc.make_genocode_checker(cdata.get("genotypes", {}), "geno"),))) add_to_errors(rconn, fqjobid, "errors-generic", tuple( err for err in gerrs if isinstance(err, rqfe.MissingFile))) add_to_errors(rconn, fqjobid, "errors-geno", tuple( err for err in gerrs if not isinstance(err, rqfe.MissingFile))) if len(gerrs) > 0: logger.error("The 'geno' file has errors.") return True logger.info("No errors found in the 'geno' file.") return False def fetch_db_geno_samples(conn: mdb.Connection, speciesid: int) -> tuple[str, ...]: """Fetch samples/cases/individuals from the database.""" samples = set()# type: ignore[var-annotated] with conn.cursor() as cursor: cursor.execute("SELECT Name, Name2 from Strain WHERE SpeciesId=%s", (speciesid,)) rows = cursor.fetchall() or tuple() for row in rows: samples.update(tuple(row)) return tuple(item.strip() for item in samples if bool(item)) def check_pheno_samples( conn: mdb.Connection, speciesid: int, zipfilepath: Union[str, Path], logger: Logger ) -> tuple[Union[InvalidValue, rqfe.MissingFile], ...]: """Check that samples in 'pheno' file exist in geno file.""" cdata = rqtl2.read_control_file(zipfilepath) genosamples = tuple( sample for perfilesamples in ( rqtl2.load_samples(zipfilepath, member, cdata["geno_transposed"]) for member in cdata["geno"]) for sample in perfilesamples) + fetch_db_geno_samples(conn, speciesid) def __check_file__(member) -> tuple[InvalidValue, ...]: logger.info("Checking samples/cases in member file '%s' …", member) sampledata = rqtl2.load_samples( zipfilepath, member, cdata["pheno_transposed"]) errors: tuple[InvalidValue, ...] = tuple() for sample in sampledata: if sample not in genosamples: errors = errors + (InvalidValue( member, "-", "-", sample, f"The individual/case/sample '{sample}' in file " f"{member} does not exist in either, any of the 'geno' " "files provided in the bundle or the GeneNetwork database." ),) logger.info("Found %s missing samples in member file '%s'.", len(errors), member) return errors allerrors: tuple[Union[InvalidValue, rqfe.MissingFile], ...] = tuple() for afile in cdata["pheno"]: try: allerrors = allerrors + __check_file__(afile) except KeyError: allerrors = allerrors + (rqfe.MissingFile( "pheno", afile, (f"The file '{afile}' does not exist in the zipfile despite " "being listed in the control file.")),) return allerrors def qc_pheno_errors(# pylint: disable=[too-many-arguments] rconn, fqjobid, dburi, speciesid, zfile, logger) -> bool: """Check for errors in `pheno` file(s).""" cdata = rqtl2.control_data(zfile) if "pheno" in cdata: logger.info("Checking for errors in the 'pheno' file…") perrs = tuple()# type: ignore[var-annotated] with database_connection(dburi) as dbconn: perrs = check_pheno_samples( dbconn, speciesid, zfile.filename, logger) + tuple( retrieve_errors_with_progress( rconn,fqjobid, zfile, "pheno", (partial(decimal_points_error, filename="pheno", mini=3),))) add_to_errors(rconn, fqjobid, "errors-generic", tuple( err for err in perrs if isinstance(err, rqfe.MissingFile))) add_to_errors(rconn, fqjobid, "errors-pheno", tuple( err for err in perrs if not isinstance(err, rqfe.MissingFile))) if len(perrs) > 0: logger.error("The 'pheno' file has errors.") return True logger.info("No errors found in the 'pheno' file.") return False def qc_phenose_errors(# pylint: disable=[too-many-arguments] rconn, fqjobid, _dburi, _speciesid, zfile, logger) -> bool: """Check for errors in `phenose` file(s).""" cdata = rqtl2.control_data(zfile) if "phenose" in cdata: logger.info("Checking for errors in the 'phenose' file…") perrs = tuple(retrieve_errors_with_progress( rconn,fqjobid, zfile, "phenose", (partial(decimal_points_error, mini=6),))) add_to_errors(rconn, fqjobid, "errors-generic", tuple( err for err in perrs if isinstance(err, rqfe.MissingFile))) add_to_errors(rconn, fqjobid, "errors-phenose", tuple( err for err in perrs if not isinstance(err, rqfe.MissingFile))) if len(perrs) > 0: logger.error("The 'phenose' file has errors.") return True logger.info("No errors found in the 'phenose' file.") return False def qc_phenocovar_errors( _rconn, _fqjobid, _dburi, _speciesid, _zfile, _logger ) -> bool: """Check for errors in `phenocovar` file(s).""" return False def run_qc(rconn: Redis, args: Namespace, logger: Logger) -> int: """Run the QC programs.""" fqjobid = jobs.job_key(args.redisprefix, args.jobid) thejob = parse_job(rconn, args.redisprefix, args.jobid) jobmeta = thejob["job-metadata"] with ZipFile(jobmeta["rqtl2-bundle-file"], "r") as zfile: if qc_missing_files(rconn, fqjobid, zfile, logger): return 1 def with_zipfile(# pylint: disable=[too-many-arguments] rconn, fqjobid, dbconn, speciesid, filename, logger, func ): with ZipFile(filename, "r") as zfile: return func(rconn, fqjobid, dbconn, speciesid, zfile, logger) def buildargs(func): return (rconn, fqjobid, args.databaseuri, args.speciesid, jobmeta["rqtl2-bundle-file"], logger, func) processes = [ mproc.Process(target=with_zipfile, args=buildargs(qc_geno_errors,)), mproc.Process(target=with_zipfile, args=buildargs(qc_pheno_errors,)), mproc.Process(target=with_zipfile, args=buildargs(qc_phenose_errors,)), mproc.Process(target=with_zipfile, args=buildargs(qc_phenocovar_errors,)) ] for process in processes: process.start() while True: processes_running = any( (process.is_alive() for process in processes)) if not processes_running: break sleep(2) if any((process.exitcode for process in processes)): # at least one process failed for some reason... return 1 def __fetch_errors__(rkey: str) -> tuple: return tuple(json.loads(rconn.hget(fqjobid, rkey) or "[]")) return (1 if any(( bool(__fetch_errors__(key)) for key in ("errors-geno", "errors-pheno", "errors-phenos", "errors-phenocovar"))) else 0) if __name__ == "__main__": def main(): """Enter R/qtl2 bundle QC runner.""" args = add_global_data_arguments(init_cli_parser( "qc-on-rqtl2-bundle", "Run QC on R/qtl2 bundle.")).parse_args() check_redis(args.redisuri) check_db(args.databaseuri) logger = getLogger("qc-on-rqtl2-bundle") logger.addHandler(StreamHandler(stream=sys.stderr)) logger.setLevel("DEBUG") fqjobid = jobs.job_key(args.redisprefix, args.jobid) with Redis.from_url(args.redisuri, decode_responses=True) as rconn: logger.addHandler(setup_redis_logger( rconn, fqjobid, f"{fqjobid}:log-messages", args.redisexpiry)) exitcode = run_qc(rconn, args, logger) rconn.hset( jobs.job_key(args.redisprefix, args.jobid), "exitcode", exitcode) return exitcode sys.exit(main())