Source code for allensdk.brain_observatory.behavior.data_objects.metadata.ophys_experiment_metadata.multi_plane_metadata.imaging_plane_group

from typing import Optional

from pynwb import NWBFile

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


class ImagingPlaneGroup(DataObject, LimsReadableInterface,
                        JsonReadableInterface, NwbReadableInterface):
    def __init__(self, plane_group: int, plane_group_count: int):
        super().__init__(name='plane_group', value=None, is_value_self=True)
        self._plane_group = plane_group
        self._plane_group_count = plane_group_count

    @property
    def plane_group(self):
        return self._plane_group

    @property
    def plane_group_count(self):
        return self._plane_group_count

    @classmethod
    def from_lims(cls, ophys_experiment_id: int,
                  lims_db: PostgresQueryMixin) -> \
            Optional["ImagingPlaneGroup"]:
        """

        Parameters
        ----------
        ophys_experiment_id
        lims_db

        Returns
        -------
        ImagingPlaneGroup instance if ophys_experiment given by
            ophys_experiment_id is part of a plane group
        else None

        """
        query = f'''
            SELECT oe.id as ophys_experiment_id, pg.group_order AS plane_group
            FROM  ophys_experiments oe
            JOIN ophys_sessions os ON oe.ophys_session_id = os.id
            JOIN  ophys_imaging_plane_groups pg
                ON pg.id = oe.ophys_imaging_plane_group_id
            WHERE os.id = (
                SELECT oe.ophys_session_id
                FROM ophys_experiments oe
                WHERE oe.id = {ophys_experiment_id}
            )
        '''
        df = lims_db.select(query=query)
        if df.empty:
            return None
        df = df.set_index('ophys_experiment_id')
        plane_group = df.loc[ophys_experiment_id, 'plane_group']
        plane_group_count = df['plane_group'].nunique()
        return cls(plane_group=plane_group,
                   plane_group_count=plane_group_count)

    @classmethod
    def from_json(cls, dict_repr: dict) -> "ImagingPlaneGroup":
        plane_group = dict_repr['imaging_plane_group']
        plane_group_count = dict_repr['plane_group_count']
        return cls(plane_group=plane_group,
                   plane_group_count=plane_group_count)

    @classmethod
    def from_nwb(cls, nwbfile: NWBFile) -> "ImagingPlaneGroup":
        metadata = nwbfile.lab_meta_data['metadata']
        return cls(plane_group=metadata.imaging_plane_group,
                   plane_group_count=metadata.imaging_plane_group_count)