senaite.core.datamanagers.content.analysis

Complexity

Total Complexity

24

Size/Duplication

Total Lines	151
Duplicated Lines	0 %

Importance

Changes

0

6 Methods

Rating	Name	Duplication	Size	Complexity
B	RoutineAnalysisDataManager.recalculate_results()	0	29	8
C	RoutineAnalysisDataManager.set()	0	44	9
A	RoutineAnalysisDataManager.get()	0	21	4
A	RoutineAnalysisDataManager.is_field_readable()	0	4	1
A	RoutineAnalysisDataManager.is_field_writeable()	0	4	1
A	RoutineAnalysisDataManager.fields()	0	3	1

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE.
#
# SENAITE.CORE is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright 2018-2024 by it's authors.
# Some rights reserved, see README and LICENSE.

from AccessControl import Unauthorized
from bika.lims import api
from bika.lims.interfaces import IReferenceAnalysis
from bika.lims.interfaces import IRoutineAnalysis
from Products.Archetypes.utils import mapply
from senaite.core import logger
from senaite.core.datamanagers.base import DataManager
from zope.component import adapter


@adapter(IRoutineAnalysis)
class RoutineAnalysisDataManager(DataManager):
    """Data Manager for Routine Analyses
    """

    @property
    def fields(self):
        return api.get_fields(self.context)

    def is_field_readable(self, field):
        """Checks if the field is readable
        """
        return field.checkPermission("get", self.context)

    def is_field_writeable(self, field):
        """Checks if the field is writeable
        """
        return field.checkPermission("set", self.context)

    def get(self, name):
        """Get analysis field
        """
        # schema field
        field = self.fields.get(name)

        # check if the field exists
        if field is None:
            raise AttributeError("Field '{}' not found".format(name))

        # Check the permission of the field
        if not self.is_field_readable(field):
            raise Unauthorized("Field '{}' not readable!".format(name))

        # return the value with the field accessor
        if hasattr(field, "getAccessor"):
            accessor = field.getAccessor(self.context)
            return accessor()
        else:
            # Set the value on the field directly
            return field.get(self.context)

    def set(self, name, value):
        """Set analysis field/interim value
        """
        # set of updated objects
        updated_objects = set()

        # schema field
        field = self.fields.get(name)

        # interims
        interims = self.context.getInterimFields()
        interim_keys = map(lambda i: i.get("keyword"), interims)

        # schema field found
        if field:
            # Check the permission of the field
            if not self.is_field_writeable(field):
                logger.error("Field '{}' not writeable!".format(name))
                return []
            # get the field mutator (works only for AT content types)
            if hasattr(field, "getMutator"):
                mutator = field.getMutator(self.context)
                mapply(mutator, value)
            else:
                # Set the value on the field directly
                field.set(self.context, value)
            updated_objects.add(self.context)

        # interim key found
        elif name in interim_keys:
            # Check the permission of the field
            interim_field = self.fields.get("InterimFields")
            if not self.is_field_writeable(interim_field):
                logger.error("Interim field '{}' not writeable!".format(name))
                return []
            self.context.setInterimValue(name, value)

        # recalculate dependent results for result and interim fields
        if name == "Result" or name in interim_keys:
            updated_objects.add(self.context)
            updated_objects.update(self.recalculate_results(self.context))

        # return a unified list of the updated objects
        return list(updated_objects)

    def recalculate_results(self, obj, recalculated=None):
        """Recalculate the result of the object and its dependents

        :returns: List of recalculated objects
        """
        if recalculated is None:
            recalculated = set()

        # avoid double recalculation in recursion
        if obj in recalculated:
            return set()

        # recalculate own result
        if obj.calculateResult(override=True):
            # append object to the list of recalculated results
            recalculated.add(obj)
        # recalculate dependent analyses
        for dep in obj.getDependents():
            if dep.calculateResult(override=True):
                # TODO the `calculateResult` method should return False here!
                if dep.getResult() in ["NA", "0/0"]:
                    continue
                recalculated.add(dep)
            # recalculate dependents of dependents
            for ddep in dep.getDependents():
                recalculated.update(
                    self.recalculate_results(
                        ddep, recalculated=recalculated))
        return recalculated


@adapter(IReferenceAnalysis)
class ReferenceAnalysisDataManager(RoutineAnalysisDataManager):
    """Data Manager for Reference Analyses
    """