build.bika.lims.content.abstractroutineanalysis.AbstractRoutineAnalysis.getDueDate() - Code Metrics - Inspection of "Consider laboratory workdays only for the late ana..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 69376c...7602ce )

by Jordi

created 2019-04-28 21:58 UTC

AbstractRoutineAnalysis.getDueDate() B

↳ Parent: build.bika.lims.content.abstractroutineanalysis

Complexity

Conditions

Size

Total Lines	46
Code Lines	24

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	24
dl	0
loc	46
rs	7.904
c	0
b	0
f	0
cc	7
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.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


# TODO Remove in >v1.3.0 - This is kept for backwards-compatibility
# 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',
    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,
    SamplePartition,
    Uncertainty,
    HiddenManually,
))


class AbstractRoutineAnalysis(AbstractAnalysis):
    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()

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

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

senaite / senaite.core

Push — master ( 69376c...7602ce )

AbstractRoutineAnalysis.getDueDate() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like