bika.lims.api.analysis.is_retested() - Code Metrics - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

bika.lims.api.analysis.is_retested() A
last analyzed 2025-07-31 09:23 UTC

↳ Parent: bika.lims.api.analysis

Complexity

Conditions

Size

Total Lines	11
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

from collections import Mapping

from bika.lims import api
from bika.lims.config import MAX_OPERATORS
from bika.lims.config import MIN_OPERATORS
from bika.lims.content.analysisspec import ResultsRangeDict
from bika.lims.interfaces import IAnalysis
from bika.lims.interfaces import IDuplicateAnalysis
from bika.lims.interfaces import IReferenceAnalysis
from bika.lims.interfaces import IRejected
from bika.lims.interfaces import IResultOutOfRange
from bika.lims.interfaces import IRetracted
from bika.lims.interfaces.analysis import IRequestAnalysis
from zope.component._api import getAdapters

_marker = object()


def is_out_of_range(brain_or_object, result=_marker):
    """Checks if the result for the analysis passed in is out of range and/or
    out of shoulders range.

            min                                                   max
            warn            min                   max             warn
    ·········|---------------|=====================|---------------|·········
    ----- out-of-range -----><----- in-range ------><----- out-of-range -----
             <-- shoulder --><----- in-range ------><-- shoulder -->

    :param brain_or_object: A single catalog brain or content object
    :param result: Tentative result. If None, use the analysis result
    :type brain_or_object: ATContentType/DexterityContentType/CatalogBrain
    :returns: Tuple of two elements. The first value is `True` if the result is
    out of range and `False` if it is in range. The second value is `True` if
    the result is out of shoulder range and `False` if it is in shoulder range
    :rtype: (bool, bool)
    """
    analysis = api.get_object(brain_or_object)
    if not IAnalysis.providedBy(analysis) and \
            not IReferenceAnalysis.providedBy(analysis):
        api.fail("{} is not supported. Needs to be IAnalysis or "
                 "IReferenceAnalysis".format(repr(analysis)))

    if result is _marker:
        result = api.safe_getattr(analysis, "getResult", None)

    if result in [None, '']:
        # Empty result
        return False, False

    if IDuplicateAnalysis.providedBy(analysis):
        # Result range for duplicate analyses is calculated from the original
        # result, applying a variation % in shoulders. If the analysis has
        # result options enabled or string results enabled, system returns an
        # empty result range for the duplicate: result must match %100 with the
        # original result
        original = analysis.getAnalysis()
        original_result = original.getResult()

        # Does original analysis have a valid result?
        if original_result in [None, '']:
            return False, False

        # Does original result type matches with duplicate result type?
        if api.is_floatable(result) != api.is_floatable(original_result):
            return True, True

        # Does analysis has result options enabled or non-floatable?
        if analysis.getResultOptions() or not api.is_floatable(original_result):
            # Let's always assume the result is 'out from shoulders', cause we
            # consider the shoulders are precisely the duplicate variation %
            out_of_range = original_result != result
            return out_of_range, out_of_range

    elif not api.is_floatable(result):
        results = api.parse_json(result)
        if not results:
            # Single, non-duplicate, non-floatable result. There is no chance
            # to know if the result is out-of-range
            return False, False

        # Multiselect result, remove empty and non-floatable 'sub' results
        results = filter(api.is_floatable, results)
        if not results:
            # No values set yet, we cannot know if out-of-range yet
            return False, False

        # Out of range only when none of the 'sub' results are within range
        for sub_result in results:
            out_range, out_shoulders = is_out_of_range(analysis, sub_result)
            if not out_range:
                # sub result within range
                return False, False

        # None of the 'sub' results are within range
        return True, True

    # Convert result to a float
    result = api.to_float(result)

    # Note that routine analyses, duplicates and reference analyses all them
    # implement the function getResultRange:
    # - For routine analyses, the function returns the valid range based on the
    #   specs assigned during the creation process.
    # - For duplicates, the valid range is the result of the analysis the
    #   the duplicate was generated from +/- the duplicate variation.
    # - For reference analyses, getResultRange returns the valid range as
    #   indicated in the Reference Sample from which the analysis was created.
    result_range = api.safe_getattr(analysis, "getResultsRange", None)
    if not result_range:
        # No result range defined or the passed in object does not suit
        return False, False

    # Maybe there is a custom adapter
    adapters = getAdapters((analysis,), IResultOutOfRange)
    for name, adapter in adapters:
        ret = adapter(result=result, specification=result_range)
        if not ret or not ret.get('out_of_range', False):
            continue
        if not ret.get('acceptable', True):
            # Out of range + out of shoulders
            return True, True
        # Out of range, but in shoulders
        return True, False

    result_range = ResultsRangeDict(result_range)

    # The assignment of result as default fallback for min and max guarantees
    # the result will be in range also if no min/max values are defined
    specs_min = api.to_float(result_range.min, result)
    specs_max = api.to_float(result_range.max, result)

    in_range = False
    min_operator = result_range.min_operator
    if min_operator == "geq":
        in_range = result >= specs_min
    else:
        in_range = result > specs_min

    max_operator = result_range.max_operator
    if in_range:
        if max_operator == "leq":
            in_range = result <= specs_max
        else:
            in_range = result < specs_max

    # If in range, no need to check shoulders
    if in_range:
        return False, False

    # Out of range, check shoulders. If no explicit warn_min or warn_max have
    # been defined, no shoulders must be considered for this analysis. Thus, use
    # specs' min and max as default fallback values
    warn_min = api.to_float(result_range.warn_min, specs_min)
    warn_max = api.to_float(result_range.warn_max, specs_max)
    in_shoulder = warn_min <= result <= warn_max
    return True, not in_shoulder


def get_formatted_interval(results_range, default=_marker):
    """Returns a string representation of the interval defined by the results
    range passed in
    :param results_range: a dict or a ResultsRangeDict
    """
    if not isinstance(results_range, Mapping):
        if default is not _marker:
            return default
        api.fail("Type not supported")
    results_range = ResultsRangeDict(results_range)
    min_str = results_range.min if api.is_floatable(results_range.min) else None
    max_str = results_range.max if api.is_floatable(results_range.max) else None
    if min_str is None and max_str is None:
        if default is not _marker:
            return default
        api.fail("Min and max values are not floatable or not defined")

    min_operator = results_range.min_operator
    max_operator = results_range.max_operator
    if max_str is None:
        return "{}{}".format(MIN_OPERATORS.getValue(min_operator), min_str)
    if min_str is None:
        return "{}{}".format(MAX_OPERATORS.getValue(max_operator), max_str)

    # Both values set. Return an interval
    min_bracket = min_operator == 'geq' and '[' or '('
    max_bracket = max_operator == 'leq' and ']' or ')'

    return "{}{};{}{}".format(min_bracket, min_str, max_str, max_bracket)


def is_result_range_compliant(analysis):
    """Returns whether the result range from the analysis matches with the
    result range for the service counterpart defined in the Sample
    """
    if not IRequestAnalysis.providedBy(analysis):
        return True

    if IDuplicateAnalysis.providedBy(analysis):
        # Does not make sense to apply compliance to a duplicate, cause its
        # valid range depends on the result of the original analysis
        return True

    rr = analysis.getResultsRange()
    service_uid = rr.get("uid", None)
    if not api.is_uid(service_uid):
        return True

    # Compare with Sample
    sample = analysis.getRequest()

    # If no Specification is set, assume is compliant
    specification = sample.getRawSpecification()
    if not specification:
        return True

    # Compare with the Specification that was initially set to the Sample
    sample_rr = sample.getResultsRange(search_by=service_uid)
    if not sample_rr:
        # This service is not defined in Sample's ResultsRange, we
        # assume this *does not* break the compliance
        return True

    return rr == sample_rr


def is_analysis(brain_or_object):
    """Checks if the object is an analysis

    :param brain_or_object: A single catalog brain or content object
    :returns: True if the object is an analysis, False otherwise
    """
    analysis = api.get_object(brain_or_object)
    return IAnalysis.providedBy(analysis)


def is_reference_analysis(brain_or_object):
    """Checks if the object is a reference analysis

    :param brain_or_object: A single catalog brain or content object
    :returns: True if the object is a reference analysis, False otherwise
    """
    analysis = api.get_object(brain_or_object)
    return IReferenceAnalysis.providedBy(analysis)


def is_retracted(brain_or_object):
    """Checks if an analysis is retracted

    :param brain_or_object: A single catalog brain or content object
    :returns: True if the analysis is retracted, False otherwise
    """
    analysis = api.get_object(brain_or_object)
    if not is_analysis(analysis) and not is_reference_analysis(analysis):
        api.fail("{} is not supported. Needs to be IAnalysis or "
                 "IReferenceAnalysis".format(repr(analysis)))
    return IRetracted.providedBy(analysis)


def is_rejected(brain_or_object):
    """Checks if the analysis is rejected

    :param brain_or_object: A single catalog brain or content object
    :returns: True if the analysis is rejected, False otherwise
    """
    analysis = api.get_object(brain_or_object)
    if not is_analysis(analysis) and not is_reference_analysis(analysis):
        api.fail("{} is not supported. Needs to be IAnalysis or "
                 "IReferenceAnalysis".format(repr(analysis)))
    return IRejected.providedBy(analysis)


def is_retested(brain_or_object):
    """Checks if the analysis is retested

    :param brain_or_object: A single catalog brain or content object
    :returns: True if the analysis is retested, False otherwise
    """
    analysis = api.get_object(brain_or_object)
    if not is_analysis(analysis) and not is_reference_analysis(analysis):
        api.fail("{} is not supported. Needs to be IAnalysis or "
                 "IReferenceAnalysis".format(repr(analysis)))
    return analysis.isRetested()


def get_dependencies(brain_or_object, with_retests=False, recursive=False):
    """Returns the list of dependent analysis UIDs for the analysis passed in

    :param brain_or_object: A single catalog brain or content object
    :returns: List analysis objects that this analysis depends on
    """
    if not is_analysis(brain_or_object):
        return []

    dependencies = set()
    analysis = api.get_object(brain_or_object)

    # no calculation, no dependencies
    calc = analysis.getCalculation()
    if not calc:
        return []

    # get the sample (might be a partition)
    sample = analysis.getRequest()
    # get calculation dependencies
    service_deps = calc.getDependentServices()
    # get the keywords of the dependent services
    keywords = [s.getKeyword() for s in service_deps]
    # no dependencies to other services, nothing to do
    if not keywords:
        return []
    # collect the analyses
    dependencies.update(sample.getAnalyses(getKeyword=keywords))

    # calculate all dependencies for our dependencies
    if recursive:
        # iterate over all dependencies and get their dependencies
        for dep in list(dependencies):
            dependencies.update(get_dependencies(
                dep, with_retests=with_retests, recursive=recursive))

    if not with_retests:
        # filter out retracted, rejected and retested analyses
        def is_retest(analysis):
            return is_retracted(analysis) or is_rejected(analysis) \
                or is_retested(analysis)
        dependencies = filter(lambda d: not is_retest(d), dependencies)


    return map(api.get_object, dependencies)


def get_dependents(brain_or_object, with_retests=False, recursive=False):
    """Returns the list of analysis UIDs that depend on the current

    :param brain_or_object: A single catalog brain or content object
    :returns: List of analysis object that depend on the current analysis
    """
    if not is_analysis(brain_or_object):
        return []

    dependents = set()
    analysis = api.get_object(brain_or_object)

    # get the service of the current analysis
    service = analysis.getAnalysisService()

    # get the sample (might be a partition)
    sample = analysis.getRequest()

    # get all analyses with calculations
    analyses_with_calcs = sample.getAnalyses(
        has_calculation=True, full_objects=True)

    # Now we check if we are part of any calculation
    for analysis in analyses_with_calcs:
        calc = analysis.getCalculation()
        if not calc:
            # in case the `has_calculation` index is not there yet
            continue
        dependencies = calc.getDependentServices()
        # check if our service is a dependency
        if service in dependencies:
            # remember the analysis that depends on us
            dependents.add(analysis)

    if recursive:
        for dep in list(dependents):
            dependents.update(get_dependents(
                dep, with_retests=with_retests, recursive=recursive))

    if not with_retests:
        # filter out retracted, rejected and retested analyses
        def is_retest(analysis):
            return is_retracted(analysis) or is_rejected(analysis) \
                or is_retested(analysis)
        dependents = filter(lambda d: not is_retest(d), dependents)


    return map(api.get_object, dependents)


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-2025 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			from collections import Mapping
22
23			from bika.lims import api
24			from bika.lims.config import MAX_OPERATORS
25			from bika.lims.config import MIN_OPERATORS
26			from bika.lims.content.analysisspec import ResultsRangeDict
27			from bika.lims.interfaces import IAnalysis
28			from bika.lims.interfaces import IDuplicateAnalysis
29			from bika.lims.interfaces import IReferenceAnalysis
30			from bika.lims.interfaces import IRejected
31			from bika.lims.interfaces import IResultOutOfRange
32			from bika.lims.interfaces import IRetracted
33			from bika.lims.interfaces.analysis import IRequestAnalysis
34			from zope.component._api import getAdapters
35
36			_marker = object()
37
38
39			def is_out_of_range(brain_or_object, result=_marker):
40			"""Checks if the result for the analysis passed in is out of range and/or
41			out of shoulders range.
42
43			min max
44			warn min max warn
45			·········\|---------------\|=====================\|---------------\|·········
46			----- out-of-range -----><----- in-range ------><----- out-of-range -----
47			<-- shoulder --><----- in-range ------><-- shoulder -->
48
49			:param brain_or_object: A single catalog brain or content object
50			:param result: Tentative result. If None, use the analysis result
51			:type brain_or_object: ATContentType/DexterityContentType/CatalogBrain
52			:returns: Tuple of two elements. The first value is `True` if the result is
53			out of range and `False` if it is in range. The second value is `True` if
54			the result is out of shoulder range and `False` if it is in shoulder range
55			:rtype: (bool, bool)
56			"""
57			analysis = api.get_object(brain_or_object)
58			if not IAnalysis.providedBy(analysis) and \
59			not IReferenceAnalysis.providedBy(analysis):
60			api.fail("{} is not supported. Needs to be IAnalysis or "
61			"IReferenceAnalysis".format(repr(analysis)))
62
63			if result is _marker:
64			result = api.safe_getattr(analysis, "getResult", None)
65
66			if result in [None, '']:
67			# Empty result
68			return False, False
69
70			if IDuplicateAnalysis.providedBy(analysis):
71			# Result range for duplicate analyses is calculated from the original
72			# result, applying a variation % in shoulders. If the analysis has
73			# result options enabled or string results enabled, system returns an
74			# empty result range for the duplicate: result must match %100 with the
75			# original result
76			original = analysis.getAnalysis()
77			original_result = original.getResult()
78
79			# Does original analysis have a valid result?
80			if original_result in [None, '']:
81			return False, False
82
83			# Does original result type matches with duplicate result type?
84			if api.is_floatable(result) != api.is_floatable(original_result):
85			return True, True
86
87			# Does analysis has result options enabled or non-floatable?
88			if analysis.getResultOptions() or not api.is_floatable(original_result):
89			# Let's always assume the result is 'out from shoulders', cause we
90			# consider the shoulders are precisely the duplicate variation %
91			out_of_range = original_result != result
92			return out_of_range, out_of_range
93
94			elif not api.is_floatable(result):
95			results = api.parse_json(result)
96			if not results:
97			# Single, non-duplicate, non-floatable result. There is no chance
98			# to know if the result is out-of-range
99			return False, False
100
101			# Multiselect result, remove empty and non-floatable 'sub' results
102			results = filter(api.is_floatable, results)
103			if not results:
104			# No values set yet, we cannot know if out-of-range yet
105			return False, False
106
107			# Out of range only when none of the 'sub' results are within range
108			for sub_result in results:
109			out_range, out_shoulders = is_out_of_range(analysis, sub_result)
110			if not out_range:
111			# sub result within range
112			return False, False
113
114			# None of the 'sub' results are within range
115			return True, True
116
117			# Convert result to a float
118			result = api.to_float(result)
119
120			# Note that routine analyses, duplicates and reference analyses all them
121			# implement the function getResultRange:
122			# - For routine analyses, the function returns the valid range based on the
123			# specs assigned during the creation process.
124			# - For duplicates, the valid range is the result of the analysis the
125			# the duplicate was generated from +/- the duplicate variation.
126			# - For reference analyses, getResultRange returns the valid range as
127			# indicated in the Reference Sample from which the analysis was created.
128			result_range = api.safe_getattr(analysis, "getResultsRange", None)
129			if not result_range:
130			# No result range defined or the passed in object does not suit
131			return False, False
132
133			# Maybe there is a custom adapter
134			adapters = getAdapters((analysis,), IResultOutOfRange)
135			for name, adapter in adapters:
136			ret = adapter(result=result, specification=result_range)
137			if not ret or not ret.get('out_of_range', False):
138			continue
139			if not ret.get('acceptable', True):
140			# Out of range + out of shoulders
141			return True, True
142			# Out of range, but in shoulders
143			return True, False
144
145			result_range = ResultsRangeDict(result_range)
146
147			# The assignment of result as default fallback for min and max guarantees
148			# the result will be in range also if no min/max values are defined
149			specs_min = api.to_float(result_range.min, result)
150			specs_max = api.to_float(result_range.max, result)
151
152			in_range = False
153			min_operator = result_range.min_operator
154			if min_operator == "geq":
155			in_range = result >= specs_min
156			else:
157			in_range = result > specs_min
158
159			max_operator = result_range.max_operator
160			if in_range:
161			if max_operator == "leq":
162			in_range = result <= specs_max
163			else:
164			in_range = result < specs_max
165
166			# If in range, no need to check shoulders
167			if in_range:
168			return False, False
169
170			# Out of range, check shoulders. If no explicit warn_min or warn_max have
171			# been defined, no shoulders must be considered for this analysis. Thus, use
172			# specs' min and max as default fallback values
173			warn_min = api.to_float(result_range.warn_min, specs_min)
174			warn_max = api.to_float(result_range.warn_max, specs_max)
175			in_shoulder = warn_min <= result <= warn_max
176			return True, not in_shoulder
177
178
179			def get_formatted_interval(results_range, default=_marker):
180			"""Returns a string representation of the interval defined by the results
181			range passed in
182			:param results_range: a dict or a ResultsRangeDict
183			"""
184			if not isinstance(results_range, Mapping):
185			if default is not _marker:
186			return default
187			api.fail("Type not supported")
188			results_range = ResultsRangeDict(results_range)
189			min_str = results_range.min if api.is_floatable(results_range.min) else None
190			max_str = results_range.max if api.is_floatable(results_range.max) else None
191			if min_str is None and max_str is None:
192			if default is not _marker:
193			return default
194			api.fail("Min and max values are not floatable or not defined")
195
196			min_operator = results_range.min_operator
197			max_operator = results_range.max_operator
198			if max_str is None:
199			return "{}{}".format(MIN_OPERATORS.getValue(min_operator), min_str)
200			if min_str is None:
201			return "{}{}".format(MAX_OPERATORS.getValue(max_operator), max_str)
202
203			# Both values set. Return an interval
204			min_bracket = min_operator == 'geq' and '[' or '('
205			max_bracket = max_operator == 'leq' and ']' or ')'
206
207			return "{}{};{}{}".format(min_bracket, min_str, max_str, max_bracket)
208
209
210			def is_result_range_compliant(analysis):
211			"""Returns whether the result range from the analysis matches with the
212			result range for the service counterpart defined in the Sample
213			"""
214			if not IRequestAnalysis.providedBy(analysis):
215			return True
216
217			if IDuplicateAnalysis.providedBy(analysis):
218			# Does not make sense to apply compliance to a duplicate, cause its
219			# valid range depends on the result of the original analysis
220			return True
221
222			rr = analysis.getResultsRange()
223			service_uid = rr.get("uid", None)
224			if not api.is_uid(service_uid):
225			return True
226
227			# Compare with Sample
228			sample = analysis.getRequest()
229
230			# If no Specification is set, assume is compliant
231			specification = sample.getRawSpecification()
232			if not specification:
233			return True
234
235			# Compare with the Specification that was initially set to the Sample
236			sample_rr = sample.getResultsRange(search_by=service_uid)
237			if not sample_rr:
238			# This service is not defined in Sample's ResultsRange, we
239			# assume this does not break the compliance
240			return True
241
242			return rr == sample_rr
243
244
245			def is_analysis(brain_or_object):
246			"""Checks if the object is an analysis
247
248			:param brain_or_object: A single catalog brain or content object
249			:returns: True if the object is an analysis, False otherwise
250			"""
251			analysis = api.get_object(brain_or_object)
252			return IAnalysis.providedBy(analysis)
253
254
255			def is_reference_analysis(brain_or_object):
256			"""Checks if the object is a reference analysis
257
258			:param brain_or_object: A single catalog brain or content object
259			:returns: True if the object is a reference analysis, False otherwise
260			"""
261			analysis = api.get_object(brain_or_object)
262			return IReferenceAnalysis.providedBy(analysis)
263
264
265			def is_retracted(brain_or_object):
266			"""Checks if an analysis is retracted
267
268			:param brain_or_object: A single catalog brain or content object
269			:returns: True if the analysis is retracted, False otherwise
270			"""
271			analysis = api.get_object(brain_or_object)
272			if not is_analysis(analysis) and not is_reference_analysis(analysis):
273			api.fail("{} is not supported. Needs to be IAnalysis or "
274			"IReferenceAnalysis".format(repr(analysis)))
275			return IRetracted.providedBy(analysis)
276
277
278			def is_rejected(brain_or_object):
279			"""Checks if the analysis is rejected
280
281			:param brain_or_object: A single catalog brain or content object
282			:returns: True if the analysis is rejected, False otherwise
283			"""
284			analysis = api.get_object(brain_or_object)
285			if not is_analysis(analysis) and not is_reference_analysis(analysis):
286			api.fail("{} is not supported. Needs to be IAnalysis or "
287			"IReferenceAnalysis".format(repr(analysis)))
288			return IRejected.providedBy(analysis)
289
290
291			def is_retested(brain_or_object):
292			"""Checks if the analysis is retested
293
294			:param brain_or_object: A single catalog brain or content object
295			:returns: True if the analysis is retested, False otherwise
296			"""
297			analysis = api.get_object(brain_or_object)
298			if not is_analysis(analysis) and not is_reference_analysis(analysis):
299			api.fail("{} is not supported. Needs to be IAnalysis or "
300			"IReferenceAnalysis".format(repr(analysis)))
301			return analysis.isRetested()
302
303
304			def get_dependencies(brain_or_object, with_retests=False, recursive=False):
305			"""Returns the list of dependent analysis UIDs for the analysis passed in
306
307			:param brain_or_object: A single catalog brain or content object
308			:returns: List analysis objects that this analysis depends on
309			"""
310			if not is_analysis(brain_or_object):
311			return []
312
313			dependencies = set()
314			analysis = api.get_object(brain_or_object)
315
316			# no calculation, no dependencies
317			calc = analysis.getCalculation()
318			if not calc:
319			return []
320
321			# get the sample (might be a partition)
322			sample = analysis.getRequest()
323			# get calculation dependencies
324			service_deps = calc.getDependentServices()
325			# get the keywords of the dependent services
326			keywords = [s.getKeyword() for s in service_deps]
327			# no dependencies to other services, nothing to do
328			if not keywords:
329			return []
330			# collect the analyses
331			dependencies.update(sample.getAnalyses(getKeyword=keywords))
332
333			# calculate all dependencies for our dependencies
334			if recursive:
335			# iterate over all dependencies and get their dependencies
336			for dep in list(dependencies):
337			dependencies.update(get_dependencies(
338			dep, with_retests=with_retests, recursive=recursive))
339
340			if not with_retests:
341			# filter out retracted, rejected and retested analyses
342			def is_retest(analysis):
343			return is_retracted(analysis) or is_rejected(analysis) \
344			or is_retested(analysis)
345			dependencies = filter(lambda d: not is_retest(d), dependencies)
			0 ignored issues – show introduced 2025-07-22 14:31 UTC by Report Bug Copy Issue Report The variable `is_retest` does not seem to be defined in case `BooleanNotNode` on line `340` is `False`. Are you sure this can never be the case? Loading history...
346
347			return map(api.get_object, dependencies)
348
349
350			def get_dependents(brain_or_object, with_retests=False, recursive=False):
351			"""Returns the list of analysis UIDs that depend on the current
352
353			:param brain_or_object: A single catalog brain or content object
354			:returns: List of analysis object that depend on the current analysis
355			"""
356			if not is_analysis(brain_or_object):
357			return []
358
359			dependents = set()
360			analysis = api.get_object(brain_or_object)
361
362			# get the service of the current analysis
363			service = analysis.getAnalysisService()
364
365			# get the sample (might be a partition)
366			sample = analysis.getRequest()
367
368			# get all analyses with calculations
369			analyses_with_calcs = sample.getAnalyses(
370			has_calculation=True, full_objects=True)
371
372			# Now we check if we are part of any calculation
373			for analysis in analyses_with_calcs:
374			calc = analysis.getCalculation()
375			if not calc:
376			# in case the `has_calculation` index is not there yet
377			continue
378			dependencies = calc.getDependentServices()
379			# check if our service is a dependency
380			if service in dependencies:
381			# remember the analysis that depends on us
382			dependents.add(analysis)
383
384			if recursive:
385			for dep in list(dependents):
386			dependents.update(get_dependents(
387			dep, with_retests=with_retests, recursive=recursive))
388
389			if not with_retests:
390			# filter out retracted, rejected and retested analyses
391			def is_retest(analysis):
392			return is_retracted(analysis) or is_rejected(analysis) \
393			or is_retested(analysis)
394			dependents = filter(lambda d: not is_retest(d), dependents)
			0 ignored issues – show introduced 2025-07-22 14:31 UTC by Report Bug Copy Issue Report The variable `is_retest` does not seem to be defined for all execution paths. Loading history...
395
396			return map(api.get_object, dependents)
397

senaite / senaite.core

bika.lims.api.analysis.is_retested() A last analyzed 2025-07-31 09:23 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

bika.lims.api.analysis.is_retested() A
last analyzed 2025-07-31 09:23 UTC