bika.lims.content.analysisservice.AnalysisService.getServiceDependents() - Code Metrics - Inspection of "Fix performance of analysis dependency calculation..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — 2.x ( 3a5aed...187ea7 )

by Jordi

created 2025-07-22 14:24 UTC

AnalysisService.getServiceDependents() A

↳ Parent: bika.lims.content.analysisservice

Complexity

Conditions

Size

Total Lines	16
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	14
dl	0
loc	16
rs	9.7
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-2025 by it's authors.
# Some rights reserved, see README and LICENSE.

import itertools

from AccessControl import ClassSecurityInfo
from bika.lims import api
from bika.lims import bikaMessageFactory as _
from bika.lims.browser.fields import InterimFieldsField
from bika.lims.browser.fields import PartitionSetupField
from bika.lims.browser.fields import UIDReferenceField
from bika.lims.browser.fields.partitionsetupfield import getContainers
from bika.lims.browser.widgets.partitionsetupwidget import PartitionSetupWidget
from bika.lims.browser.widgets.recordswidget import RecordsWidget
from senaite.core.browser.widgets.referencewidget import ReferenceWidget
from bika.lims.catalog import SETUP_CATALOG
from bika.lims.config import PROJECTNAME
from bika.lims.content.abstractbaseanalysis import AbstractBaseAnalysis
from bika.lims.content.abstractbaseanalysis import schema
from bika.lims.interfaces import IAnalysisService
from bika.lims.interfaces import IDeactivable
from Products.Archetypes.atapi import PicklistWidget
from Products.Archetypes.Field import StringField
from Products.Archetypes.public import BooleanField
from Products.Archetypes.public import BooleanWidget
from Products.Archetypes.public import DisplayList
from Products.Archetypes.public import Schema
from Products.Archetypes.public import SelectionWidget
from Products.Archetypes.public import registerType
from Products.Archetypes.Widget import StringWidget
from senaite.core.browser.fields.records import RecordsField
from zope.interface import implements

Methods = UIDReferenceField(
    "Methods",
    schemata="Method",
    required=0,
    multiValued=1,
    vocabulary="_methods_vocabulary",
    allowed_types=("Method", ),
    accessor="getRawMethods",
    widget=PicklistWidget(
        label=_("Methods"),
        description=_(
            "Available methods to perform the test"),
    )
)

Instruments = UIDReferenceField(
    "Instruments",
    schemata="Method",
    required=0,
    multiValued=1,
    vocabulary="_instruments_vocabulary",
    allowed_types=("Instrument", ),
    accessor="getRawInstruments",
    widget=PicklistWidget(
        label=_("Instruments"),
        description=_("Available instruments based on the selected methods."),
    )
)

# XXX: HIDDEN -> TO BE REMOVED
UseDefaultCalculation = BooleanField(
    "UseDefaultCalculation",
    schemata="Method",
    default=True,
    widget=BooleanWidget(
        visible=False,
        label=_("Use the Default Calculation of Method"),
        description=_(
            "Select if the calculation to be used is the calculation set by "
            "default in the default method. If unselected, the calculation "
            "can be selected manually"),
    )
)

Calculation = UIDReferenceField(
    "Calculation",
    schemata="Method",
    required=0,
    relationship="AnalysisServiceCalculation",
    vocabulary="_default_calculation_vocabulary",
    allowed_types=("Calculation", ),
    accessor="getRawCalculation",
    widget=SelectionWidget(
        format="select",
        label=_("Calculation"),
        description=_("Calculation to be assigned to this content."),
        catalog_name=SETUP_CATALOG,
        base_query={"is_active": True},
    )
)

InterimFields = InterimFieldsField(
    "InterimFields",
    schemata="Result Options",
    widget=RecordsWidget(
        label=_("Result variables"),
        description=_("Additional result values"),
    )
)

# XXX: HIDDEN -> TO BE REMOVED
Separate = BooleanField(
    "Separate",
    schemata="Container and Preservation",
    default=False,
    required=0,
    widget=BooleanWidget(
        visible=False,
        label=_("Separate Container"),
        description=_("Check this box to ensure a separate sample container "
                      "is used for this analysis service"),
    )
)

# XXX: HIDDEN -> TO BE REMOVED
Preservation = UIDReferenceField(
    "Preservation",
    schemata="Container and Preservation",
    allowed_types=("SamplePreservation",),
    vocabulary="getPreservations",
    required=0,
    multiValued=0,
    widget=ReferenceWidget(
        visible=False,
        label=_("Default Preservation"),
        description=_(
            "Select a default preservation for this analysis service. If the "
            "preservation depends on the sample type combination, specify a "
            "preservation per sample type in the table below"),
        catalog_name=SETUP_CATALOG,
        base_query={"is_active": True},
    )
)

# XXX: HIDDEN -> TO BE REMOVED
Container = UIDReferenceField(
    "Container",
    schemata="Container and Preservation",
    allowed_types=("Container", "ContainerType"),
    vocabulary="getContainers",
    required=0,
    multiValued=0,
    widget=ReferenceWidget(
        visible=False,
        label=_("Default Container"),
        description=_(
            "Select the default container to be used for this analysis "
            "service. If the container to be used depends on the sample type "
            "and preservation combination, specify the container in the "
            "sample type table below"),
        catalog_name=SETUP_CATALOG,
        base_query={"is_active": True},
    )
)

# XXX: HIDDEN -> TO BE REMOVED
PartitionSetup = PartitionSetupField(
    "PartitionSetup",
    schemata="Container and Preservation",
    widget=PartitionSetupWidget(
        visible=False,
        label=_("Preservation per sample type"),
        description=_(
            "Please specify preservations that differ from the analysis "
            "service's default preservation per sample type here."),
    )
)

# Allow/disallow the capture of text as the result of the analysis
DefaultResult = StringField(
    "DefaultResult",
    schemata="Result Options",
    validators=('service_defaultresult_validator',),
    widget=StringWidget(
        label=_("Default result"),
        description=_(
            "Default result to display on result entry"
        )
    )
)

Conditions = RecordsField(
    "Conditions",
    schemata="Advanced",
    type="conditions",
    subfields=(
        "title",
        "description",
        "type",
        "choices",
        "default",
        "required",
        "report",
    ),
    required_subfields=(
        "title",
        "type",
    ),
    subfield_labels={
        "title": _("Title"),
        "description": _("Description"),
        "type": _("Control type"),
        "choices": _("Choices"),
        "default": _("Default value"),
        "required": _("Required"),
        "report": _("Report"),
    },
    subfield_descriptions={
        "choices": _("Please use the following format for select options: "
                     "key1:value1|key2:value2|...|keyN:valueN"),
    },
    subfield_types={
        "title": "string",
        "description": "string",
        "type": "string",
        "default": "string",
        "choices": "string",
        "required": "boolean",
        "report": "boolean",
    },
    subfield_sizes={
        "title": 20,
        "description": 50,
        "type": 1,
        "choices": 30,
        "default": 20,
    },
    subfield_validators={
        "title": "service_conditions_validator",
    },
    subfield_maxlength={
        "title": 50,
        "description": 200,
    },
    subfield_vocabularies={
        "type": DisplayList((
            ('', ''),
            ('text', _('Text')),
            ('number', _('Number')),
            ('checkbox', _('Checkbox')),
            ('select', _('Select')),
            ('file', _('File upload')),
        )),
    },
    widget=RecordsWidget(
        label=_("Analysis conditions"),
        description=_(
            "Conditions to ask for this analysis on sample registration. For "
            "instance, laboratory may want the user to input the temperature, "
            "the ramp and flow when a thermogravimetric (TGA) analysis is "
            "selected on sample registration. The information provided will be "
            "later considered by the laboratory personnel when performing the "
            "test."
        ),
    )
)


schema = schema.copy() + Schema((
    Methods,
    Instruments,
    UseDefaultCalculation,
    Calculation,
    InterimFields,
    Separate,
    Preservation,
    Container,
    PartitionSetup,
    DefaultResult,
    Conditions,
))

# Move default method field after available methods field
schema.moveField("Method", after="Methods")
# Move default instrument field after available instruments field
schema.moveField("Instrument", after="Instruments")
# Move default result field after Result Options
schema.moveField("DefaultResult", after="ResultOptions")


class AnalysisService(AbstractBaseAnalysis):
    """Analysis Service Content Holder
    """
    implements(IAnalysisService, IDeactivable)
    security = ClassSecurityInfo()
    schema = schema

    _at_rename_after_creation = True

    def _renameAfterCreation(self, check_auto_id=False):
        from senaite.core.idserver import renameAfterCreation

        return renameAfterCreation(self)

    def getMethods(self):
        """Returns the assigned methods

        :returns: List of method objects
        """
        return self.getField("Methods").get(self)

    def getRawMethods(self):
        """Returns the assigned method UIDs

        :returns: List of method UIDs
        """
        return self.getField("Methods").getRaw(self)

    def getMethod(self):
        """Get the default method
        """
        field = self.getField("Method")
        method = field.get(self)
        if not method:
            return None
        # check if the method is in the selected methods
        methods = self.getMethods()
        if method not in methods:
            return None
        return method

    def getRawMethod(self):
        """Returns the UID of the default method

        :returns: method UID
        """
        field = self.getField("Method")
        method = field.getRaw(self)
        if not method:
            return None
        # check if the method is in the selected methods
        methods = self.getRawMethods()
        if method not in methods:
            return None
        return method

    def getInstruments(self):
        """Returns the assigned instruments

        :returns: List of instrument objects
        """
        return self.getField("Instruments").get(self)

    def getRawInstruments(self):
        """List of assigned Instrument UIDs
        """
        return self.getField("Instruments").getRaw(self)

    def getInstrument(self):
        """Return the default instrument
        """
        field = self.getField("Instrument")
        instrument = field.get(self)
        if not instrument:
            return None
        # check if the instrument is in the selected instruments
        instruments = self.getInstruments()
        if instrument not in instruments:
            return None
        return instrument

    def getRawInstrument(self):
        """Return the UID of the default instrument
        """
        field = self.getField("Instrument")
        instrument = field.getRaw(self)
        if not instrument:
            return None
        # check if the instrument is in the selected instruments
        instruments = self.getRawInstruments()
        if instrument not in instruments:
            return None
        return instrument

    def getCalculation(self):
        """Get the default calculation
        """
        field = self.getField("Calculation")
        calculation = field.get(self)
        if not calculation:
            return None
        return calculation

    def getRawCalculation(self):
        """Returns the UID of the assigned calculation

        :returns: Calculation UID
        """
        field = self.getField("Calculation")
        calculation = field.getRaw(self)
        if not calculation:
            return None
        return calculation

    def getServiceDependencies(self):
        """Return calculation dependencies of the service

        :return: a list of analysis services objects.
        """
        calc = self.getCalculation()
        if calc:
            return calc.getCalculationDependencies(flat=True)
        return []

    def getServiceDependenciesUIDs(self):
        """Return calculation dependency UIDs of the service

        :return: a list of uids
        """
        return map(api.get_uid, self.getServiceDependencies())

    def getServiceDependents(self):
        """Return services depending on us
        """
        catalog = api.get_tool(SETUP_CATALOG)
        active_calcs = catalog(portal_type="Calculation", is_active=True)
        calculations = map(api.get_object, active_calcs)
        dependents = []
        for calc in calculations:
            calc_dependents = calc.getDependentServices()
            if self in calc_dependents:
                calc_dependencies = calc.getCalculationDependents()
                dependents = dependents + calc_dependencies
        dependents = list(set(dependents))
        if self in dependents:
            dependents.remove(self)
        return dependents

    def getServiceDependentsUIDs(self):
        """Return service UIDs depending on us
        """
        return map(api.get_uid, self.getServiceDependents())

    def query_available_methods(self):
        """Return all available methods
        """
        catalog = api.get_tool(SETUP_CATALOG)
        query = {
            "portal_type": "Method",
            "is_active": True,
            "sort_on": "sortable_title",
            "sort_order": "ascending",
        }
        return catalog(query)

    def _methods_vocabulary(self):
        """Vocabulary used for methods field
        """
        methods = self.query_available_methods()
        items = [(api.get_uid(m), api.get_title(m)) for m in methods]
        dlist = DisplayList(items)
        return dlist

    def _default_method_vocabulary(self):
        """Vocabulary used for default method field
        """
        # check if we selected methods
        methods = self.getMethods()
        if not methods:
            # query all available methods
            methods = self.query_available_methods()
        items = [(api.get_uid(m), api.get_title(m)) for m in methods]
        dlist = DisplayList(items).sortedByValue()
        # allow to leave this field empty
        dlist.add("", _("None"))
        return dlist

    def query_available_instruments(self):
        """Return all available Instruments
        """
        catalog = api.get_tool(SETUP_CATALOG)
        query = {
            "portal_type": "Instrument",
            "is_active": True,
            "sort_on": "sortable_title",
            "sort_order": "ascending",
        }
        return catalog(query)

    def _instruments_vocabulary(self):
        """Vocabulary used for instruments field
        """
        instruments = []
        # When methods are selected, display only instruments from the methods
        methods = self.getMethods()
        for method in methods:
            for instrument in method.getInstruments():
                if instrument in instruments:
                    continue
                instruments.append(instrument)

        if not methods:
            # query all available instruments when no methods are selected
            instruments = self.query_available_instruments()

        items = [(api.get_uid(i), api.get_title(i)) for i in instruments]
        dlist = DisplayList(items)
        return dlist

    def _default_instrument_vocabulary(self):
        """Vocabulary used for default instrument field
        """
        # check if we selected instruments
        instruments = self.getInstruments()
        items = [(api.get_uid(i), api.get_title(i)) for i in instruments]
        dlist = DisplayList(items).sortedByValue()
        # allow to leave this field empty
        dlist.add("", _("None"))
        return dlist

    def query_available_calculations(self):
        """Return all available calculations
        """
        catalog = api.get_tool(SETUP_CATALOG)
        query = {
            "portal_type": "Calculation",
            "is_active": True,
            "sort_on": "sortable_title",
            "sort_order": "ascending",
        }
        return catalog(query)

    def get_methods_calculations(self):
        """Return calculations assigned to the selected methods
        """
        methods = self.getMethods()
        if not methods:
            return None
        methods_calcs = map(lambda m: m.getCalculations(), methods)
        return list(itertools.chain(*methods_calcs))

    def _default_calculation_vocabulary(self):
        """Vocabulary used for the default calculation field
        """
        calculations = self.get_methods_calculations()
        if calculations is None:
            # query all available calculations
            calculations = self.query_available_calculations()
        items = [(api.get_uid(c), api.get_title(c)) for c in calculations]
        # allow to leave this field empty
        dlist = DisplayList(items).sortedByValue()
        dlist.add("", _("None"))
        return dlist

    # XXX DECIDE IF NEEDED
    # --------------------

    def getContainers(self, instance=None):
        # On first render, the containers must be filtered
        instance = instance or self
        separate = self.getSeparate()
        containers = getContainers(instance,
                                   allow_blank=True,
                                   show_container_types=not separate,
                                   show_containers=separate)
        return DisplayList(containers)

    def getPreservations(self):
        query = {
            "portal_type": "SamplePreservation",
            "is_active": True,
            "sort_on": "sortable_title",
            "sort_order": "ascending",
        }
        brains = api.search(query, SETUP_CATALOG)
        items = [(brain.UID, brain.Title) for brain in brains]
        return DisplayList(items)

    def getAvailableMethods(self):
        """Returns the methods available for this analysis.
        """
        return self.getMethods()

    def getAvailableMethodUIDs(self):
        """Returns the UIDs of the available methods

        Used here:
        bika.lims.catalog.indexers.bikasetup.method_available_uid
        """
        # N.B. we return a copy of the list to avoid accidental writes
        return self.getRawMethods()[:]


registerType(AnalysisService, PROJECTNAME)


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			import itertools
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 InterimFieldsField
27			from bika.lims.browser.fields import PartitionSetupField
28			from bika.lims.browser.fields import UIDReferenceField
29			from bika.lims.browser.fields.partitionsetupfield import getContainers
30			from bika.lims.browser.widgets.partitionsetupwidget import PartitionSetupWidget
31			from bika.lims.browser.widgets.recordswidget import RecordsWidget
32			from senaite.core.browser.widgets.referencewidget import ReferenceWidget
33			from bika.lims.catalog import SETUP_CATALOG
34			from bika.lims.config import PROJECTNAME
35			from bika.lims.content.abstractbaseanalysis import AbstractBaseAnalysis
36			from bika.lims.content.abstractbaseanalysis import schema
37			from bika.lims.interfaces import IAnalysisService
38			from bika.lims.interfaces import IDeactivable
39			from Products.Archetypes.atapi import PicklistWidget
40			from Products.Archetypes.Field import StringField
41			from Products.Archetypes.public import BooleanField
42			from Products.Archetypes.public import BooleanWidget
43			from Products.Archetypes.public import DisplayList
44			from Products.Archetypes.public import Schema
45			from Products.Archetypes.public import SelectionWidget
46			from Products.Archetypes.public import registerType
47			from Products.Archetypes.Widget import StringWidget
48			from senaite.core.browser.fields.records import RecordsField
49			from zope.interface import implements
50
51			Methods = UIDReferenceField(
52			"Methods",
53			schemata="Method",
54			required=0,
55			multiValued=1,
56			vocabulary="_methods_vocabulary",
57			allowed_types=("Method", ),
58			accessor="getRawMethods",
59			widget=PicklistWidget(
60			label=_("Methods"),
61			description=_(
62			"Available methods to perform the test"),
63			)
64			)
65
66			Instruments = UIDReferenceField(
67			"Instruments",
68			schemata="Method",
69			required=0,
70			multiValued=1,
71			vocabulary="_instruments_vocabulary",
72			allowed_types=("Instrument", ),
73			accessor="getRawInstruments",
74			widget=PicklistWidget(
75			label=_("Instruments"),
76			description=_("Available instruments based on the selected methods."),
77			)
78			)
79
80			# XXX: HIDDEN -> TO BE REMOVED
81			UseDefaultCalculation = BooleanField(
82			"UseDefaultCalculation",
83			schemata="Method",
84			default=True,
85			widget=BooleanWidget(
86			visible=False,
87			label=_("Use the Default Calculation of Method"),
88			description=_(
89			"Select if the calculation to be used is the calculation set by "
90			"default in the default method. If unselected, the calculation "
91			"can be selected manually"),
92			)
93			)
94
95			Calculation = UIDReferenceField(
96			"Calculation",
97			schemata="Method",
98			required=0,
99			relationship="AnalysisServiceCalculation",
100			vocabulary="_default_calculation_vocabulary",
101			allowed_types=("Calculation", ),
102			accessor="getRawCalculation",
103			widget=SelectionWidget(
104			format="select",
105			label=_("Calculation"),
106			description=_("Calculation to be assigned to this content."),
107			catalog_name=SETUP_CATALOG,
108			base_query={"is_active": True},
109			)
110			)
111
112			InterimFields = InterimFieldsField(
113			"InterimFields",
114			schemata="Result Options",
115			widget=RecordsWidget(
116			label=_("Result variables"),
117			description=_("Additional result values"),
118			)
119			)
120
121			# XXX: HIDDEN -> TO BE REMOVED
122			Separate = BooleanField(
123			"Separate",
124			schemata="Container and Preservation",
125			default=False,
126			required=0,
127			widget=BooleanWidget(
128			visible=False,
129			label=_("Separate Container"),
130			description=_("Check this box to ensure a separate sample container "
131			"is used for this analysis service"),
132			)
133			)
134
135			# XXX: HIDDEN -> TO BE REMOVED
136			Preservation = UIDReferenceField(
137			"Preservation",
138			schemata="Container and Preservation",
139			allowed_types=("SamplePreservation",),
140			vocabulary="getPreservations",
141			required=0,
142			multiValued=0,
143			widget=ReferenceWidget(
144			visible=False,
145			label=_("Default Preservation"),
146			description=_(
147			"Select a default preservation for this analysis service. If the "
148			"preservation depends on the sample type combination, specify a "
149			"preservation per sample type in the table below"),
150			catalog_name=SETUP_CATALOG,
151			base_query={"is_active": True},
152			)
153			)
154
155			# XXX: HIDDEN -> TO BE REMOVED
156			Container = UIDReferenceField(
157			"Container",
158			schemata="Container and Preservation",
159			allowed_types=("Container", "ContainerType"),
160			vocabulary="getContainers",
161			required=0,
162			multiValued=0,
163			widget=ReferenceWidget(
164			visible=False,
165			label=_("Default Container"),
166			description=_(
167			"Select the default container to be used for this analysis "
168			"service. If the container to be used depends on the sample type "
169			"and preservation combination, specify the container in the "
170			"sample type table below"),
171			catalog_name=SETUP_CATALOG,
172			base_query={"is_active": True},
173			)
174			)
175
176			# XXX: HIDDEN -> TO BE REMOVED
177			PartitionSetup = PartitionSetupField(
178			"PartitionSetup",
179			schemata="Container and Preservation",
180			widget=PartitionSetupWidget(
181			visible=False,
182			label=_("Preservation per sample type"),
183			description=_(
184			"Please specify preservations that differ from the analysis "
185			"service's default preservation per sample type here."),
186			)
187			)
188
189			# Allow/disallow the capture of text as the result of the analysis
190			DefaultResult = StringField(
191			"DefaultResult",
192			schemata="Result Options",
193			validators=('service_defaultresult_validator',),
194			widget=StringWidget(
195			label=_("Default result"),
196			description=_(
197			"Default result to display on result entry"
198			)
199			)
200			)
201
202			Conditions = RecordsField(
203			"Conditions",
204			schemata="Advanced",
205			type="conditions",
206			subfields=(
207			"title",
208			"description",
209			"type",
210			"choices",
211			"default",
212			"required",
213			"report",
214			),
215			required_subfields=(
216			"title",
217			"type",
218			),
219			subfield_labels={
220			"title": _("Title"),
221			"description": _("Description"),
222			"type": _("Control type"),
223			"choices": _("Choices"),
224			"default": _("Default value"),
225			"required": _("Required"),
226			"report": _("Report"),
227			},
228			subfield_descriptions={
229			"choices": _("Please use the following format for select options: "
230			"key1:value1\|key2:value2\|...\|keyN:valueN"),
231			},
232			subfield_types={
233			"title": "string",
234			"description": "string",
235			"type": "string",
236			"default": "string",
237			"choices": "string",
238			"required": "boolean",
239			"report": "boolean",
240			},
241			subfield_sizes={
242			"title": 20,
243			"description": 50,
244			"type": 1,
245			"choices": 30,
246			"default": 20,
247			},
248			subfield_validators={
249			"title": "service_conditions_validator",
250			},
251			subfield_maxlength={
252			"title": 50,
253			"description": 200,
254			},
255			subfield_vocabularies={
256			"type": DisplayList((
257			('', ''),
258			('text', _('Text')),
259			('number', _('Number')),
260			('checkbox', _('Checkbox')),
261			('select', _('Select')),
262			('file', _('File upload')),
263			)),
264			},
265			widget=RecordsWidget(
266			label=_("Analysis conditions"),
267			description=_(
268			"Conditions to ask for this analysis on sample registration. For "
269			"instance, laboratory may want the user to input the temperature, "
270			"the ramp and flow when a thermogravimetric (TGA) analysis is "
271			"selected on sample registration. The information provided will be "
272			"later considered by the laboratory personnel when performing the "
273			"test."
274			),
275			)
276			)
277
278
279			schema = schema.copy() + Schema((
280			Methods,
281			Instruments,
282			UseDefaultCalculation,
283			Calculation,
284			InterimFields,
285			Separate,
286			Preservation,
287			Container,
288			PartitionSetup,
289			DefaultResult,
290			Conditions,
291			))
292
293			# Move default method field after available methods field
294			schema.moveField("Method", after="Methods")
295			# Move default instrument field after available instruments field
296			schema.moveField("Instrument", after="Instruments")
297			# Move default result field after Result Options
298			schema.moveField("DefaultResult", after="ResultOptions")
299
300
301			class AnalysisService(AbstractBaseAnalysis):
302			"""Analysis Service Content Holder
303			"""
304			implements(IAnalysisService, IDeactivable)
305			security = ClassSecurityInfo()
306			schema = schema
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-01-11 23:48 UTC by Report Bug Copy Issue Report The variable `schema` does not seem to be defined. Loading history...
307			_at_rename_after_creation = True
308
309			def _renameAfterCreation(self, check_auto_id=False):
310			from senaite.core.idserver import renameAfterCreation
311
312			return renameAfterCreation(self)
313
314			def getMethods(self):
315			"""Returns the assigned methods
316
317			:returns: List of method objects
318			"""
319			return self.getField("Methods").get(self)
320
321			def getRawMethods(self):
322			"""Returns the assigned method UIDs
323
324			:returns: List of method UIDs
325			"""
326			return self.getField("Methods").getRaw(self)
327
328			def getMethod(self):
329			"""Get the default method
330			"""
331			field = self.getField("Method")
332			method = field.get(self)
333			if not method:
334			return None
335			# check if the method is in the selected methods
336			methods = self.getMethods()
337			if method not in methods:
338			return None
339			return method
340
341			def getRawMethod(self):
342			"""Returns the UID of the default method
343
344			:returns: method UID
345			"""
346			field = self.getField("Method")
347			method = field.getRaw(self)
348			if not method:
349			return None
350			# check if the method is in the selected methods
351			methods = self.getRawMethods()
352			if method not in methods:
353			return None
354			return method
355
356			def getInstruments(self):
357			"""Returns the assigned instruments
358
359			:returns: List of instrument objects
360			"""
361			return self.getField("Instruments").get(self)
362
363			def getRawInstruments(self):
364			"""List of assigned Instrument UIDs
365			"""
366			return self.getField("Instruments").getRaw(self)
367
368			def getInstrument(self):
369			"""Return the default instrument
370			"""
371			field = self.getField("Instrument")
372			instrument = field.get(self)
373			if not instrument:
374			return None
375			# check if the instrument is in the selected instruments
376			instruments = self.getInstruments()
377			if instrument not in instruments:
378			return None
379			return instrument
380
381			def getRawInstrument(self):
382			"""Return the UID of the default instrument
383			"""
384			field = self.getField("Instrument")
385			instrument = field.getRaw(self)
386			if not instrument:
387			return None
388			# check if the instrument is in the selected instruments
389			instruments = self.getRawInstruments()
390			if instrument not in instruments:
391			return None
392			return instrument
393
394			def getCalculation(self):
395			"""Get the default calculation
396			"""
397			field = self.getField("Calculation")
398			calculation = field.get(self)
399			if not calculation:
400			return None
401			return calculation
402
403			def getRawCalculation(self):
404			"""Returns the UID of the assigned calculation
405
406			:returns: Calculation UID
407			"""
408			field = self.getField("Calculation")
409			calculation = field.getRaw(self)
410			if not calculation:
411			return None
412			return calculation
413
414			def getServiceDependencies(self):
415			"""Return calculation dependencies of the service
416
417			:return: a list of analysis services objects.
418			"""
419			calc = self.getCalculation()
420			if calc:
421			return calc.getCalculationDependencies(flat=True)
422			return []
423
424			def getServiceDependenciesUIDs(self):
425			"""Return calculation dependency UIDs of the service
426
427			:return: a list of uids
428			"""
429			return map(api.get_uid, self.getServiceDependencies())
430
431			def getServiceDependents(self):
432			"""Return services depending on us
433			"""
434			catalog = api.get_tool(SETUP_CATALOG)
435			active_calcs = catalog(portal_type="Calculation", is_active=True)
436			calculations = map(api.get_object, active_calcs)
437			dependents = []
438			for calc in calculations:
439			calc_dependents = calc.getDependentServices()
440			if self in calc_dependents:
441			calc_dependencies = calc.getCalculationDependents()
442			dependents = dependents + calc_dependencies
443			dependents = list(set(dependents))
444			if self in dependents:
445			dependents.remove(self)
446			return dependents
447
448			def getServiceDependentsUIDs(self):
449			"""Return service UIDs depending on us
450			"""
451			return map(api.get_uid, self.getServiceDependents())
452
453			def query_available_methods(self):
454			"""Return all available methods
455			"""
456			catalog = api.get_tool(SETUP_CATALOG)
457			query = {
458			"portal_type": "Method",
459			"is_active": True,
460			"sort_on": "sortable_title",
461			"sort_order": "ascending",
462			}
463			return catalog(query)
464
465			def _methods_vocabulary(self):
466			"""Vocabulary used for methods field
467			"""
468			methods = self.query_available_methods()
469			items = [(api.get_uid(m), api.get_title(m)) for m in methods]
470			dlist = DisplayList(items)
471			return dlist
472
473			def _default_method_vocabulary(self):
474			"""Vocabulary used for default method field
475			"""
476			# check if we selected methods
477			methods = self.getMethods()
478			if not methods:
479			# query all available methods
480			methods = self.query_available_methods()
481			items = [(api.get_uid(m), api.get_title(m)) for m in methods]
482			dlist = DisplayList(items).sortedByValue()
483			# allow to leave this field empty
484			dlist.add("", _("None"))
485			return dlist
486
487			def query_available_instruments(self):
488			"""Return all available Instruments
489			"""
490			catalog = api.get_tool(SETUP_CATALOG)
491			query = {
492			"portal_type": "Instrument",
493			"is_active": True,
494			"sort_on": "sortable_title",
495			"sort_order": "ascending",
496			}
497			return catalog(query)
498
499			def _instruments_vocabulary(self):
500			"""Vocabulary used for instruments field
501			"""
502			instruments = []
503			# When methods are selected, display only instruments from the methods
504			methods = self.getMethods()
505			for method in methods:
506			for instrument in method.getInstruments():
507			if instrument in instruments:
508			continue
509			instruments.append(instrument)
510
511			if not methods:
512			# query all available instruments when no methods are selected
513			instruments = self.query_available_instruments()
514
515			items = [(api.get_uid(i), api.get_title(i)) for i in instruments]
516			dlist = DisplayList(items)
517			return dlist
518
519			def _default_instrument_vocabulary(self):
520			"""Vocabulary used for default instrument field
521			"""
522			# check if we selected instruments
523			instruments = self.getInstruments()
524			items = [(api.get_uid(i), api.get_title(i)) for i in instruments]
525			dlist = DisplayList(items).sortedByValue()
526			# allow to leave this field empty
527			dlist.add("", _("None"))
528			return dlist
529
530			def query_available_calculations(self):
531			"""Return all available calculations
532			"""
533			catalog = api.get_tool(SETUP_CATALOG)
534			query = {
535			"portal_type": "Calculation",
536			"is_active": True,
537			"sort_on": "sortable_title",
538			"sort_order": "ascending",
539			}
540			return catalog(query)
541
542			def get_methods_calculations(self):
543			"""Return calculations assigned to the selected methods
544			"""
545			methods = self.getMethods()
546			if not methods:
547			return None
548			methods_calcs = map(lambda m: m.getCalculations(), methods)
549			return list(itertools.chain(*methods_calcs))
550
551			def _default_calculation_vocabulary(self):
552			"""Vocabulary used for the default calculation field
553			"""
554			calculations = self.get_methods_calculations()
555			if calculations is None:
556			# query all available calculations
557			calculations = self.query_available_calculations()
558			items = [(api.get_uid(c), api.get_title(c)) for c in calculations]
559			# allow to leave this field empty
560			dlist = DisplayList(items).sortedByValue()
561			dlist.add("", _("None"))
562			return dlist
563
564			# XXX DECIDE IF NEEDED
565			# --------------------
566
567			def getContainers(self, instance=None):
568			# On first render, the containers must be filtered
569			instance = instance or self
570			separate = self.getSeparate()
571			containers = getContainers(instance,
572			allow_blank=True,
573			show_container_types=not separate,
574			show_containers=separate)
575			return DisplayList(containers)
576
577			def getPreservations(self):
578			query = {
579			"portal_type": "SamplePreservation",
580			"is_active": True,
581			"sort_on": "sortable_title",
582			"sort_order": "ascending",
583			}
584			brains = api.search(query, SETUP_CATALOG)
585			items = [(brain.UID, brain.Title) for brain in brains]
586			return DisplayList(items)
587
588			def getAvailableMethods(self):
589			"""Returns the methods available for this analysis.
590			"""
591			return self.getMethods()
592
593			def getAvailableMethodUIDs(self):
594			"""Returns the UIDs of the available methods
595
596			Used here:
597			bika.lims.catalog.indexers.bikasetup.method_available_uid
598			"""
599			# N.B. we return a copy of the list to avoid accidental writes
600			return self.getRawMethods()[:]
601
602
603			registerType(AnalysisService, PROJECTNAME)
604

senaite / senaite.core

Push — 2.x ( 3a5aed...187ea7 )

AnalysisService.getServiceDependents() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like