bika.lims.workflow.analysis.events.check_all_verified() - Code Metrics - Inspection of "Improve performance of analysis verification (#264..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — 2.x ( 9538a4...3858ed )

by Jordi

created 2024-11-26 11:07 UTC

check_all_verified() A

↳ Parent: bika.lims.workflow.analysis.events

Complexity

Conditions

Size

Total Lines	35
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	13
dl	0
loc	35
rs	9.75
c	0
b	0
f	0
cc	4
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-2024 by it's authors.
# Some rights reserved, see README and LICENSE.

from bika.lims import api
from bika.lims.interfaces import IDuplicateAnalysis
from bika.lims.interfaces import IRejected
from bika.lims.interfaces import IRetracted
from bika.lims.interfaces import ISubmitted
from bika.lims.interfaces import IVerified
from bika.lims.interfaces.analysis import IRequestAnalysis
from bika.lims.utils.analysis import create_retest
from bika.lims.workflow import doActionFor
from bika.lims.workflow.analysis import STATE_REJECTED
from bika.lims.workflow.analysis import STATE_RETRACTED
from DateTime import DateTime
from zope.interface import alsoProvides


def after_assign(analysis):
    """Function triggered after an 'assign' transition for the analysis passed
    in is performed.
    """
    reindex_request(analysis)


def before_unassign(analysis):
    """Function triggered before 'unassign' transition takes place
    """
    worksheet = analysis.getWorksheet()
    if not worksheet:
        return

    # Removal of a routine analysis causes the removal of their duplicates
    for dup in worksheet.get_duplicates_for(analysis):
        doActionFor(dup, "unassign")


def before_reject(analysis):
    """Function triggered before 'unassign' transition takes place
    """
    worksheet = analysis.getWorksheet()
    if not worksheet:
        return

    # Rejection of a routine analysis causes the removal of their duplicates
    for dup in worksheet.get_duplicates_for(analysis):
        doActionFor(dup, "unassign")


def after_retest(analysis):
    """Function triggered before 'retest' transition takes place. Creates a
    copy of the current analysis
    """
    # When an analysis is retested, it automatically transitions to verified,
    # so we need to mark the analysis as such
    alsoProvides(analysis, IVerified)

    def verify_and_retest(relative):
        if not ISubmitted.providedBy(relative):
            # Result not yet submitted, no need to create a retest
            return

        # Apply the transition manually, but only if analysis can be verified
        doActionFor(relative, "verify")

        # Create the retest
        create_retest(relative)

    # Retest and auto-verify relatives, from bottom to top
    relatives = list(reversed(analysis.getDependents(recursive=True)))
    relatives.extend(analysis.getDependencies(recursive=True))
    map(verify_and_retest, relatives)

    # Create the retest
    create_retest(analysis)

    # Try to rollback the Analysis Request
    if IRequestAnalysis.providedBy(analysis):
        doActionFor(analysis.getRequest(), "rollback_to_receive")
        reindex_request(analysis)


def after_unassign(analysis):
    """Function triggered after an 'unassign' transition for the analysis passed
    in is performed.
    """
    # Remove from the worksheet
    remove_analysis_from_worksheet(analysis)
    # Reindex the Analysis Request
    reindex_request(analysis)


def after_cancel(analysis):
    """Function triggered after a "cancel" transition is performed. Removes the
    cancelled analysis from the worksheet, if any.
    """
    # Remove from the worksheet
    remove_analysis_from_worksheet(analysis)


def after_reinstate(analysis):
    """Function triggered after a "reinstate" transition is performed.
    """
    pass


def after_submit(analysis):
    """Method triggered after a 'submit' transition for the analysis passed in
    is performed. Promotes the submit transition to the Worksheet to which the
    analysis belongs to. Note that for the worksheet there is already a guard
    that assures the transition to the worksheet will only be performed if all
    analyses within the worksheet have already been transitioned.
    This function is called automatically by
    bika.lims.workfow.AfterTransitionEventHandler
    """
    # Ensure there is a Result Capture Date even if the result was set
    # automatically on creation because of a "DefaultResult"
    if not analysis.getResultCaptureDate():
        analysis.setResultCaptureDate(DateTime())

    # Mark this analysis as ISubmitted
    alsoProvides(analysis, ISubmitted)

    # Promote to analyses this analysis depends on
    promote_to_dependencies(analysis, "submit")

    # Promote transition to worksheet
    ws = analysis.getWorksheet()
    if ws:
        doActionFor(ws, "submit")
        ws.reindexObject()

    # Promote transition to Analysis Request
    if IRequestAnalysis.providedBy(analysis):
        doActionFor(analysis.getRequest(), 'submit')
        reindex_request(analysis)


def after_retract(analysis):
    """Function triggered after a 'retract' transition for the analysis passed
    in is performed. The analysis transitions to "retracted" state and a new
    copy of the analysis is created. The copy initial state is "unassigned",
    unless the the retracted analysis was assigned to a worksheet. In such
    case, the copy is transitioned to 'assigned' state too
    """
    # Mark this analysis as IRetracted
    alsoProvides(analysis, IRetracted)

    # Ignore attachments of this analysis in results report
    for attachment in analysis.getAttachment():
        attachment.setRenderInReport(False)

    # Retract our dependents (analyses that depend on this analysis)
    cascade_to_dependents(analysis, "retract")

    # Retract our dependencies (analyses this analysis depends on)
    promote_to_dependencies(analysis, "retract")

    # Create the retest
    create_retest(analysis)

    # Try to rollback the Analysis Request
    if IRequestAnalysis.providedBy(analysis):
        doActionFor(analysis.getRequest(), "rollback_to_receive")
        reindex_request(analysis)


def after_reject(analysis):
    """Function triggered after the "reject" transition for the analysis passed
    in is performed."""
    # Mark this analysis with IRejected
    alsoProvides(analysis, IRejected)

    # Remove from the worksheet
    remove_analysis_from_worksheet(analysis)

    # Ignore attachments of this analysis in results report
    for attachment in analysis.getAttachment():
        attachment.setRenderInReport(False)

    # Reject our dependents (analyses that depend on this analysis)
    cascade_to_dependents(analysis, "reject")

    if IRequestAnalysis.providedBy(analysis):
        # Try verify (for when remaining analyses are in 'verified')
        doActionFor(analysis.getRequest(), "verify")

        # Try submit (remaining analyses are in 'to_be_verified')
        doActionFor(analysis.getRequest(), "submit")

        # Try rollback (no remaining analyses or some not submitted)
        doActionFor(analysis.getRequest(), "rollback_to_receive")
        reindex_request(analysis)


def after_verify(analysis):
    """
    Method triggered after a 'verify' transition for the analysis passed in
    is performed. Promotes the transition to the Analysis Request and to
    Worksheet (if the analysis is assigned to any)
    This function is called automatically by
    bika.lims.workfow.AfterTransitionEventHandler
    """
    # Mark this analysis as IVerified
    alsoProvides(analysis, IVerified)

    # Promote to analyses this analysis depends on
    promote_to_dependencies(analysis, "verify")

    # Promote transition to worksheet
    ws = analysis.getWorksheet()
    if ws:
        doActionFor(ws, "verify")
        ws.reindexObject()

    # Promote transition to Analysis Request if Sample auto-verify is enabled
    if IRequestAnalysis.providedBy(analysis) and check_all_verified(analysis):
        setup = api.get_setup()
        if setup.getAutoVerifySamples():
            doActionFor(analysis.getRequest(), "verify")

        # Reindex the sample (and ancestors) this analysis belongs to
        reindex_request(analysis)


def check_all_verified(analysis):
    """Checks if all analyses are verified

    NOTE: This check is provided solely for performance reasons of the `verify`
    transition, because it is a less expensive calculation than executing the
    `doActionFor` method on the sample for each verified analysis.

    The worst case that can happen is that the sample does not get
    automatically verified and needs to be transitioned manually.

    :param analysis: The current verified analysis
    :returns: True if all other routine analyses of the sample are verified
    """

    parent = api.get_parent(analysis)
    sample = analysis.getRequest()
    uid = api.get_uid(analysis)

    def is_valid(an):
        state = api.get_review_status(an)
        return state not in [STATE_REJECTED, STATE_RETRACTED]

    # get all *valid* analyses of the sample
    analyses = filter(is_valid, sample.getAnalyses())
    # get all *verified* analyses of the sample
    verified = sample.getAnalyses(object_provides=IVerified.__identifier__)

    # NOTE: We remove the current processed routine analysis (if not a WS
    #       duplicate/reference analysis), because it is either not yet
    #       verified or processed already in multi-verify scenarios.
    if sample == parent:
        analyses = filter(lambda x: api.get_uid(x) != uid, analyses)
        verified = filter(lambda x: api.get_uid(x) != uid, verified)

    return len(analyses) == len(verified)


def after_publish(analysis):
    """Function triggered after a "publish" transition is performed.
    """
    pass


# TODO Workflow - Analysis - revisit reindexing of ancestors
def reindex_request(analysis, idxs=None):
    """Reindex the Analysis Request the analysis belongs to, as well as the
    ancestors recursively
    """
    if not IRequestAnalysis.providedBy(analysis) or \
            IDuplicateAnalysis.providedBy(analysis):
        # Analysis not directly bound to an Analysis Request. Do nothing
        return

    request = analysis.getRequest()
    ancestors = [request] + request.getAncestors(all_ancestors=True)
    for ancestor in ancestors:
        ancestor.reindexObject()


def remove_analysis_from_worksheet(analysis):
    """Removes the analysis passed in from the worksheet, if assigned to any
    """
    worksheet = analysis.getWorksheet()
    if not worksheet:
        return

    analyses = filter(lambda an: an != analysis, worksheet.getAnalyses())
    worksheet.setAnalyses(analyses)
    worksheet.purgeLayout()
    if analyses:
        # Maybe this analysis was the only one that was not yet submitted or
        # verified, so try to submit or verify the Worksheet to be aligned
        # with the current states of the analyses it contains.
        doActionFor(worksheet, "submit")
        doActionFor(worksheet, "verify")
    else:
        # We've removed all analyses. Rollback to "open"
        doActionFor(worksheet, "rollback_to_open")

    # Reindex the Worksheet
    worksheet.reindexObject()


def cascade_to_dependents(analysis, transition_id):
    """Cascades the transition to dependent analyses (those that depend on the
    analysis passed in), if any
    """
    for dependent in analysis.getDependents():
        doActionFor(dependent, transition_id)


def promote_to_dependencies(analysis, transition_id):
    """Promotes the transition to the analyses this analysis depends on
    (dependencies), if any
    """
    for dependency in analysis.getDependencies():
        doActionFor(dependency, transition_id)


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-2024 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			from bika.lims import api
22			from bika.lims.interfaces import IDuplicateAnalysis
23			from bika.lims.interfaces import IRejected
24			from bika.lims.interfaces import IRetracted
25			from bika.lims.interfaces import ISubmitted
26			from bika.lims.interfaces import IVerified
27			from bika.lims.interfaces.analysis import IRequestAnalysis
28			from bika.lims.utils.analysis import create_retest
29			from bika.lims.workflow import doActionFor
30			from bika.lims.workflow.analysis import STATE_REJECTED
31			from bika.lims.workflow.analysis import STATE_RETRACTED
32			from DateTime import DateTime
33			from zope.interface import alsoProvides
34
35
36			def after_assign(analysis):
37			"""Function triggered after an 'assign' transition for the analysis passed
38			in is performed.
39			"""
40			reindex_request(analysis)
41
42
43			def before_unassign(analysis):
44			"""Function triggered before 'unassign' transition takes place
45			"""
46			worksheet = analysis.getWorksheet()
47			if not worksheet:
48			return
49
50			# Removal of a routine analysis causes the removal of their duplicates
51			for dup in worksheet.get_duplicates_for(analysis):
52			doActionFor(dup, "unassign")
53
54
55			def before_reject(analysis):
56			"""Function triggered before 'unassign' transition takes place
57			"""
58			worksheet = analysis.getWorksheet()
59			if not worksheet:
60			return
61
62			# Rejection of a routine analysis causes the removal of their duplicates
63			for dup in worksheet.get_duplicates_for(analysis):
64			doActionFor(dup, "unassign")
65
66
67			def after_retest(analysis):
68			"""Function triggered before 'retest' transition takes place. Creates a
69			copy of the current analysis
70			"""
71			# When an analysis is retested, it automatically transitions to verified,
72			# so we need to mark the analysis as such
73			alsoProvides(analysis, IVerified)
74
75			def verify_and_retest(relative):
76			if not ISubmitted.providedBy(relative):
77			# Result not yet submitted, no need to create a retest
78			return
79
80			# Apply the transition manually, but only if analysis can be verified
81			doActionFor(relative, "verify")
82
83			# Create the retest
84			create_retest(relative)
85
86			# Retest and auto-verify relatives, from bottom to top
87			relatives = list(reversed(analysis.getDependents(recursive=True)))
88			relatives.extend(analysis.getDependencies(recursive=True))
89			map(verify_and_retest, relatives)
90
91			# Create the retest
92			create_retest(analysis)
93
94			# Try to rollback the Analysis Request
95			if IRequestAnalysis.providedBy(analysis):
96			doActionFor(analysis.getRequest(), "rollback_to_receive")
97			reindex_request(analysis)
98
99
100			def after_unassign(analysis):
101			"""Function triggered after an 'unassign' transition for the analysis passed
102			in is performed.
103			"""
104			# Remove from the worksheet
105			remove_analysis_from_worksheet(analysis)
106			# Reindex the Analysis Request
107			reindex_request(analysis)
108
109
110			def after_cancel(analysis):
111			"""Function triggered after a "cancel" transition is performed. Removes the
112			cancelled analysis from the worksheet, if any.
113			"""
114			# Remove from the worksheet
115			remove_analysis_from_worksheet(analysis)
116
117
118			def after_reinstate(analysis):
119			"""Function triggered after a "reinstate" transition is performed.
120			"""
121			pass
122
123
124			def after_submit(analysis):
125			"""Method triggered after a 'submit' transition for the analysis passed in
126			is performed. Promotes the submit transition to the Worksheet to which the
127			analysis belongs to. Note that for the worksheet there is already a guard
128			that assures the transition to the worksheet will only be performed if all
129			analyses within the worksheet have already been transitioned.
130			This function is called automatically by
131			bika.lims.workfow.AfterTransitionEventHandler
132			"""
133			# Ensure there is a Result Capture Date even if the result was set
134			# automatically on creation because of a "DefaultResult"
135			if not analysis.getResultCaptureDate():
136			analysis.setResultCaptureDate(DateTime())
137
138			# Mark this analysis as ISubmitted
139			alsoProvides(analysis, ISubmitted)
140
141			# Promote to analyses this analysis depends on
142			promote_to_dependencies(analysis, "submit")
143
144			# Promote transition to worksheet
145			ws = analysis.getWorksheet()
146			if ws:
147			doActionFor(ws, "submit")
148			ws.reindexObject()
149
150			# Promote transition to Analysis Request
151			if IRequestAnalysis.providedBy(analysis):
152			doActionFor(analysis.getRequest(), 'submit')
153			reindex_request(analysis)
154
155
156			def after_retract(analysis):
157			"""Function triggered after a 'retract' transition for the analysis passed
158			in is performed. The analysis transitions to "retracted" state and a new
159			copy of the analysis is created. The copy initial state is "unassigned",
160			unless the the retracted analysis was assigned to a worksheet. In such
161			case, the copy is transitioned to 'assigned' state too
162			"""
163			# Mark this analysis as IRetracted
164			alsoProvides(analysis, IRetracted)
165
166			# Ignore attachments of this analysis in results report
167			for attachment in analysis.getAttachment():
168			attachment.setRenderInReport(False)
169
170			# Retract our dependents (analyses that depend on this analysis)
171			cascade_to_dependents(analysis, "retract")
172
173			# Retract our dependencies (analyses this analysis depends on)
174			promote_to_dependencies(analysis, "retract")
175
176			# Create the retest
177			create_retest(analysis)
178
179			# Try to rollback the Analysis Request
180			if IRequestAnalysis.providedBy(analysis):
181			doActionFor(analysis.getRequest(), "rollback_to_receive")
182			reindex_request(analysis)
183
184
185			def after_reject(analysis):
186			"""Function triggered after the "reject" transition for the analysis passed
187			in is performed."""
188			# Mark this analysis with IRejected
189			alsoProvides(analysis, IRejected)
190
191			# Remove from the worksheet
192			remove_analysis_from_worksheet(analysis)
193
194			# Ignore attachments of this analysis in results report
195			for attachment in analysis.getAttachment():
196			attachment.setRenderInReport(False)
197
198			# Reject our dependents (analyses that depend on this analysis)
199			cascade_to_dependents(analysis, "reject")
200
201			if IRequestAnalysis.providedBy(analysis):
202			# Try verify (for when remaining analyses are in 'verified')
203			doActionFor(analysis.getRequest(), "verify")
204
205			# Try submit (remaining analyses are in 'to_be_verified')
206			doActionFor(analysis.getRequest(), "submit")
207
208			# Try rollback (no remaining analyses or some not submitted)
209			doActionFor(analysis.getRequest(), "rollback_to_receive")
210			reindex_request(analysis)
211
212
213			def after_verify(analysis):
214			"""
215			Method triggered after a 'verify' transition for the analysis passed in
216			is performed. Promotes the transition to the Analysis Request and to
217			Worksheet (if the analysis is assigned to any)
218			This function is called automatically by
219			bika.lims.workfow.AfterTransitionEventHandler
220			"""
221			# Mark this analysis as IVerified
222			alsoProvides(analysis, IVerified)
223
224			# Promote to analyses this analysis depends on
225			promote_to_dependencies(analysis, "verify")
226
227			# Promote transition to worksheet
228			ws = analysis.getWorksheet()
229			if ws:
230			doActionFor(ws, "verify")
231			ws.reindexObject()
232
233			# Promote transition to Analysis Request if Sample auto-verify is enabled
234			if IRequestAnalysis.providedBy(analysis) and check_all_verified(analysis):
235			setup = api.get_setup()
236			if setup.getAutoVerifySamples():
237			doActionFor(analysis.getRequest(), "verify")
238
239			# Reindex the sample (and ancestors) this analysis belongs to
240			reindex_request(analysis)
241
242
243			def check_all_verified(analysis):
244			"""Checks if all analyses are verified
245
246			NOTE: This check is provided solely for performance reasons of the `verify`
247			transition, because it is a less expensive calculation than executing the
248			`doActionFor` method on the sample for each verified analysis.
249
250			The worst case that can happen is that the sample does not get
251			automatically verified and needs to be transitioned manually.
252
253			:param analysis: The current verified analysis
254			:returns: True if all other routine analyses of the sample are verified
255			"""
256
257			parent = api.get_parent(analysis)
258			sample = analysis.getRequest()
259			uid = api.get_uid(analysis)
260
261			def is_valid(an):
262			state = api.get_review_status(an)
263			return state not in [STATE_REJECTED, STATE_RETRACTED]
264
265			# get all valid analyses of the sample
266			analyses = filter(is_valid, sample.getAnalyses())
267			# get all verified analyses of the sample
268			verified = sample.getAnalyses(object_provides=IVerified.__identifier__)
269
270			# NOTE: We remove the current processed routine analysis (if not a WS
271			# duplicate/reference analysis), because it is either not yet
272			# verified or processed already in multi-verify scenarios.
273			if sample == parent:
274			analyses = filter(lambda x: api.get_uid(x) != uid, analyses)
275			verified = filter(lambda x: api.get_uid(x) != uid, verified)
276
277			return len(analyses) == len(verified)
278
279
280			def after_publish(analysis):
281			"""Function triggered after a "publish" transition is performed.
282			"""
283			pass
284
285
286			# TODO Workflow - Analysis - revisit reindexing of ancestors
287			def reindex_request(analysis, idxs=None):
288			"""Reindex the Analysis Request the analysis belongs to, as well as the
289			ancestors recursively
290			"""
291			if not IRequestAnalysis.providedBy(analysis) or \
292			IDuplicateAnalysis.providedBy(analysis):
293			# Analysis not directly bound to an Analysis Request. Do nothing
294			return
295
296			request = analysis.getRequest()
297			ancestors = [request] + request.getAncestors(all_ancestors=True)
298			for ancestor in ancestors:
299			ancestor.reindexObject()
300
301
302			def remove_analysis_from_worksheet(analysis):
303			"""Removes the analysis passed in from the worksheet, if assigned to any
304			"""
305			worksheet = analysis.getWorksheet()
306			if not worksheet:
307			return
308
309			analyses = filter(lambda an: an != analysis, worksheet.getAnalyses())
310			worksheet.setAnalyses(analyses)
311			worksheet.purgeLayout()
312			if analyses:
313			# Maybe this analysis was the only one that was not yet submitted or
314			# verified, so try to submit or verify the Worksheet to be aligned
315			# with the current states of the analyses it contains.
316			doActionFor(worksheet, "submit")
317			doActionFor(worksheet, "verify")
318			else:
319			# We've removed all analyses. Rollback to "open"
320			doActionFor(worksheet, "rollback_to_open")
321
322			# Reindex the Worksheet
323			worksheet.reindexObject()
324
325
326			def cascade_to_dependents(analysis, transition_id):
327			"""Cascades the transition to dependent analyses (those that depend on the
328			analysis passed in), if any
329			"""
330			for dependent in analysis.getDependents():
331			doActionFor(dependent, transition_id)
332
333
334			def promote_to_dependencies(analysis, transition_id):
335			"""Promotes the transition to the analyses this analysis depends on
336			(dependencies), if any
337			"""
338			for dependency in analysis.getDependencies():
339			doActionFor(dependency, transition_id)
340

senaite / senaite.core

Push — 2.x ( 9538a4...3858ed )

check_all_verified() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like