bika.lims.content.abstractroutineanalysis.AbstractRoutineAnalysis.guard_to_be_preserved() - Code Metrics - Inspection of "Remove guard_to_be_preserved from analysis (#1099)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e317ea...56d09e )

by Ramon

created 2018-11-23 21:41 UTC

bika.lims.content.abstractroutineanalysis.AbstractRoutineAnalysis.guard_to_be_preserved() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	9
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	9
rs	10
c	0
b	0
f	0
cc	1
nop	1

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE
#
# Copyright 2018 by it's authors.
# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.

from AccessControl import ClassSecurityInfo
from datetime import timedelta
from Products.Archetypes.Field import BooleanField, FixedPointField, \
    StringField
from Products.Archetypes.Schema import Schema
from Products.ATContentTypes.utils import DT2dt, dt2DT
from bika.lims import api
from bika.lims import bikaMessageFactory as _
from bika.lims.browser.fields import UIDReferenceField
from bika.lims.browser.widgets import DecimalWidget
from bika.lims.catalog.indexers.baseanalysis import sortable_title
from bika.lims.content.abstractanalysis import AbstractAnalysis
from bika.lims.content.abstractanalysis import schema
from bika.lims.content.analysisspec import ResultsRangeDict
from bika.lims.content.reflexrule import doReflexRuleAction
from bika.lims.interfaces import IAnalysis, IRoutineAnalysis
from bika.lims.interfaces.analysis import IRequestAnalysis
from bika.lims.workflow import doActionFor
from bika.lims.workflow import getTransitionDate
from bika.lims.workflow import skip
from bika.lims.workflow import wasTransitionPerformed
from zope.interface import implements

# The physical sample partition linked to the Analysis.
SamplePartition = UIDReferenceField(
    'SamplePartition',
    required=0,
    allowed_types=('SamplePartition',)
)

# True if the analysis is created by a reflex rule
IsReflexAnalysis = BooleanField(
    'IsReflexAnalysis',
    default=False,
    required=0
)

# This field contains the original analysis which was reflected
OriginalReflexedAnalysis = UIDReferenceField(
    'OriginalReflexedAnalysis',
    required=0,
    allowed_types=('Analysis',)
)

# This field contains the analysis which has been reflected following
# a reflex rule
ReflexAnalysisOf = UIDReferenceField(
    'ReflexAnalysisOf',
    required=0,
    allowed_types=('Analysis',)
)

# Which is the Reflex Rule action that has created this analysis
ReflexRuleAction = StringField(
    'ReflexRuleAction',
    required=0,
    default=0
)

# Which is the 'local_id' inside the reflex rule
ReflexRuleLocalID = StringField(
    'ReflexRuleLocalID',
    required=0,
    default=0
)

# Reflex rule triggered actions which the current analysis is responsible for.
# Separated by '|'
ReflexRuleActionsTriggered = StringField(
    'ReflexRuleActionsTriggered',
    required=0,
    default=''
)

# The actual uncertainty for this analysis' result, populated when the result
# is submitted.
Uncertainty = FixedPointField(
    'Uncertainty',
    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((
    IsReflexAnalysis,
    OriginalReflexedAnalysis,
    ReflexAnalysisOf,
    ReflexRuleAction,
    ReflexRuleActionsTriggered,
    ReflexRuleLocalID,
    SamplePartition,
    Uncertainty,
    HiddenManually,
))


class AbstractRoutineAnalysis(AbstractAnalysis):
    implements(IAnalysis, IRequestAnalysis, IRoutineAnalysis)
    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()

    @security.public
    def getClientTitle(self):
        """Used to populate catalog values.
        Returns the Title of the client for this analysis' AR.
        """
        request = self.getRequest()
        if request:
            client = request.getClient()
            if client:
                return client.Title()

    @security.public
    def getClientUID(self):
        """Used to populate catalog values.
        Returns the UID of the client for this analysis' AR.
        """
        request = self.getRequest()
        if request:
            client = request.getClient()
            if client:
                return client.UID()

    @security.public
    def getClientURL(self):
        """This method is used to populate catalog values
        Returns the URL of the client for this analysis' AR.
        """
        request = self.getRequest()
        if request:
            client = request.getClient()
            if client:
                return client.absolute_url_path()

    @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(instance):
        """Returns whether if the Analysis Request this analysis comes from has
        been received or not
        """
        return instance.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 getSamplePartitionUID(self):
        part = self.getSamplePartition()
        if part:
            return part.UID()

    @security.public
    def getSamplePointUID(self):
        """Used to populate catalog values.
        """
        sample = self.getSample()
        if sample:
            samplepoint = sample.getSamplePoint()
            if samplepoint:
                return samplepoint.UID()

    @security.public
    def getSamplePartitionID(self):
        """Used to populate catalog values.
        Returns the sample partition ID
        """
        partition = self.getSamplePartition()
        if partition:
            return partition.getId()
        return ''

    @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
        return dt2DT(DT2dt(start) + timedelta(minutes=api.to_minutes(**tat)))

    @security.public
    def getSampleTypeUID(self):
        """Used to populate catalog values.
        """
        sample = self.getSample()
        if sample:
            sampletype = sample.getSampleType()
            if sampletype:
                return sampletype.UID()

    @security.public
    def getBatchUID(self):
        """This method is used to populate catalog values
        """
        request = self.getRequest()
        if request:
            return request.getBatchUID()

    @security.public
    def getAnalysisRequestPrintStatus(self):
        """This method is used to populate catalog values
        """
        request = self.getRequest()
        if request:
            return request.getPrinted()

    @security.public
    def getResultsRange(self):
        """Returns the valid result range for this routine analysis based on the
        results ranges defined in the Analysis Request this routine analysis is
        assigned to.

        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
        """
        specs = ResultsRangeDict()
        analysis_request = self.getRequest()
        if not analysis_request:
            return specs

        keyword = self.getKeyword()
        ar_ranges = analysis_request.getResultsRange()
        # Get the result range that corresponds to this specific analysis
        an_range = [rr for rr in ar_ranges if rr.get('keyword', '') == keyword]
        return an_range and an_range[0] or specs

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

    @security.public
    def getDependents(self, retracted=False):
        """
        Returns a list of siblings who depend on us to calculate their result.
        :param retracted: If false, retracted/rejected dependents are dismissed
        :type retracted: bool
        :return: Analyses the current analysis depends on
        :rtype: list of IAnalysis
        """
        dependents = []
        for sibling in self.getSiblings(retracted=retracted):
            calculation = sibling.getCalculation()
            if not calculation:
                continue
            depservices = calculation.getDependentServices()
            dep_keywords = [dep.getKeyword() for dep in depservices]
            if self.getKeyword() in dep_keywords:
                dependents.append(sibling)
        return dependents

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

        dependencies = []
        for sibling in self.getSiblings(retracted=retracted):
            # We get all analyses that depend on me, also if retracted (maybe
            # I am one of those that are retracted!)
            deps = sibling.getDependents(retracted=True)
            deps = [dep.UID() for dep in deps]
            if self.UID() in deps:
                dependencies.append(sibling)
        return 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 setReflexAnalysisOf(self, analysis):
        """Sets the analysis that has been reflexed in order to create this
        one, but if the analysis is the same as self, do nothing.
        :param analysis: an analysis object or UID
        """
        if not analysis or analysis.UID() == self.UID():
            pass
        else:
            self.getField('ReflexAnalysisOf').set(self, analysis)

    @security.public
    def addReflexRuleActionsTriggered(self, text):
        """This function adds a new item to the string field
        ReflexRuleActionsTriggered. From the field: Reflex rule triggered
        actions from which the current analysis is responsible of. Separated
        by '|'
        :param text: is a str object with the format '<UID>.<rulename>' ->
        '123354.1'
        """
        old = self.getReflexRuleActionsTriggered()
        self.setReflexRuleActionsTriggered(old + text + '|')

    @security.public
    def getOriginalReflexedAnalysisUID(self):
        """
        Returns the uid of the original reflexed analysis.
        """
        original = self.getOriginalReflexedAnalysis()
        if original:
            return original.UID()
        return ''

    @security.private
    def _reflex_rule_process(self, wf_action):
        """This function does all the reflex rule process.
        :param wf_action: is a string containing the workflow action triggered
        """
        # Check out if the analysis has any reflex rule bound to it.
        # First we have get the analysis' method because the Reflex Rule
        # objects are related to a method.
        a_method = self.getMethod()
        if not a_method:
            return
        # After getting the analysis' method we have to get all Reflex Rules
        # related to that method.
        all_rrs = a_method.getBackReferences('ReflexRuleMethod')
        if not all_rrs:
            return
        # Once we have all the Reflex Rules with the same method as the
        # analysis has, it is time to get the rules that are bound to the
        # same analysis service that is using the analysis.
        for rule in all_rrs:
            if not api.is_active(rule):
                continue
            # Getting the rules to be done from the reflex rule taking
            # in consideration the analysis service, the result and
            # the state change
            action_row = rule.getActionReflexRules(self, wf_action)
            # Once we have the rules, the system has to execute its
            # instructions if the result has the expected result.
            doReflexRuleAction(self, action_row)


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE
4			#
5			# Copyright 2018 by it's authors.
6			# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.
7
8			from AccessControl import ClassSecurityInfo
9			from datetime import timedelta
10			from Products.Archetypes.Field import BooleanField, FixedPointField, \
11			StringField
12			from Products.Archetypes.Schema import Schema
13			from Products.ATContentTypes.utils import DT2dt, dt2DT
14			from bika.lims import api
15			from bika.lims import bikaMessageFactory as _
16			from bika.lims.browser.fields import UIDReferenceField
17			from bika.lims.browser.widgets import DecimalWidget
18			from bika.lims.catalog.indexers.baseanalysis import sortable_title
19			from bika.lims.content.abstractanalysis import AbstractAnalysis
20			from bika.lims.content.abstractanalysis import schema
21			from bika.lims.content.analysisspec import ResultsRangeDict
22			from bika.lims.content.reflexrule import doReflexRuleAction
23			from bika.lims.interfaces import IAnalysis, IRoutineAnalysis
24			from bika.lims.interfaces.analysis import IRequestAnalysis
25			from bika.lims.workflow import doActionFor
26			from bika.lims.workflow import getTransitionDate
27			from bika.lims.workflow import skip
28			from bika.lims.workflow import wasTransitionPerformed
29			from zope.interface import implements
30
31			# The physical sample partition linked to the Analysis.
32			SamplePartition = UIDReferenceField(
33			'SamplePartition',
34			required=0,
35			allowed_types=('SamplePartition',)
36			)
37
38			# True if the analysis is created by a reflex rule
39			IsReflexAnalysis = BooleanField(
40			'IsReflexAnalysis',
41			default=False,
42			required=0
43			)
44
45			# This field contains the original analysis which was reflected
46			OriginalReflexedAnalysis = UIDReferenceField(
47			'OriginalReflexedAnalysis',
48			required=0,
49			allowed_types=('Analysis',)
50			)
51
52			# This field contains the analysis which has been reflected following
53			# a reflex rule
54			ReflexAnalysisOf = UIDReferenceField(
55			'ReflexAnalysisOf',
56			required=0,
57			allowed_types=('Analysis',)
58			)
59
60			# Which is the Reflex Rule action that has created this analysis
61			ReflexRuleAction = StringField(
62			'ReflexRuleAction',
63			required=0,
64			default=0
65			)
66
67			# Which is the 'local_id' inside the reflex rule
68			ReflexRuleLocalID = StringField(
69			'ReflexRuleLocalID',
70			required=0,
71			default=0
72			)
73
74			# Reflex rule triggered actions which the current analysis is responsible for.
75			# Separated by '\|'
76			ReflexRuleActionsTriggered = StringField(
77			'ReflexRuleActionsTriggered',
78			required=0,
79			default=''
80			)
81
82			# The actual uncertainty for this analysis' result, populated when the result
83			# is submitted.
84			Uncertainty = FixedPointField(
85			'Uncertainty',
86			precision=10,
87			widget=DecimalWidget(
88			label=_("Uncertainty")
89			)
90			)
91			# This field keep track if the field hidden has been set manually or not. If
92			# this value is false, the system will assume the visibility of this analysis
93			# in results report will depend on the value set at AR, Profile or Template
94			# levels (see AnalysisServiceSettings fields in AR). If the value for this
95			# field is set to true, the system will assume the visibility of the analysis
96			# will only depend on the value set for the field Hidden (bool).
97			HiddenManually = BooleanField(
98			'HiddenManually',
99			default=False,
100			)
101
102
103			schema = schema.copy() + Schema((
104			IsReflexAnalysis,
105			OriginalReflexedAnalysis,
106			ReflexAnalysisOf,
107			ReflexRuleAction,
108			ReflexRuleActionsTriggered,
109			ReflexRuleLocalID,
110			SamplePartition,
111			Uncertainty,
112			HiddenManually,
113			))
114
115
116			class AbstractRoutineAnalysis(AbstractAnalysis):
117			implements(IAnalysis, IRequestAnalysis, IRoutineAnalysis)
118			security = ClassSecurityInfo()
119			displayContentsTab = False
120			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...
121
122			@security.public
123			def getRequest(self):
124			"""Returns the Analysis Request this analysis belongs to.
125			Delegates to self.aq_parent
126			"""
127			ar = self.aq_parent
128			return ar
129
130			@security.public
131			def getRequestID(self):
132			"""Used to populate catalog values.
133			Returns the ID of the parent analysis request.
134			"""
135			ar = self.getRequest()
136			if ar:
137			return ar.getId()
138
139			@security.public
140			def getRequestUID(self):
141			"""Returns the UID of the parent analysis request.
142			"""
143			ar = self.getRequest()
144			if ar:
145			return ar.UID()
146
147			@security.public
148			def getRequestURL(self):
149			"""Returns the url path of the Analysis Request object this analysis
150			belongs to. Returns None if there is no Request assigned.
151			:return: the Analysis Request URL path this analysis belongs to
152			:rtype: str
153			"""
154			request = self.getRequest()
155			if request:
156			return request.absolute_url_path()
157
158			@security.public
159			def getClientTitle(self):
160			"""Used to populate catalog values.
161			Returns the Title of the client for this analysis' AR.
162			"""
163			request = self.getRequest()
164			if request:
165			client = request.getClient()
166			if client:
167			return client.Title()
168
169			@security.public
170			def getClientUID(self):
171			"""Used to populate catalog values.
172			Returns the UID of the client for this analysis' AR.
173			"""
174			request = self.getRequest()
175			if request:
176			client = request.getClient()
177			if client:
178			return client.UID()
179
180			@security.public
181			def getClientURL(self):
182			"""This method is used to populate catalog values
183			Returns the URL of the client for this analysis' AR.
184			"""
185			request = self.getRequest()
186			if request:
187			client = request.getClient()
188			if client:
189			return client.absolute_url_path()
190
191			@security.public
192			def getClientOrderNumber(self):
193			"""Used to populate catalog values.
194			Returns the ClientOrderNumber of the associated AR
195			"""
196			request = self.getRequest()
197			if request:
198			return request.getClientOrderNumber()
199
200			@security.public
201			def getDateReceived(self):
202			"""Used to populate catalog values.
203			Returns the date the Analysis Request this analysis belongs to was
204			received. If the analysis was created after, then returns the date
205			the analysis was created.
206			"""
207			request = self.getRequest()
208			if request:
209			ar_date = request.getDateReceived()
210			if ar_date and self.created() > ar_date:
211			return self.created()
212			return ar_date
213			return None
214
215			@security.public
216			def isSampleReceived(instance):
217			"""Returns whether if the Analysis Request this analysis comes from has
218			been received or not
219			"""
220			return instance.getDateReceived() and True or False
221
222			@security.public
223			def getDatePublished(self):
224			"""Used to populate catalog values.
225			Returns the date on which the "publish" transition was invoked on this
226			analysis.
227			"""
228			return getTransitionDate(self, 'publish', return_as_datetime=True)
229
230			@security.public
231			def getDateSampled(self):
232			"""Returns the date when the Sample was Sampled
233			"""
234			request = self.getRequest()
235			if request:
236			return request.getDateSampled()
237			return None
238
239			@security.public
240			def isSampleSampled(instance):
241			"""Returns whether if the Analysis Request this analysis comes from has
242			been received or not
243			"""
244			return instance.getDateSampled() and True or False
245
246			@security.public
247			def getStartProcessDate(self):
248			"""Returns the date time when the analysis request the analysis belongs
249			to was received. If the analysis request hasn't yet been received,
250			returns None
251			Overrides getStartProcessDateTime from the base class
252			:return: Date time when the analysis is ready to be processed.
253			:rtype: DateTime
254			"""
255			return self.getDateReceived()
256
257			@security.public
258			def getSamplePartitionUID(self):
259			part = self.getSamplePartition()
260			if part:
261			return part.UID()
262
263			@security.public
264			def getSamplePointUID(self):
265			"""Used to populate catalog values.
266			"""
267			sample = self.getSample()
268			if sample:
269			samplepoint = sample.getSamplePoint()
270			if samplepoint:
271			return samplepoint.UID()
272
273			@security.public
274			def getSamplePartitionID(self):
275			"""Used to populate catalog values.
276			Returns the sample partition ID
277			"""
278			partition = self.getSamplePartition()
279			if partition:
280			return partition.getId()
281			return ''
282
283			@security.public
284			def getDueDate(self):
285			"""Used to populate getDueDate index and metadata.
286			This calculates the difference between the time the analysis processing
287			started and the maximum turnaround time. If the analysis has no
288			turnaround time set or is not yet ready for proces, returns None
289			"""
290			tat = self.getMaxTimeAllowed()
291			if not tat:
292			return None
293			start = self.getStartProcessDate()
294			if not start:
295			return None
296			return dt2DT(DT2dt(start) + timedelta(minutes=api.to_minutes(**tat)))
297
298			@security.public
299			def getSampleTypeUID(self):
300			"""Used to populate catalog values.
301			"""
302			sample = self.getSample()
303			if sample:
304			sampletype = sample.getSampleType()
305			if sampletype:
306			return sampletype.UID()
307
308			@security.public
309			def getBatchUID(self):
310			"""This method is used to populate catalog values
311			"""
312			request = self.getRequest()
313			if request:
314			return request.getBatchUID()
315
316			@security.public
317			def getAnalysisRequestPrintStatus(self):
318			"""This method is used to populate catalog values
319			"""
320			request = self.getRequest()
321			if request:
322			return request.getPrinted()
323
324			@security.public
325			def getResultsRange(self):
326			"""Returns the valid result range for this routine analysis based on the
327			results ranges defined in the Analysis Request this routine analysis is
328			assigned to.
329
330			A routine analysis will be considered out of range if it result falls
331			out of the range defined in "min" and "max". If there are values set for
332			"warn_min" and "warn_max", these are used to compute the shoulders in
333			both ends of the range. Thus, an analysis can be out of range, but be
334			within shoulders still.
335			:return: A dictionary with keys "min", "max", "warn_min" and "warn_max"
336			:rtype: dict
337			"""
338			specs = ResultsRangeDict()
339			analysis_request = self.getRequest()
340			if not analysis_request:
341			return specs
342
343			keyword = self.getKeyword()
344			ar_ranges = analysis_request.getResultsRange()
345			# Get the result range that corresponds to this specific analysis
346			an_range = [rr for rr in ar_ranges if rr.get('keyword', '') == keyword]
347			return an_range and an_range[0] or specs
348
349			@security.public
350			def getSiblings(self, retracted=False):
351			"""
352			Return the siblings analyses, using the parent to which the current
353			analysis belongs to as the source
354			:param retracted: If false, retracted/rejected siblings are dismissed
355			:type retracted: bool
356			:return: list of siblings for this analysis
357			:rtype: list of IAnalysis
358			"""
359			raise NotImplementedError("getSiblings is not implemented.")
360
361			@security.public
362			def getDependents(self, retracted=False):
363			"""
364			Returns a list of siblings who depend on us to calculate their result.
365			:param retracted: If false, retracted/rejected dependents are dismissed
366			:type retracted: bool
367			:return: Analyses the current analysis depends on
368			:rtype: list of IAnalysis
369			"""
370			dependents = []
371			for sibling in self.getSiblings(retracted=retracted):
372			calculation = sibling.getCalculation()
373			if not calculation:
374			continue
375			depservices = calculation.getDependentServices()
376			dep_keywords = [dep.getKeyword() for dep in depservices]
377			if self.getKeyword() in dep_keywords:
378			dependents.append(sibling)
379			return dependents
380
381			@security.public
382			def getDependencies(self, retracted=False):
383			"""
384			Return a list of siblings who we depend on to calculate our result.
385			:param retracted: If false retracted/rejected dependencies are dismissed
386			:type retracted: bool
387			:return: Analyses the current analysis depends on
388			:rtype: list of IAnalysis
389			"""
390			calc = self.getCalculation()
391			if not calc:
392			return []
393
394			dependencies = []
395			for sibling in self.getSiblings(retracted=retracted):
396			# We get all analyses that depend on me, also if retracted (maybe
397			# I am one of those that are retracted!)
398			deps = sibling.getDependents(retracted=True)
399			deps = [dep.UID() for dep in deps]
400			if self.UID() in deps:
401			dependencies.append(sibling)
402			return dependencies
403
404			@security.public
405			def getPrioritySortkey(self):
406			"""
407			Returns the key that will be used to sort the current Analysis, from
408			most prioritary to less prioritary.
409			:return: string used for sorting
410			"""
411			analysis_request = self.getRequest()
412			if analysis_request is None:
413			return None
414			ar_sort_key = analysis_request.getPrioritySortkey()
415			ar_id = analysis_request.getId().lower()
416			title = sortable_title(self)
417			if callable(title):
418			title = title()
419			return '{}.{}.{}'.format(ar_sort_key, ar_id, title)
420
421			@security.public
422			def getHidden(self):
423			""" Returns whether if the analysis must be displayed in results
424			reports or not, as well as in analyses view when the user logged in
425			is a Client Contact.
426
427			If the value for the field HiddenManually is set to False, this function
428			will delegate the action to the method getAnalysisServiceSettings() from
429			the Analysis Request.
430
431			If the value for the field HiddenManually is set to True, this function
432			will return the value of the field Hidden.
433			:return: true or false
434			:rtype: bool
435			"""
436			if self.getHiddenManually():
437			return self.getField('Hidden').get(self)
438			request = self.getRequest()
439			if request:
440			service_uid = self.getServiceUID()
441			ar_settings = request.getAnalysisServiceSettings(service_uid)
442			return ar_settings.get('hidden', False)
443			return False
444
445			@security.public
446			def setHidden(self, hidden):
447			""" Sets if this analysis must be displayed or not in results report and
448			in manage analyses view if the user is a lab contact as well.
449
450			The value set by using this field will have priority over the visibility
451			criteria set at Analysis Request, Template or Profile levels (see
452			field AnalysisServiceSettings from Analysis Request. To achieve this
453			behavior, this setter also sets the value to HiddenManually to true.
454			:param hidden: true if the analysis must be hidden in report
455			:type hidden: bool
456			"""
457			self.setHiddenManually(True)
458			self.getField('Hidden').set(self, hidden)
459
460			@security.public
461			def setReflexAnalysisOf(self, analysis):
462			"""Sets the analysis that has been reflexed in order to create this
463			one, but if the analysis is the same as self, do nothing.
464			:param analysis: an analysis object or UID
465			"""
466			if not analysis or analysis.UID() == self.UID():
467			pass
468			else:
469			self.getField('ReflexAnalysisOf').set(self, analysis)
470
471			@security.public
472			def addReflexRuleActionsTriggered(self, text):
473			"""This function adds a new item to the string field
474			ReflexRuleActionsTriggered. From the field: Reflex rule triggered
475			actions from which the current analysis is responsible of. Separated
476			by '\|'
477			:param text: is a str object with the format '<UID>.<rulename>' ->
478			'123354.1'
479			"""
480			old = self.getReflexRuleActionsTriggered()
481			self.setReflexRuleActionsTriggered(old + text + '\|')
482
483			@security.public
484			def getOriginalReflexedAnalysisUID(self):
485			"""
486			Returns the uid of the original reflexed analysis.
487			"""
488			original = self.getOriginalReflexedAnalysis()
489			if original:
490			return original.UID()
491			return ''
492
493			@security.private
494			def _reflex_rule_process(self, wf_action):
495			"""This function does all the reflex rule process.
496			:param wf_action: is a string containing the workflow action triggered
497			"""
498			# Check out if the analysis has any reflex rule bound to it.
499			# First we have get the analysis' method because the Reflex Rule
500			# objects are related to a method.
501			a_method = self.getMethod()
502			if not a_method:
503			return
504			# After getting the analysis' method we have to get all Reflex Rules
505			# related to that method.
506			all_rrs = a_method.getBackReferences('ReflexRuleMethod')
507			if not all_rrs:
508			return
509			# Once we have all the Reflex Rules with the same method as the
510			# analysis has, it is time to get the rules that are bound to the
511			# same analysis service that is using the analysis.
512			for rule in all_rrs:
513			if not api.is_active(rule):
514			continue
515			# Getting the rules to be done from the reflex rule taking
516			# in consideration the analysis service, the result and
517			# the state change
518			action_row = rule.getActionReflexRules(self, wf_action)
519			# Once we have the rules, the system has to execute its
520			# instructions if the result has the expected result.
521			doReflexRuleAction(self, action_row)
522

senaite / senaite.core

Push — master ( e317ea...56d09e )

bika.lims.content.abstractroutineanalysis.AbstractRoutineAnalysis.guard_to_be_preserved() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like