IRC-SPHERE/HyperStream

View on GitHub
hyperstream/plate/plate_manager.py

Summary

Maintainability
B
4 hrs
Test Coverage
# The MIT License (MIT)
# Copyright (c) 2014-2017 University of Bristol
# 
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# 
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# 
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
# DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
# OR OTHER DEALINGS IN THE SOFTWARE.

import logging

from mongoengine import DoesNotExist, MultipleObjectsReturned
from mongoengine.context_managers import switch_db

from ..meta_data import MetaDataManager
from ..models import PlateDefinitionModel
from ..plate import Plate
from ..utils import Printable, PlateDefinitionError


class PlateManager(Printable):
    """
    Plate manager. Manages the mapping between plates defined in the database with the global meta data definition.
    """
    def __init__(self):
        """
        Initialise the manager

        Want to get meta-data dictionaries for all plate combinations
        e.g.

        H plate, want

        {'H': [
          (('house', '1'),),
          {'house': '2'}
        ]}

        H1 plate, want

        {'H1': [(('house', '1'),)]}

        H.R plate, want

        {'H.R': [
          {'house': '1', 'resident': '1'},
          {'house': '1': 'resident': '2'},
          {'house': '2': 'resident': '1'}
        }
        """
        self.plates = {}
        self.meta_data_manager = MetaDataManager()

        # Plate definitions (arrays of streams)
        with switch_db(PlateDefinitionModel, db_alias="hyperstream"):
            for p in PlateDefinitionModel.objects:
                self.add_plate(p)
                
    def delete_plate(self, plate_id, delete_meta_data=False):
        """
        Delete a plate from the database
        
        :param plate_id: The plate id
        :param delete_meta_data: Optionally delete all meta data associated with this plate as well
        :return: None
        """

        if plate_id not in self.plates:
            logging.info("Plate {} not found for deletion".format(plate_id))
            return

        plate = self.plates[plate_id]

        if delete_meta_data:
            for pv in plate.values:
                identifier = ".".join(map(lambda x: "_".join(x), pv))
                self.meta_data_manager.delete(identifier=identifier)

        with switch_db(PlateDefinitionModel, "hyperstream"):
            try:
                p = PlateDefinitionModel.objects.get(plate_id=plate_id)
                p.delete()
                del self.plates[plate_id]
            except DoesNotExist as e:
                logging.warn(e)
        logging.info("Plate {} deleted".format(plate_id))

    def create_plate(self, plate_id, description, meta_data_id, values, complement, parent_plate):
        """
        Create a new plate, and commit it to the database

        :param plate_id: The plate id - required to be unique
        :param description: A human readable description
        :param meta_data_id: The meta data id, which should correspond to the tag in the global meta data
        :param values: Either a list of string values, or the empty list (for use with complement)
        :param complement: If complement is true, then the complement of the values list will be used when getting
        values from the global meta data
        :param parent_plate: The parent plate identifier
        :return: The newly created plate
        :type plate_id: str | unicode
        :type complement: bool
        :type values: list | tuple
        """
        # Make sure the plate id doesn't already exist
        with switch_db(PlateDefinitionModel, db_alias='hyperstream'):
            try:
                p = PlateDefinitionModel.objects.get(plate_id=plate_id)
                if p:
                    logging.info("Plate with id {} already exists".format(plate_id))
                    return self.plates[plate_id]
            except DoesNotExist:
                pass
            except MultipleObjectsReturned:
                raise

            plate_definition = PlateDefinitionModel(
                plate_id=plate_id,
                description=description,
                meta_data_id=meta_data_id,
                values=values,
                complement=complement,
                parent_plate=parent_plate
            )

            self.add_plate(plate_definition)
            plate_definition.save()
            return self.plates[plate_id]

    def add_plate(self, plate_definition):
        """
        Add a plate using the plate definition
        :param plate_definition: The plate definition
        :return: None
        :type plate_definition: PlateDefinitionModel
        """
        values = self.get_plate_values(plate_definition)

        # TODO: We should also be checking that the plate is defined over all of the values of the parent plate
        self.plates[plate_definition.plate_id] = Plate(
            plate_id=plate_definition.plate_id,
            meta_data_id=plate_definition.meta_data_id,
            values=values,
            parent_plate=self.plates[plate_definition.parent_plate] if plate_definition.parent_plate else None)

        logging.debug("Added plate: {}".format(self.plates[plate_definition.plate_id]))

    def get_plate_values(self, plate_definition):
        """
        Gets the plate values from the global meta data according to the given plate definition
        :param plate_definition: The plate definition
        :return: The plate values
        :type plate_definition: PlateDefinitionModel
        """
        if not plate_definition.values and not plate_definition.complement:
            raise PlateDefinitionError()

        values = []
        for n in self.meta_data_manager.global_plate_definitions.all_nodes():
            if n.tag == plate_definition.meta_data_id:
                if not plate_definition.values or n.data in plate_definition.values:
                    if plate_definition.parent_plate:
                        # This plate has parent plates, so we need to get parent data for the node
                        parent_plate_value = self.get_parent_plate_value(
                            self.meta_data_manager.global_plate_definitions, n)
                        # if tuple(parent_plate_value) not in self.plates[plate_definition.parent_plate].values:
                        if set(parent_plate_value) not in map(set, self.plates[plate_definition.parent_plate].values):
                            continue
                        values.insert(0, self.get_parent_data(
                            self.meta_data_manager.global_plate_definitions, n, [(n.tag, n.data), ]))
                    else:
                        values.insert(0, [(n.tag, n.data), ])
        if not values:
            # raise PlateEmptyError(plate_definition.plate_id)
            logging.warn("Plate {} is empty during the creation".format(plate_definition.plate_id))
        return values

    @staticmethod
    def get_parent_plate_value(tree, node, value=None):
        """
        Recurse up the tree getting parent plate values
        :param tree: The tree
        :param node: The current node
        :param value: The initial plate value
        :return: The plate value as a list of tuples
        """
        if value is None:
            value = []
        parent = tree.parent(node.identifier)
        if parent.is_root():
            # value.append((parent.tag, parent.identifier))
            return value
        value = PlateManager.get_parent_plate_value(tree, parent, value)
        if "." in parent.identifier:
            pass
        value.append((parent.tag, parent.data))
        return value

    @staticmethod
    def get_parent_data(tree, node, current):
        """
        Recurse up the tree getting parent data
        :param tree: The tree
        :param node: The current node
        :param current: The current list
        :return: The hierarchical dictionary
        """
        if not current:
            current = []

        parent = tree.parent(node.identifier)
        if parent.is_root():
            return current

        current.insert(0, (parent.tag, parent.data))
        return PlateManager.get_parent_data(tree, parent, current)