bika.lims.content.abstractroutineanalysis.AbstractRoutineAnalysis.isSampleReceived() - Code Metrics - Inspection of "Sample edit form - some selection widgets empty (#..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f42093...226b52 )

by Ramon

created 2019-11-27 13:57 UTC

AbstractRoutineAnalysis.isSampleReceived() A

↳ Parent: bika.lims.content.abstractroutineanalysis

Complexity

Conditions

Size

Total Lines	6
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	1
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 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.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 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.Field import StringField
from Products.Archetypes.Schema import Schema
from Products.ATContentTypes.utils import DT2dt
from Products.ATContentTypes.utils import dt2DT
from Products.CMFCore.permissions import View
from zope.interface import implements


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

senaite / senaite.core

Push — master ( f42093...226b52 )

AbstractRoutineAnalysis.isSampleReceived() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like