bika.lims.content.abstractroutineanalysis.AbstractRoutineAnalysis.setConditions() - Code Metrics - Inspection of "Allow to edit analysis (pre) conditions (#1935)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — 2.x ( 4667ca...3d98cb )

by Ramon

created 2022-02-15 09:53 UTC

AbstractRoutineAnalysis.setConditions() A

↳ Parent: bika.lims.content.abstractroutineanalysis

Complexity

Conditions

Size

Total Lines	16
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	10
dl	0
loc	16
rs	9.9
c	0
b	0
f	0
cc	3
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 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 FixedPointField
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 = FixedPointField(
    '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 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):
        """Returns the conditions of this analysis. These conditions are usually
        set on sample registration and are stored at sample level
        """
        sample = self.getRequest()
        service_uid = self.getRawAnalysisService()

        def is_valid(condition):
            uid = condition.get("uid")
            if api.is_uid(uid) and uid == service_uid:
                value = condition.get("value", None)
                if value:
                    return "title" in condition
            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)
        sample.setServiceConditions(other_conditions + conditions)


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 FixedPointField
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 = FixedPointField(
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 getDependents(self, with_retests=False, recursive=False):
299			"""
300			Returns a list of siblings who depend on us to calculate their result.
301			:param with_retests: If false, dependents with retests are dismissed
302			:param recursive: If true, returns all dependents recursively down
303			:type with_retests: bool
304			:return: Analyses the current analysis depends on
305			:rtype: list of IAnalysis
306			"""
307			def is_dependent(analysis):
308			# Never consider myself as dependent
309			if analysis.UID() == self.UID():
310			return False
311
312			# Never consider analyses from same service as dependents
313			self_service_uid = self.getRawAnalysisService()
314			if analysis.getRawAnalysisService() == self_service_uid:
315			return False
316
317			# Without calculation, no dependency relationship is possible
318			calculation = analysis.getCalculation()
319			if not calculation:
320			return False
321
322			# Calculation must have the service I belong to
323			services = calculation.getRawDependentServices()
324			return self_service_uid in services
325
326			request = self.getRequest()
327			if request.isPartition():
328			parent = request.getParentAnalysisRequest()
329			siblings = parent.getAnalyses(full_objects=True)
330			else:
331			siblings = self.getSiblings(with_retests=with_retests)
332
333			dependents = filter(lambda sib: is_dependent(sib), siblings)
334			if not recursive:
335			return dependents
336
337			# Return all dependents recursively
338			deps = dependents
339			for dep in dependents:
340			down_dependencies = dep.getDependents(with_retests=with_retests,
341			recursive=True)
342			deps.extend(down_dependencies)
343			return deps
344
345			@security.public
346			def getDependencies(self, with_retests=False, recursive=False):
347			"""
348			Return a list of siblings who we depend on to calculate our result.
349			:param with_retests: If false, siblings with retests are dismissed
350			:param recursive: If true, looks for dependencies recursively up
351			:type with_retests: bool
352			:return: Analyses the current analysis depends on
353			:rtype: list of IAnalysis
354			"""
355			calc = self.getCalculation()
356			if not calc:
357			return []
358
359			# If the calculation this analysis is bound does not have analysis
360			# keywords (only interims), no need to go further
361			service_uids = calc.getRawDependentServices()
362
363			# Ensure we exclude ourselves
364			service_uid = self.getRawAnalysisService()
365			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...
366			if len(service_uids) == 0:
367			return []
368
369			dependencies = []
370			for sibling in self.getSiblings(with_retests=with_retests):
371			# We get all analyses that depend on me, also if retracted (maybe
372			# I am one of those that are retracted!)
373			deps = map(api.get_uid, sibling.getDependents(with_retests=True))
374			if self.UID() in deps:
375			dependencies.append(sibling)
376			if recursive:
377			# Append the dependencies of this dependency
378			up_deps = sibling.getDependencies(with_retests=with_retests,
379			recursive=True)
380			dependencies.extend(up_deps)
381
382			# Exclude analyses of same service as me to prevent max recursion depth
383			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...
384			dependencies)
385
386			@security.public
387			def getPrioritySortkey(self):
388			"""
389			Returns the key that will be used to sort the current Analysis, from
390			most prioritary to less prioritary.
391			:return: string used for sorting
392			"""
393			analysis_request = self.getRequest()
394			if analysis_request is None:
395			return None
396			ar_sort_key = analysis_request.getPrioritySortkey()
397			ar_id = analysis_request.getId().lower()
398			title = sortable_title(self)
399			if callable(title):
400			title = title()
401			return '{}.{}.{}'.format(ar_sort_key, ar_id, title)
402
403			@security.public
404			def getHidden(self):
405			""" Returns whether if the analysis must be displayed in results
406			reports or not, as well as in analyses view when the user logged in
407			is a Client Contact.
408
409			If the value for the field HiddenManually is set to False, this function
410			will delegate the action to the method getAnalysisServiceSettings() from
411			the Analysis Request.
412
413			If the value for the field HiddenManually is set to True, this function
414			will return the value of the field Hidden.
415			:return: true or false
416			:rtype: bool
417			"""
418			if self.getHiddenManually():
419			return self.getField('Hidden').get(self)
420			request = self.getRequest()
421			if request:
422			service_uid = self.getServiceUID()
423			ar_settings = request.getAnalysisServiceSettings(service_uid)
424			return ar_settings.get('hidden', False)
425			return False
426
427			@security.public
428			def setHidden(self, hidden):
429			""" Sets if this analysis must be displayed or not in results report and
430			in manage analyses view if the user is a lab contact as well.
431
432			The value set by using this field will have priority over the visibility
433			criteria set at Analysis Request, Template or Profile levels (see
434			field AnalysisServiceSettings from Analysis Request. To achieve this
435			behavior, this setter also sets the value to HiddenManually to true.
436			:param hidden: true if the analysis must be hidden in report
437			:type hidden: bool
438			"""
439			self.setHiddenManually(True)
440			self.getField('Hidden').set(self, hidden)
441
442			@security.public
443			def setInternalUse(self, internal_use):
444			"""Applies the internal use of this Analysis. Analyses set for internal
445			use are not accessible to clients and are not visible in reports
446			"""
447			if internal_use:
448			alsoProvides(self, IInternalUse)
449			else:
450			noLongerProvides(self, IInternalUse)
451
452			def getConditions(self):
453			"""Returns the conditions of this analysis. These conditions are usually
454			set on sample registration and are stored at sample level
455			"""
456			sample = self.getRequest()
457			service_uid = self.getRawAnalysisService()
458
459			def is_valid(condition):
460			uid = condition.get("uid")
461			if api.is_uid(uid) and uid == service_uid:
462			value = condition.get("value", None)
463			if value:
464			return "title" in condition
465			return False
466
467			conditions = sample.getServiceConditions()
468			conditions = filter(is_valid, conditions)
469			return copy.deepcopy(conditions)
470
471			def setConditions(self, conditions):
472			"""Sets the conditions of this analysis. These conditions are usually
473			set on sample registration and are stored at sample level
474			"""
475			if not conditions:
476			conditions = []
477
478			sample = self.getRequest()
479			service_uid = self.getRawAnalysisService()
480			sample_conditions = sample.getServiceConditions()
481			sample_conditions = copy.deepcopy(sample_conditions)
482
483			# Keep the conditions from sample for analyses other than this one
484			other_conditions = filter(lambda c: c.get("uid") != service_uid,
485			sample_conditions)
486			sample.setServiceConditions(other_conditions + conditions)
487

senaite / senaite.core

Push — 2.x ( 4667ca...3d98cb )

AbstractRoutineAnalysis.setConditions() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like