bika.lims.content.abstractroutineanalysis.AbstractRoutineAnalysis.getRequestUID() - Code Metrics - Inspection of "Clean some stale indexes from analysis_catalog (#1..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( d96fba...10f6ca )

by Ramon

created 2020-02-10 22:13 UTC

AbstractRoutineAnalysis.getRequestUID() A

↳ Parent: bika.lims.content.abstractroutineanalysis

Complexity

Conditions

Size

Total Lines	7
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

from datetime import timedelta

from AccessControl import ClassSecurityInfo
from Products.ATContentTypes.utils import DT2dt
from Products.ATContentTypes.utils import dt2DT
from Products.Archetypes.Field import BooleanField
from Products.Archetypes.Field import FixedPointField
from Products.Archetypes.Field import StringField
from Products.Archetypes.Schema import Schema
from Products.CMFCore.permissions import View
from zope.interface import alsoProvides
from zope.interface import implements
from zope.interface import noLongerProvides

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.clientawaremixin import ClientAwareMixin
from bika.lims.content.reflexrule import doReflexRuleAction
from bika.lims.interfaces import IAnalysis
from bika.lims.interfaces import ICancellable
from bika.lims.interfaces import IDynamicResultsRange
from bika.lims.interfaces import IInternalUse
from bika.lims.interfaces import IRoutineAnalysis
from bika.lims.interfaces.analysis import IRequestAnalysis
from bika.lims.workflow import getTransitionDate


# 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',
    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((
    IsReflexAnalysis,
    OriginalReflexedAnalysis,
    ReflexAnalysisOf,
    ReflexRuleAction,
    ReflexRuleActionsTriggered,
    ReflexRuleLocalID,
    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 getSamplePointUID(self):
        """Used to populate catalog values.
        """
        sample_point = self.getSamplePoint()
        if sample_point:
            return api.get_uid(sample_point)

    @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
        """
        results_range = self.getField("ResultsRange").get(self)
        # dynamic results range adapter
        adapter = IDynamicResultsRange(self, None)
        if adapter:
            results_range.update(adapter())
        return results_range

    @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
        """
        def is_dependent(analysis):
            calculation = analysis.getCalculation()
            if not calculation:
                return False

            services = calculation.getRawDependentServices()
            if not services:
                return False

            query = dict(UID=services, getKeyword=self.getKeyword())
            services = api.search(query, "bika_setup_catalog")
            return len(services) > 0

        siblings = self.getSiblings(retracted=retracted)
        return filter(lambda sib: is_dependent(sib), siblings)

    @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 []

        # If the calculation this analysis is bound does not have analysis
        # keywords (only interims), no need to go further
        service_uids = calc.getRawDependentServices()
        if len(service_uids) == 0:
            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 = map(api.get_uid, sibling.getDependents(retracted=True))
            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 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)

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

senaite / senaite.core

Push — master ( d96fba...10f6ca )

AbstractRoutineAnalysis.getRequestUID() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like