bika.lims.content.abstractroutineanalysis.AbstractRoutineAnalysis.setResultsRange() - Code Metrics - Inspection of "Apply dynamic analyses specs for new added analyse..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — 2.x ( bd647c...8bac86 )

by Jordi

created 2023-05-29 11:54 UTC

AbstractRoutineAnalysis.setResultsRange() A

↳ Parent: bika.lims.content.abstractroutineanalysis

Complexity

Conditions

Size

Total Lines	15
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	7
dl	0
loc	15
rs	10
c	0
b	0
f	0
cc	3
nop	3

# -*- 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-2021 by it's authors.
# Some rights reserved, see README and LICENSE.

import copy
from collections import OrderedDict
from datetime import timedelta

from AccessControl import ClassSecurityInfo
from bika.lims import api
from bika.lims import bikaMessageFactory as _
from bika.lims.browser.widgets import DecimalWidget
from bika.lims.content.abstractanalysis import AbstractAnalysis
from bika.lims.content.abstractanalysis import schema
from bika.lims.content.attachment import Attachment
from bika.lims.content.clientawaremixin import ClientAwareMixin
from bika.lims.interfaces import IAnalysis
from bika.lims.interfaces import ICancellable
from bika.lims.interfaces import IDynamicResultsRange
from bika.lims.interfaces import IInternalUse
from bika.lims.interfaces import IRoutineAnalysis
from bika.lims.interfaces.analysis import IRequestAnalysis
from bika.lims.workflow import getTransitionDate
from Products.Archetypes.Field import BooleanField
from Products.Archetypes.Field import StringField
from Products.Archetypes.Schema import Schema
from Products.ATContentTypes.utils import DT2dt
from Products.ATContentTypes.utils import dt2DT
from Products.CMFCore.permissions import View
from senaite.core.catalog.indexer.baseanalysis import sortable_title
from zope.interface import alsoProvides
from zope.interface import implements
from zope.interface import noLongerProvides

# The actual uncertainty for this analysis' result, populated when the result
# is submitted.
Uncertainty = StringField(
    'Uncertainty',
    read_permission=View,
    write_permission="Field: Edit Result",
    precision=10,
    widget=DecimalWidget(
        label=_("Uncertainty")
    )
)
# This field keep track if the field hidden has been set manually or not. If
# this value is false, the system will assume the visibility of this analysis
# in results report will depend on the value set at AR, Profile or Template
# levels (see AnalysisServiceSettings fields in AR). If the value for this
# field is set to true, the system will assume the visibility of the analysis
# will only depend on the value set for the field Hidden (bool).
HiddenManually = BooleanField(
    'HiddenManually',
    default=False,
)


schema = schema.copy() + Schema((
    Uncertainty,
    HiddenManually,
))


class AbstractRoutineAnalysis(AbstractAnalysis, ClientAwareMixin):
    implements(IAnalysis, IRequestAnalysis, IRoutineAnalysis, ICancellable)
    security = ClassSecurityInfo()
    displayContentsTab = False
    schema = schema


    @security.public
    def getRequest(self):
        """Returns the Analysis Request this analysis belongs to.
        Delegates to self.aq_parent
        """
        ar = self.aq_parent
        return ar

    @security.public
    def getRequestID(self):
        """Used to populate catalog values.
        Returns the ID of the parent analysis request.
        """
        ar = self.getRequest()
        if ar:
            return ar.getId()

    @security.public
    def getRequestUID(self):
        """Returns the UID of the parent analysis request.
        """
        ar = self.getRequest()
        if ar:
            return ar.UID()

    @security.public
    def getRequestURL(self):
        """Returns the url path of the Analysis Request object this analysis
        belongs to. Returns None if there is no Request assigned.
        :return: the Analysis Request URL path this analysis belongs to
        :rtype: str
        """
        request = self.getRequest()
        if request:
            return request.absolute_url_path()

    def getClient(self):
        """Returns the Client this analysis is bound to, if any
        """
        request = self.getRequest()
        return request and request.getClient() or None

    @security.public
    def getClientOrderNumber(self):
        """Used to populate catalog values.
        Returns the ClientOrderNumber of the associated AR
        """
        request = self.getRequest()
        if request:
            return request.getClientOrderNumber()

    @security.public
    def getDateReceived(self):
        """Used to populate catalog values.
        Returns the date the Analysis Request this analysis belongs to was
        received. If the analysis was created after, then returns the date
        the analysis was created.
        """
        request = self.getRequest()
        if request:
            ar_date = request.getDateReceived()
            if ar_date and self.created() > ar_date:
                return self.created()
            return ar_date
        return None

    @security.public
    def isSampleReceived(instance):
        """Returns whether if the Analysis Request this analysis comes from has
        been received or not
        """
        return instance.getDateReceived() and True or False

    @security.public
    def getDatePublished(self):
        """Used to populate catalog values.
        Returns the date on which the "publish" transition was invoked on this
        analysis.
        """
        return getTransitionDate(self, 'publish', return_as_datetime=True)

    @security.public
    def getDateSampled(self):
        """Returns the date when the Sample was Sampled
        """
        request = self.getRequest()
        if request:
            return request.getDateSampled()
        return None

    @security.public
    def isSampleSampled(self):
        """Returns whether if the Analysis Request this analysis comes from has
        been received or not
        """
        return self.getDateSampled() and True or False

    @security.public
    def getStartProcessDate(self):
        """Returns the date time when the analysis request the analysis belongs
        to was received. If the analysis request hasn't yet been received,
        returns None
        Overrides getStartProcessDateTime from the base class
        :return: Date time when the analysis is ready to be processed.
        :rtype: DateTime
        """
        return self.getDateReceived()

    @security.public
    def getSamplePoint(self):
        request = self.getRequest()
        if request:
            return request.getSamplePoint()
        return None

    @security.public
    def getDueDate(self):
        """Used to populate getDueDate index and metadata.
        This calculates the difference between the time the analysis processing
        started and the maximum turnaround time. If the analysis has no
        turnaround time set or is not yet ready for proces, returns None
        """
        tat = self.getMaxTimeAllowed()
        if not tat:
            return None
        start = self.getStartProcessDate()
        if not start:
            return None

        # delta time when the first analysis is considered as late
        delta = timedelta(minutes=api.to_minutes(**tat))

        # calculated due date
        end = dt2DT(DT2dt(start) + delta)

        # delta is within one day, return immediately
        if delta.days == 0:
            return end

        # get the laboratory workdays
        setup = api.get_setup()
        workdays = setup.getWorkdays()

        # every day is a workday, no need for calculation
        if workdays == tuple(map(str, range(7))):
            return end

        # reset the due date to the received date, and add only for configured
        # workdays another day
        due_date = end - delta.days

        days = 0
        while days < delta.days:
            # add one day to the new due date
            due_date += 1
            # skip if the weekday is a non working day
            if str(due_date.asdatetime().weekday()) not in workdays:
                continue
            days += 1

        return due_date

    @security.public
    def getSampleType(self):
        request = self.getRequest()
        if request:
            return request.getSampleType()
        return None

    @security.public
    def getSampleTypeUID(self):
        """Used to populate catalog values.
        """
        sample = self.getRequest()
        if not sample:
            return None
        return sample.getRawSampleType()

    @security.public
    def getResultsRange(self):
        """Returns the valid result range for this routine analysis

        A routine analysis will be considered out of range if it result falls
        out of the range defined in "min" and "max". If there are values set
        for "warn_min" and "warn_max", these are used to compute the shoulders
        in both ends of the range. Thus, an analysis can be out of range, but
        be within shoulders still.

        :return: A dictionary with keys "min", "max", "warn_min" and "warn_max"
        :rtype: dict
        """
        return self.getField("ResultsRange").get(self)

    @security.private
    def setResultsRange(self, spec, update_dynamic_spec=True):
        """Set the results range for this routine analysis

        NOTE: This custom setter also applies dynamic specifications

        :param spec: The result range to set
        :param update_dynamic_spec: If True, dynamic specifications are update
        """
        adapter = IDynamicResultsRange(self, None)
        if adapter and update_dynamic_spec:
            # update the result range with the dynamic values
            spec.update(adapter())
        field = self.getField("ResultsRange")
        field.set(self, spec)

    @security.public
    def getSiblings(self, with_retests=False):
        """
        Return the siblings analyses, using the parent to which the current
        analysis belongs to as the source
        :param with_retests: If false, siblings with retests are dismissed
        :type with_retests: bool
        :return: list of siblings for this analysis
        :rtype: list of IAnalysis
        """
        raise NotImplementedError("getSiblings is not implemented.")

    @security.public
    def getCalculation(self):
        """Return current assigned calculation
        """
        field = self.getField("Calculation")
        calculation = field.get(self)
        if not calculation:
            return None
        return calculation

    @security.public
    def setCalculation(self, value):
        self.getField("Calculation").set(self, value)
        # TODO Something weird here
        # Reset interims so they get extended with those from calculation
        # see bika.lims.browser.fields.interimfieldsfield.set
        interim_fields = copy.deepcopy(self.getInterimFields())
        self.setInterimFields(interim_fields)

    @security.public
    def getDependents(self, with_retests=False, recursive=False):
        """
        Returns a list of siblings who depend on us to calculate their result.
        :param with_retests: If false, dependents with retests are dismissed
        :param recursive: If true, returns all dependents recursively down
        :type with_retests: bool
        :return: Analyses the current analysis depends on
        :rtype: list of IAnalysis
        """
        def is_dependent(analysis):
            # Never consider myself as dependent
            if analysis.UID() == self.UID():
                return False

            # Never consider analyses from same service as dependents
            self_service_uid = self.getRawAnalysisService()
            if analysis.getRawAnalysisService() == self_service_uid:
                return False

            # Without calculation, no dependency relationship is possible
            calculation = analysis.getCalculation()
            if not calculation:
                return False

            # Calculation must have the service I belong to
            services = calculation.getRawDependentServices()
            return self_service_uid in services
        
        request = self.getRequest()
        if request.isPartition():
            parent = request.getParentAnalysisRequest()
            siblings = parent.getAnalyses(full_objects=True)
        else:
            siblings = self.getSiblings(with_retests=with_retests)

        dependents = filter(lambda sib: is_dependent(sib), siblings)
        if not recursive:
            return dependents

        # Return all dependents recursively
        deps = dependents
        for dep in dependents:
            down_dependencies = dep.getDependents(with_retests=with_retests,
                                                  recursive=True)
            deps.extend(down_dependencies)
        return deps

    @security.public
    def getDependencies(self, with_retests=False, recursive=False):
        """
        Return a list of siblings who we depend on to calculate our result.
        :param with_retests: If false, siblings with retests are dismissed
        :param recursive: If true, looks for dependencies recursively up
        :type with_retests: bool
        :return: Analyses the current analysis depends on
        :rtype: list of IAnalysis
        """
        calc = self.getCalculation()
        if not calc:
            return []

        # If the calculation this analysis is bound does not have analysis
        # keywords (only interims), no need to go further
        service_uids = calc.getRawDependentServices()

        # Ensure we exclude ourselves
        service_uid = self.getRawAnalysisService()
        service_uids = filter(lambda serv: serv != service_uid, service_uids)

        if len(service_uids) == 0:
            return []

        dependencies = []
        for sibling in self.getSiblings(with_retests=with_retests):
            # We get all analyses that depend on me, also if retracted (maybe
            # I am one of those that are retracted!)
            deps = map(api.get_uid, sibling.getDependents(with_retests=True))
            if self.UID() in deps:
                dependencies.append(sibling)
                if recursive:
                    # Append the dependencies of this dependency
                    up_deps = sibling.getDependencies(with_retests=with_retests,
                                                      recursive=True)
                    dependencies.extend(up_deps)

        # Exclude analyses of same service as me to prevent max recursion depth
        return filter(lambda dep: dep.getRawAnalysisService() != service_uid,

                      dependencies)

    @security.public
    def getPrioritySortkey(self):
        """
        Returns the key that will be used to sort the current Analysis, from
        most prioritary to less prioritary.
        :return: string used for sorting
        """
        analysis_request = self.getRequest()
        if analysis_request is None:
            return None
        ar_sort_key = analysis_request.getPrioritySortkey()
        ar_id = analysis_request.getId().lower()
        title = sortable_title(self)
        if callable(title):
            title = title()
        return '{}.{}.{}'.format(ar_sort_key, ar_id, title)

    @security.public
    def getHidden(self):
        """ Returns whether if the analysis must be displayed in results
        reports or not, as well as in analyses view when the user logged in
        is a Client Contact.

        If the value for the field HiddenManually is set to False, this function
        will delegate the action to the method getAnalysisServiceSettings() from
        the Analysis Request.

        If the value for the field HiddenManually is set to True, this function
        will return the value of the field Hidden.
        :return: true or false
        :rtype: bool
        """
        if self.getHiddenManually():
            return self.getField('Hidden').get(self)
        request = self.getRequest()
        if request:
            service_uid = self.getServiceUID()
            ar_settings = request.getAnalysisServiceSettings(service_uid)
            return ar_settings.get('hidden', False)
        return False

    @security.public
    def setHidden(self, hidden):
        """ Sets if this analysis must be displayed or not in results report and
        in manage analyses view if the user is a lab contact as well.

        The value set by using this field will have priority over the visibility
        criteria set at Analysis Request, Template or Profile levels (see
        field AnalysisServiceSettings from Analysis Request. To achieve this
        behavior, this setter also sets the value to HiddenManually to true.
        :param hidden: true if the analysis must be hidden in report
        :type hidden: bool
        """
        self.setHiddenManually(True)
        self.getField('Hidden').set(self, hidden)

    @security.public
    def setInternalUse(self, internal_use):
        """Applies the internal use of this Analysis. Analyses set for internal
        use are not accessible to clients and are not visible in reports
        """
        if internal_use:
            alsoProvides(self, IInternalUse)
        else:
            noLongerProvides(self, IInternalUse)

    def getConditions(self, empties=False):
        """Returns the conditions of this analysis. These conditions are usually
        set on sample registration and are stored at sample level. Do not return
        conditions with empty value unless `empties` is True
        """
        sample = self.getRequest()
        service_uid = self.getRawAnalysisService()
        attachments = self.getRawAttachment()

        def is_valid(condition):
            uid = condition.get("uid")
            if api.is_uid(uid) and uid == service_uid:
                if empties:
                    return True
                value = condition.get("value", None)
                if condition.get("type") == "file":
                    return value in attachments
                return value not in [None, ""]
            return False

        conditions = sample.getServiceConditions()
        conditions = filter(is_valid, conditions)
        return copy.deepcopy(conditions)

    def setConditions(self, conditions):
        """Sets the conditions of this analysis. These conditions are usually
        set on sample registration and are stored at sample level
        """
        if not conditions:
            conditions = []

        sample = self.getRequest()
        service_uid = self.getRawAnalysisService()
        sample_conditions = sample.getServiceConditions()
        sample_conditions = copy.deepcopy(sample_conditions)

        # Keep the conditions from sample for analyses other than this one
        other_conditions = filter(lambda c: c.get("uid") != service_uid,
                                  sample_conditions)

        def to_condition(condition):
            # Type and title are required
            title = condition.get("title")
            cond_type = condition.get("type")
            if not all([title, cond_type]):
                return None

            condition_info = {
                "uid": service_uid,
                "type": cond_type,
                "title": title,
                "description": "",
                "choices": "",
                "default": "",
                "required": "",
                "value": "",
            }
            condition = dict(condition)
            condition_info.update(condition)
            return condition_info

        # Sanitize the conditions
        conditions = filter(None, [to_condition(cond) for cond in conditions])

        # Conditions from "file" type are stored as attachments
        attachments = []
        for condition in conditions:
            if condition.get("type") != "file":
                continue

            value = condition.get("value")
            orig_attachment = condition.pop("attachment", None)
            if api.is_uid(value):
                # link to an existing attachment
                attachments.append(value)
            elif isinstance(value, Attachment):
                # link to an existing attachment
                uid = api.get_uid(value)
                attachments.append(uid)
                condition["value"] = uid
            elif getattr(value, "filename", ""):
                # create new attachment
                client = sample.getClient()
                att = api.create(client, "Attachment", AttachmentFile=value)
                attachments.append(api.get_uid(att))
                # update with the attachment uid
                condition["value"] = api.get_uid(att)
            elif api.is_uid(orig_attachment):
                # restore to original attachment
                attachments.append(orig_attachment)
                condition["value"] = orig_attachment
            else:
                # nothing we can handle, update with an empty value
                condition["value"] = ""

        # Link the analysis to the attachments
        if attachments:
            # Remove duplicates while keeping the order
            attachments.extend(self.getRawAttachment() or [])
            attachments = list(OrderedDict.fromkeys(attachments))
            self.setAttachment(attachments)

        # Store the conditions in sample
        sample.setServiceConditions(other_conditions + conditions)

    @security.public
    def getPrice(self):
        """The function obtains the analysis' price without VAT and without
        member discount
        :return: the price (without VAT or Member Discount) in decimal format
        """
        client = self.getClient()
        if client and client.getBulkDiscount():
            return self.getBulkPrice()
        return self.getField('Price').get(self)


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE.
4			#
5			# SENAITE.CORE is free software: you can redistribute it and/or modify it under
6			# the terms of the GNU General Public License as published by the Free Software
7			# Foundation, version 2.
8			#
9			# This program is distributed in the hope that it will be useful, but WITHOUT
10			# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11			# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
12			# details.
13			#
14			# You should have received a copy of the GNU General Public License along with
15			# this program; if not, write to the Free Software Foundation, Inc., 51
16			# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17			#
18			# Copyright 2018-2021 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			import copy
22			from collections import OrderedDict
23			from datetime import timedelta
24
25			from AccessControl import ClassSecurityInfo
26			from bika.lims import api
27			from bika.lims import bikaMessageFactory as _
28			from bika.lims.browser.widgets import DecimalWidget
29			from bika.lims.content.abstractanalysis import AbstractAnalysis
30			from bika.lims.content.abstractanalysis import schema
31			from bika.lims.content.attachment import Attachment
32			from bika.lims.content.clientawaremixin import ClientAwareMixin
33			from bika.lims.interfaces import IAnalysis
34			from bika.lims.interfaces import ICancellable
35			from bika.lims.interfaces import IDynamicResultsRange
36			from bika.lims.interfaces import IInternalUse
37			from bika.lims.interfaces import IRoutineAnalysis
38			from bika.lims.interfaces.analysis import IRequestAnalysis
39			from bika.lims.workflow import getTransitionDate
40			from Products.Archetypes.Field import BooleanField
41			from Products.Archetypes.Field import StringField
42			from Products.Archetypes.Schema import Schema
43			from Products.ATContentTypes.utils import DT2dt
44			from Products.ATContentTypes.utils import dt2DT
45			from Products.CMFCore.permissions import View
46			from senaite.core.catalog.indexer.baseanalysis import sortable_title
47			from zope.interface import alsoProvides
48			from zope.interface import implements
49			from zope.interface import noLongerProvides
50
51			# The actual uncertainty for this analysis' result, populated when the result
52			# is submitted.
53			Uncertainty = StringField(
54			'Uncertainty',
55			read_permission=View,
56			write_permission="Field: Edit Result",
57			precision=10,
58			widget=DecimalWidget(
59			label=_("Uncertainty")
60			)
61			)
62			# This field keep track if the field hidden has been set manually or not. If
63			# this value is false, the system will assume the visibility of this analysis
64			# in results report will depend on the value set at AR, Profile or Template
65			# levels (see AnalysisServiceSettings fields in AR). If the value for this
66			# field is set to true, the system will assume the visibility of the analysis
67			# will only depend on the value set for the field Hidden (bool).
68			HiddenManually = BooleanField(
69			'HiddenManually',
70			default=False,
71			)
72
73
74			schema = schema.copy() + Schema((
75			Uncertainty,
76			HiddenManually,
77			))
78
79
80			class AbstractRoutineAnalysis(AbstractAnalysis, ClientAwareMixin):
81			implements(IAnalysis, IRequestAnalysis, IRoutineAnalysis, ICancellable)
82			security = ClassSecurityInfo()
83			displayContentsTab = False
84			schema = schema
			0 ignored issues – show Comprehensibility Best Practice introduced 2018-07-25 08:58 UTC by Report Bug Copy Issue Report The variable `schema` does not seem to be defined. Loading history...
85
86			@security.public
87			def getRequest(self):
88			"""Returns the Analysis Request this analysis belongs to.
89			Delegates to self.aq_parent
90			"""
91			ar = self.aq_parent
92			return ar
93
94			@security.public
95			def getRequestID(self):
96			"""Used to populate catalog values.
97			Returns the ID of the parent analysis request.
98			"""
99			ar = self.getRequest()
100			if ar:
101			return ar.getId()
102
103			@security.public
104			def getRequestUID(self):
105			"""Returns the UID of the parent analysis request.
106			"""
107			ar = self.getRequest()
108			if ar:
109			return ar.UID()
110
111			@security.public
112			def getRequestURL(self):
113			"""Returns the url path of the Analysis Request object this analysis
114			belongs to. Returns None if there is no Request assigned.
115			:return: the Analysis Request URL path this analysis belongs to
116			:rtype: str
117			"""
118			request = self.getRequest()
119			if request:
120			return request.absolute_url_path()
121
122			def getClient(self):
123			"""Returns the Client this analysis is bound to, if any
124			"""
125			request = self.getRequest()
126			return request and request.getClient() or None
127
128			@security.public
129			def getClientOrderNumber(self):
130			"""Used to populate catalog values.
131			Returns the ClientOrderNumber of the associated AR
132			"""
133			request = self.getRequest()
134			if request:
135			return request.getClientOrderNumber()
136
137			@security.public
138			def getDateReceived(self):
139			"""Used to populate catalog values.
140			Returns the date the Analysis Request this analysis belongs to was
141			received. If the analysis was created after, then returns the date
142			the analysis was created.
143			"""
144			request = self.getRequest()
145			if request:
146			ar_date = request.getDateReceived()
147			if ar_date and self.created() > ar_date:
148			return self.created()
149			return ar_date
150			return None
151
152			@security.public
153			def isSampleReceived(instance):
154			"""Returns whether if the Analysis Request this analysis comes from has
155			been received or not
156			"""
157			return instance.getDateReceived() and True or False
158
159			@security.public
160			def getDatePublished(self):
161			"""Used to populate catalog values.
162			Returns the date on which the "publish" transition was invoked on this
163			analysis.
164			"""
165			return getTransitionDate(self, 'publish', return_as_datetime=True)
166
167			@security.public
168			def getDateSampled(self):
169			"""Returns the date when the Sample was Sampled
170			"""
171			request = self.getRequest()
172			if request:
173			return request.getDateSampled()
174			return None
175
176			@security.public
177			def isSampleSampled(self):
178			"""Returns whether if the Analysis Request this analysis comes from has
179			been received or not
180			"""
181			return self.getDateSampled() and True or False
182
183			@security.public
184			def getStartProcessDate(self):
185			"""Returns the date time when the analysis request the analysis belongs
186			to was received. If the analysis request hasn't yet been received,
187			returns None
188			Overrides getStartProcessDateTime from the base class
189			:return: Date time when the analysis is ready to be processed.
190			:rtype: DateTime
191			"""
192			return self.getDateReceived()
193
194			@security.public
195			def getSamplePoint(self):
196			request = self.getRequest()
197			if request:
198			return request.getSamplePoint()
199			return None
200
201			@security.public
202			def getDueDate(self):
203			"""Used to populate getDueDate index and metadata.
204			This calculates the difference between the time the analysis processing
205			started and the maximum turnaround time. If the analysis has no
206			turnaround time set or is not yet ready for proces, returns None
207			"""
208			tat = self.getMaxTimeAllowed()
209			if not tat:
210			return None
211			start = self.getStartProcessDate()
212			if not start:
213			return None
214
215			# delta time when the first analysis is considered as late
216			delta = timedelta(minutes=api.to_minutes(**tat))
217
218			# calculated due date
219			end = dt2DT(DT2dt(start) + delta)
220
221			# delta is within one day, return immediately
222			if delta.days == 0:
223			return end
224
225			# get the laboratory workdays
226			setup = api.get_setup()
227			workdays = setup.getWorkdays()
228
229			# every day is a workday, no need for calculation
230			if workdays == tuple(map(str, range(7))):
231			return end
232
233			# reset the due date to the received date, and add only for configured
234			# workdays another day
235			due_date = end - delta.days
236
237			days = 0
238			while days < delta.days:
239			# add one day to the new due date
240			due_date += 1
241			# skip if the weekday is a non working day
242			if str(due_date.asdatetime().weekday()) not in workdays:
243			continue
244			days += 1
245
246			return due_date
247
248			@security.public
249			def getSampleType(self):
250			request = self.getRequest()
251			if request:
252			return request.getSampleType()
253			return None
254
255			@security.public
256			def getSampleTypeUID(self):
257			"""Used to populate catalog values.
258			"""
259			sample = self.getRequest()
260			if not sample:
261			return None
262			return sample.getRawSampleType()
263
264			@security.public
265			def getResultsRange(self):
266			"""Returns the valid result range for this routine analysis
267
268			A routine analysis will be considered out of range if it result falls
269			out of the range defined in "min" and "max". If there are values set
270			for "warn_min" and "warn_max", these are used to compute the shoulders
271			in both ends of the range. Thus, an analysis can be out of range, but
272			be within shoulders still.
273
274			:return: A dictionary with keys "min", "max", "warn_min" and "warn_max"
275			:rtype: dict
276			"""
277			return self.getField("ResultsRange").get(self)
278
279			@security.private
280			def setResultsRange(self, spec, update_dynamic_spec=True):
281			"""Set the results range for this routine analysis
282
283			NOTE: This custom setter also applies dynamic specifications
284
285			:param spec: The result range to set
286			:param update_dynamic_spec: If True, dynamic specifications are update
287			"""
288			adapter = IDynamicResultsRange(self, None)
289			if adapter and update_dynamic_spec:
290			# update the result range with the dynamic values
291			spec.update(adapter())
292			field = self.getField("ResultsRange")
293			field.set(self, spec)
294
295			@security.public
296			def getSiblings(self, with_retests=False):
297			"""
298			Return the siblings analyses, using the parent to which the current
299			analysis belongs to as the source
300			:param with_retests: If false, siblings with retests are dismissed
301			:type with_retests: bool
302			:return: list of siblings for this analysis
303			:rtype: list of IAnalysis
304			"""
305			raise NotImplementedError("getSiblings is not implemented.")
306
307			@security.public
308			def getCalculation(self):
309			"""Return current assigned calculation
310			"""
311			field = self.getField("Calculation")
312			calculation = field.get(self)
313			if not calculation:
314			return None
315			return calculation
316
317			@security.public
318			def setCalculation(self, value):
319			self.getField("Calculation").set(self, value)
320			# TODO Something weird here
321			# Reset interims so they get extended with those from calculation
322			# see bika.lims.browser.fields.interimfieldsfield.set
323			interim_fields = copy.deepcopy(self.getInterimFields())
324			self.setInterimFields(interim_fields)
325
326			@security.public
327			def getDependents(self, with_retests=False, recursive=False):
328			"""
329			Returns a list of siblings who depend on us to calculate their result.
330			:param with_retests: If false, dependents with retests are dismissed
331			:param recursive: If true, returns all dependents recursively down
332			:type with_retests: bool
333			:return: Analyses the current analysis depends on
334			:rtype: list of IAnalysis
335			"""
336			def is_dependent(analysis):
337			# Never consider myself as dependent
338			if analysis.UID() == self.UID():
339			return False
340
341			# Never consider analyses from same service as dependents
342			self_service_uid = self.getRawAnalysisService()
343			if analysis.getRawAnalysisService() == self_service_uid:
344			return False
345
346			# Without calculation, no dependency relationship is possible
347			calculation = analysis.getCalculation()
348			if not calculation:
349			return False
350
351			# Calculation must have the service I belong to
352			services = calculation.getRawDependentServices()
353			return self_service_uid in services
354
355			request = self.getRequest()
356			if request.isPartition():
357			parent = request.getParentAnalysisRequest()
358			siblings = parent.getAnalyses(full_objects=True)
359			else:
360			siblings = self.getSiblings(with_retests=with_retests)
361
362			dependents = filter(lambda sib: is_dependent(sib), siblings)
363			if not recursive:
364			return dependents
365
366			# Return all dependents recursively
367			deps = dependents
368			for dep in dependents:
369			down_dependencies = dep.getDependents(with_retests=with_retests,
370			recursive=True)
371			deps.extend(down_dependencies)
372			return deps
373
374			@security.public
375			def getDependencies(self, with_retests=False, recursive=False):
376			"""
377			Return a list of siblings who we depend on to calculate our result.
378			:param with_retests: If false, siblings with retests are dismissed
379			:param recursive: If true, looks for dependencies recursively up
380			:type with_retests: bool
381			:return: Analyses the current analysis depends on
382			:rtype: list of IAnalysis
383			"""
384			calc = self.getCalculation()
385			if not calc:
386			return []
387
388			# If the calculation this analysis is bound does not have analysis
389			# keywords (only interims), no need to go further
390			service_uids = calc.getRawDependentServices()
391
392			# Ensure we exclude ourselves
393			service_uid = self.getRawAnalysisService()
394			service_uids = filter(lambda serv: serv != service_uid, service_uids)
			0 ignored issues – show introduced 2021-11-11 16:26 UTC by Report Bug Copy Issue Report The variable `service_uid` does not seem to be defined for all execution paths. Loading history...
395			if len(service_uids) == 0:
396			return []
397
398			dependencies = []
399			for sibling in self.getSiblings(with_retests=with_retests):
400			# We get all analyses that depend on me, also if retracted (maybe
401			# I am one of those that are retracted!)
402			deps = map(api.get_uid, sibling.getDependents(with_retests=True))
403			if self.UID() in deps:
404			dependencies.append(sibling)
405			if recursive:
406			# Append the dependencies of this dependency
407			up_deps = sibling.getDependencies(with_retests=with_retests,
408			recursive=True)
409			dependencies.extend(up_deps)
410
411			# Exclude analyses of same service as me to prevent max recursion depth
412			return filter(lambda dep: dep.getRawAnalysisService() != service_uid,
			0 ignored issues – show introduced 2021-11-11 16:26 UTC by Report Bug Copy Issue Report The variable `service_uid` does not seem to be defined for all execution paths. Loading history...
413			dependencies)
414
415			@security.public
416			def getPrioritySortkey(self):
417			"""
418			Returns the key that will be used to sort the current Analysis, from
419			most prioritary to less prioritary.
420			:return: string used for sorting
421			"""
422			analysis_request = self.getRequest()
423			if analysis_request is None:
424			return None
425			ar_sort_key = analysis_request.getPrioritySortkey()
426			ar_id = analysis_request.getId().lower()
427			title = sortable_title(self)
428			if callable(title):
429			title = title()
430			return '{}.{}.{}'.format(ar_sort_key, ar_id, title)
431
432			@security.public
433			def getHidden(self):
434			""" Returns whether if the analysis must be displayed in results
435			reports or not, as well as in analyses view when the user logged in
436			is a Client Contact.
437
438			If the value for the field HiddenManually is set to False, this function
439			will delegate the action to the method getAnalysisServiceSettings() from
440			the Analysis Request.
441
442			If the value for the field HiddenManually is set to True, this function
443			will return the value of the field Hidden.
444			:return: true or false
445			:rtype: bool
446			"""
447			if self.getHiddenManually():
448			return self.getField('Hidden').get(self)
449			request = self.getRequest()
450			if request:
451			service_uid = self.getServiceUID()
452			ar_settings = request.getAnalysisServiceSettings(service_uid)
453			return ar_settings.get('hidden', False)
454			return False
455
456			@security.public
457			def setHidden(self, hidden):
458			""" Sets if this analysis must be displayed or not in results report and
459			in manage analyses view if the user is a lab contact as well.
460
461			The value set by using this field will have priority over the visibility
462			criteria set at Analysis Request, Template or Profile levels (see
463			field AnalysisServiceSettings from Analysis Request. To achieve this
464			behavior, this setter also sets the value to HiddenManually to true.
465			:param hidden: true if the analysis must be hidden in report
466			:type hidden: bool
467			"""
468			self.setHiddenManually(True)
469			self.getField('Hidden').set(self, hidden)
470
471			@security.public
472			def setInternalUse(self, internal_use):
473			"""Applies the internal use of this Analysis. Analyses set for internal
474			use are not accessible to clients and are not visible in reports
475			"""
476			if internal_use:
477			alsoProvides(self, IInternalUse)
478			else:
479			noLongerProvides(self, IInternalUse)
480
481			def getConditions(self, empties=False):
482			"""Returns the conditions of this analysis. These conditions are usually
483			set on sample registration and are stored at sample level. Do not return
484			conditions with empty value unless `empties` is True
485			"""
486			sample = self.getRequest()
487			service_uid = self.getRawAnalysisService()
488			attachments = self.getRawAttachment()
489
490			def is_valid(condition):
491			uid = condition.get("uid")
492			if api.is_uid(uid) and uid == service_uid:
493			if empties:
494			return True
495			value = condition.get("value", None)
496			if condition.get("type") == "file":
497			return value in attachments
498			return value not in [None, ""]
499			return False
500
501			conditions = sample.getServiceConditions()
502			conditions = filter(is_valid, conditions)
503			return copy.deepcopy(conditions)
504
505			def setConditions(self, conditions):
506			"""Sets the conditions of this analysis. These conditions are usually
507			set on sample registration and are stored at sample level
508			"""
509			if not conditions:
510			conditions = []
511
512			sample = self.getRequest()
513			service_uid = self.getRawAnalysisService()
514			sample_conditions = sample.getServiceConditions()
515			sample_conditions = copy.deepcopy(sample_conditions)
516
517			# Keep the conditions from sample for analyses other than this one
518			other_conditions = filter(lambda c: c.get("uid") != service_uid,
519			sample_conditions)
520
521			def to_condition(condition):
522			# Type and title are required
523			title = condition.get("title")
524			cond_type = condition.get("type")
525			if not all([title, cond_type]):
526			return None
527
528			condition_info = {
529			"uid": service_uid,
530			"type": cond_type,
531			"title": title,
532			"description": "",
533			"choices": "",
534			"default": "",
535			"required": "",
536			"value": "",
537			}
538			condition = dict(condition)
539			condition_info.update(condition)
540			return condition_info
541
542			# Sanitize the conditions
543			conditions = filter(None, [to_condition(cond) for cond in conditions])
544
545			# Conditions from "file" type are stored as attachments
546			attachments = []
547			for condition in conditions:
548			if condition.get("type") != "file":
549			continue
550
551			value = condition.get("value")
552			orig_attachment = condition.pop("attachment", None)
553			if api.is_uid(value):
554			# link to an existing attachment
555			attachments.append(value)
556			elif isinstance(value, Attachment):
557			# link to an existing attachment
558			uid = api.get_uid(value)
559			attachments.append(uid)
560			condition["value"] = uid
561			elif getattr(value, "filename", ""):
562			# create new attachment
563			client = sample.getClient()
564			att = api.create(client, "Attachment", AttachmentFile=value)
565			attachments.append(api.get_uid(att))
566			# update with the attachment uid
567			condition["value"] = api.get_uid(att)
568			elif api.is_uid(orig_attachment):
569			# restore to original attachment
570			attachments.append(orig_attachment)
571			condition["value"] = orig_attachment
572			else:
573			# nothing we can handle, update with an empty value
574			condition["value"] = ""
575
576			# Link the analysis to the attachments
577			if attachments:
578			# Remove duplicates while keeping the order
579			attachments.extend(self.getRawAttachment() or [])
580			attachments = list(OrderedDict.fromkeys(attachments))
581			self.setAttachment(attachments)
582
583			# Store the conditions in sample
584			sample.setServiceConditions(other_conditions + conditions)
585
586			@security.public
587			def getPrice(self):
588			"""The function obtains the analysis' price without VAT and without
589			member discount
590			:return: the price (without VAT or Member Discount) in decimal format
591			"""
592			client = self.getClient()
593			if client and client.getBulkDiscount():
594			return self.getBulkPrice()
595			return self.getField('Price').get(self)
596

senaite / senaite.core

Push — 2.x ( bd647c...8bac86 )

AbstractRoutineAnalysis.setResultsRange() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like