bika.lims.content.abstractroutineanalysis.AbstractRoutineAnalysis.getPrice() - Code Metrics - Inspection of "Cleanup the internal logic used for the creation o..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — 2.x ( d668b2...78a11b )

by Ramon

created 2022-10-06 21:00 UTC

AbstractRoutineAnalysis.getPrice() A

↳ Parent: bika.lims.content.abstractroutineanalysis

Complexity

Conditions

Size

Total Lines	10
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	10
rs	10
c	0
b	0
f	0
cc	3
nop	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-2021 by it's authors.
# Some rights reserved, see README and LICENSE.

import copy
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.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_type = self.getSampleType()
        if sample_type:
            return api.get_uid(sample_type)

    @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()

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

senaite / senaite.core

Push — 2.x ( d668b2...78a11b )

AbstractRoutineAnalysis.getPrice() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like