Source code for allensdk.brain_observatory.behavior.data_objects.metadata.subject_metadata.driver_line

from typing import List, Optional

from pynwb import NWBFile

from allensdk.core import DataObject
from allensdk.core import \
    JsonReadableInterface, LimsReadableInterface, NwbReadableInterface
from allensdk.internal.api import PostgresQueryMixin, \
    OneOrMoreResultExpectedError


class DriverLine(DataObject, LimsReadableInterface, JsonReadableInterface,
                 NwbReadableInterface):
    """the genotype name(s) of the driver line(s)"""
    def __init__(self, driver_line: Optional[List[str]]):
        super().__init__(name="driver_line", value=driver_line)

    @classmethod
    def from_json(cls, dict_repr: dict) -> "DriverLine":
        return cls(driver_line=dict_repr['driver_line'])

    @classmethod
    def from_lims(cls, behavior_session_id: int,
                  lims_db: PostgresQueryMixin,
                  allow_none: bool = True) -> "DriverLine":
        """
        Parameters
        ----------
        behavior_session_id: int

        lims_db: PostgresQueryMixin

        allow_none: bool
            if True, allow None as a valid result

        Returns
        -------
        An instance of DriverLine with the value resulting
        from a query to the LIMS database
        """

        query = f"""
            SELECT g.name AS driver_line
            FROM behavior_sessions bs
            JOIN donors d ON bs.donor_id=d.id
            JOIN donors_genotypes dg ON dg.donor_id=d.id
            JOIN genotypes g ON g.id=dg.genotype_id
            JOIN genotype_types gt
                ON gt.id=g.genotype_type_id AND gt.name = 'driver'
            WHERE bs.id={behavior_session_id};
        """
        result = lims_db.fetchall(query)
        if result is None or len(result) < 1:
            if allow_none:
                return cls(driver_line=None)

            raise OneOrMoreResultExpectedError(
                f"Expected one or more, but received: '{result}' "
                f"from query:\n'{query}'")

        driver_line = sorted(result)
        return cls(driver_line=driver_line)

    @classmethod
    def from_nwb(cls, nwbfile: NWBFile) -> "DriverLine":
        if nwbfile.subject.driver_line is None:
            return cls(driver_line=None)
        driver_line = sorted(list(nwbfile.subject.driver_line))
        return cls(driver_line=driver_line)