path: root/scripts/insert_data.py

"""Insert means/averages or standard-error data into the database."""
import sys
import string
import random
import argparse
from typing import Tuple
from functools import reduce

import MySQLdb as mdb
from redis import Redis
from MySQLdb.cursors import DictCursor

from quality_control.parsing import take
from quality_control.file_utils import open_file
from qc_app.db_utils import database_connection
from qc_app.check_connections import check_db, check_redis

def random_string(count: int = 10) -> str:
    """Generate a random, alphanumeric string."""
    return "".join(random.choices(
        string.digits + string.ascii_letters, k=count))

def translate_alias(heading):
    "Translate strain aliases into canonical names"
    translations = {"B6": "C57BL/6J", "D2": "DBA/2J"}
    return translations.get(heading, heading)

def read_file_headings(filepath) -> Tuple[str, ...]:
    "Get the file headings"
    with open_file(filepath) as input_file:
        for line_contents in input_file:
            headings = tuple(
                translate_alias(heading.strip())
                for heading in line_contents.split("\t"))
            break

    return headings

def read_file_contents(filepath):
    "Get the file contents"
    with open_file(filepath) as input_file:
        for line_number, line_contents in enumerate(input_file):
            if line_number == 0:
                continue
            if line_number > 0:
                yield tuple(
                    field.strip() for field in line_contents.split("\t"))

def strains_info(
        dbconn: mdb.Connection, strain_names: Tuple[str, ...],
        speciesid: int) -> dict:
    "Retrieve information for the strains"
    with dbconn.cursor(cursorclass=DictCursor) as cursor:
        query = (
            "SELECT * FROM Strain WHERE Name IN "
            f"({', '.join(['%s']*len(strain_names))}) "
            "AND SpeciesId = %s")
        cursor.execute(query, tuple(strain_names) + (speciesid,))
        return {strain["Name"]: strain for strain in cursor.fetchall()}

def read_datavalues(filepath, headings, strain_info):
    "Read data values from file"
    for row in (
            dict(zip(headings, line))
            for line in read_file_contents(filepath)):
        for sname in headings[1:]:
            yield {
                "ProbeSetId": int(row["ProbeSetID"]),
                "StrainId": strain_info[sname]["Id"],
                "DataValue": float(row[sname])
            }

def read_probesets(filepath, headings):
    """Read the ProbeSet names."""
    for row in (dict(zip(headings, line))
                for line in read_file_contents(filepath)):
        yield {"Name": int(row["ProbeSetID"])}

def last_data_id(dbconn: mdb.Connection) -> int:
    "Get the last id from the database"
    with dbconn.cursor() as cursor:
        cursor.execute("SELECT MAX(Id) FROM ProbeSetData")
        return int(cursor.fetchone()[0])

def check_strains(headings_strains, db_strains):
    "Check strains in headings exist in database"
    from_db = tuple(db_strains.keys())
    not_in_db = tuple(
        strain for strain in headings_strains if strain not in from_db)
    if len(not_in_db) == 0:
        return True

    str_not_in_db = "', '".join(not_in_db)
    print(
        (f"ERROR: The strain(s) '{str_not_in_db}' w(as|ere) not found in the "
         "database."),
        file=sys.stderr)
    sys.exit(1)

def annotationinfo(
        dbconn: mdb.Connection, platformid: int, datasetid: int) -> dict:
    "Get annotation information from the database."
    # This is somewhat slow. Look into optimising the behaviour
    def __organise_annotations__(accm, item):
        names_dict = (
            {**accm[0], item["Name"]: item} if bool(item["Name"]) else accm[0])
        targs_dict = (
            {**accm[1], item["TargetId"]: item}
            if bool(item["TargetId"]) else accm[1])
        return (names_dict, targs_dict)

    query = (
        "SELECT ProbeSet.Name, ProbeSet.ChipId, ProbeSet.TargetId, "
        "ProbeSetXRef.DataId, ProbeSetXRef.ProbeSetFreezeId "
        "FROM ProbeSet INNER JOIN ProbeSetXRef "
        "ON ProbeSet.Id=ProbeSetXRef.ProbeSetId "
        "WHERE ProbeSet.ChipId=%s AND ProbeSetXRef.ProbeSetFreezeId=%s")
    with dbconn.cursor(cursorclass=DictCursor) as cursor:
        cursor.execute(query, (platformid, datasetid))
        annot_dicts = reduce(# type: ignore[var-annotated]
            __organise_annotations__, cursor.fetchall(), ({}, {}))
        return {**annot_dicts[0], **annot_dicts[1]}
    return {}

def __format_query__(query, params):
    "Format the query for output"
    def __param_str__(param):
        return "', '".join(str(elt) for elt in param)
    idx = query.find("VALUES")
    idx = query.find("%")
    fields = tuple(
        elt.replace("%(", "").replace(")s", "").replace(")", "").strip()
        for elt in query[idx:-1].split(","))
    values = (tuple(param[field] for field in fields) for param in params)
    values_str = ", ".join(
        f"('{__param_str__(value_tup)}')" for value_tup in values)
    insert_str = query[:idx].replace(
        "INSERT INTO ", "INSERT INTO\n\t")
    return f"{insert_str}\nVALUES\n\t{values_str};"

def __xref_params__(dbconn: mdb.Connection, means: list[dict]) -> dict:
    """Process params for cross-reference table."""
    xref_names = tuple({mean["ProbeSetId"] for mean in means})
    with dbconn.cursor(cursorclass=DictCursor) as cursor:
        params_str = ", ".join(["%s"] * len(xref_names))
        cursor.execute(
            f"SELECT Name, Id FROM ProbeSet WHERE Name IN ({params_str})",
            xref_names)
        ids = {row["Name"]: row["Id"] for row in cursor.fetchall()}
        return tuple({
            **mean,
            "ProbeSetName": mean["ProbeSetId"],
            "ProbeSetId": ids[str(mean["ProbeSetId"])]
        } for mean in means)

def insert_means(
        filepath: str, speciesid: int, platform_id: int, datasetid: int,
        dbconn: mdb.Connection, rconn: Redis) -> int: # pylint: disable=[unused-argument]
    "Insert the means/averages data into the database"
    headings = read_file_headings(filepath)
    strains = strains_info(dbconn, headings[1:], speciesid)
    check_strains(headings[1:], strains),
    probeset_query = (
        "INSERT INTO ProbeSet(ChipId, Name) "
        "VALUES (%(ChipId)s, %(Name)s) ")
    means_query = (
        "INSERT INTO ProbeSetData "
        "VALUES(%(ProbeSetDataId)s, %(StrainId)s, %(DataValue)s)")
    xref_query = (
        "INSERT INTO ProbeSetXRef(ProbeSetFreezeId, ProbeSetId, DataId) "
        "VALUES(%(ProbeSetFreezeId)s, %(ProbeSetId)s, %(ProbeSetDataId)s)")
    the_means = ({
        "ProbeSetFreezeId": datasetid, "ProbeSetDataId": data_id,
        "ChipId": platform_id, **mean
    } for data_id, mean in enumerate(
        read_datavalues(filepath, headings, strains),
        start=(last_data_id(dbconn)+1)))
    the_probesets = ({
        **row,
        "Name": f"{row['Name']}::RAND_{random_string()}",
        "ChipId": platform_id
    } for row in read_probesets(filepath, headings))
    with dbconn.cursor(cursorclass=DictCursor) as cursor:
        while True:
            means = tuple(take(the_means, 10000))
            probeset_params = tuple(take(the_probesets, 10000))
            if not bool(means):
                break
            print(__format_query__(probeset_query, probeset_params))
            print()
            print(__format_query__(means_query, means))
            print()
            print(__format_query__(xref_query, means))
            cursor.executemany(probeset_query, probeset_params)
            cursor.executemany(means_query, means)
            cursor.executemany(xref_query, __xref_params__(dbconn, means))
    return 0

def insert_se(# pylint: disable = [too-many-arguments]
        filepath: str, speciesid: int, platformid: int, datasetid: int,
        dbconn: mdb.Connection, rconn: Redis) -> int: # pylint: disable=[unused-argument]
    "Insert the standard-error data into the database"
    headings = read_file_headings(filepath)
    strains = strains_info(dbconn, headings[1:], speciesid)
    check_strains(headings[1:], strains)
    se_query = (
        "INSERT INTO ProbeSetSE "
        "VALUES(%(DataId)s, %(StrainId)s, %(DataValue)s)")
    annotations = annotationinfo(dbconn, platformid, datasetid)
    if not bool(annotations):
        print(
            (f"ERROR: No annotations found for platform {platformid} and "
             f"dataset {datasetid}. Quiting!"),
            file=sys.stderr)
        return 1

    se_values = (
        {"DataId": annotations[str(item["ProbeSetId"])]["DataId"], **item}
        for item in read_datavalues(filepath, headings, strains))
    with dbconn.cursor(cursorclass=DictCursor) as cursor:
        while True:
            serrors = tuple(take(se_values, 1000))
            if not bool(serrors):
                break
            print(__format_query__(se_query, serrors))
            cursor.executemany(se_query, serrors)
    return 0

if __name__ == "__main__":
    def cli_args():
        "Compute the CLI arguments"
        parser = argparse.ArgumentParser(
            prog="InsertData", description=(
                "Script to insert data from an 'averages' file into the "
                "database."))
        parser.add_argument(
            "filetype", help="type of data to insert.",
            choices=("average", "standard-error"))
        parser.add_argument(
            "filepath", help="path to the file with the 'averages' data.")
        parser.add_argument(
            "speciesid", help="Identifier for the species in the database.",
            type=int)
        parser.add_argument(
            "platformid", help="Identifier for the platform in the database.",
            type=int)
        parser.add_argument(
            "datasetid", help="Identifier for the dataset in the database.",
            type=int)
        parser.add_argument(
            "database_uri",
            help="URL to be used to initialise the connection to the database")
        parser.add_argument(
            "redisuri",
            help="URL to initialise connection to redis",
            default="redis:///")

        args = parser.parse_args()
        check_db(args.database_uri)
        check_redis(args.redisuri)
        return args

    insert_fns = {
        "average": insert_means,
        "standard-error": insert_se
    }

    extract_args = {
        "average": lambda args, dbconn, rconn: (
            args.filepath, args.speciesid, args.platformid, args.datasetid, dbconn,
            rconn),
        "standard-error": lambda args, dbconn, rconn: (
            args.filepath, args.speciesid, args.platformid, args.datasetid,
            dbconn, rconn),
    }

    def main():
        "Main entry point"
        args = cli_args()
        with Redis.from_url(args.redisuri, decode_responses=True) as rconn:
            with database_connection(args.database_uri) as dbconn:
                return insert_fns[args.filetype](
                    *extract_args[args.filetype](args, dbconn, rconn))

        return 2

    sys.exit(main())