bika.lims.content.abstractroutineanalysis.AbstractRoutineAnalysis.getConditions() - Code Metrics - Inspection of "Support for file upload on analysis (pre) conditio..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — 2.x ( 8bc3ab...07088f )

by Ramon

created 2022-12-29 11:15 UTC

AbstractRoutineAnalysis.getConditions() A

↳ Parent: bika.lims.content.abstractroutineanalysis

Complexity

Conditions

Size

Total Lines	23
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	17
dl	0
loc	23
rs	9.0833
c	0
b	0
f	0
cc	5
nop	2

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

senaite / senaite.core

Push — 2.x ( 8bc3ab...07088f )

AbstractRoutineAnalysis.getConditions() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like