bika.lims.browser.worksheet.views.results.ManageResultsView.getInstruments() - Code Metrics - Inspection of "New worksheet results listing (#1095)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 8785a0...af2087 )

by Jordi

created 2018-11-22 20:24 UTC

ManageResultsView.getInstruments() A

↳ Parent: bika.lims.browser.worksheet.views.results

Complexity

Conditions

Size

Total Lines	12
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	10
dl	0
loc	12
rs	9.9
c	0
b	0
f	0
cc	4
nop	1

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE
#
# Copyright 2018 by it's authors.
# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.

from bika.lims import api
from bika.lims import bikaMessageFactory as _
from bika.lims.browser import BrowserView
from bika.lims.browser.worksheet.tools import showRejectionMessage
from bika.lims.config import WORKSHEET_LAYOUT_OPTIONS
from bika.lims.utils import getUsers
from plone.app.layout.globals.interfaces import IViewView
from Products.Archetypes.public import DisplayList
from Products.CMFCore.utils import getToolByName
from Products.CMFPlone.utils import safe_unicode
from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
from zope.interface import implements
from plone.memoize import view
from bika.lims.permissions import EditWorksheet
from bika.lims.permissions import ManageWorksheets
from bika.lims.browser.worksheet.views import AnalysesTransposedView


class ManageResultsView(BrowserView):
    """Worksheet Manage Results View
    """
    implements(IViewView)
    template = ViewPageTemplateFile("../templates/results.pt")

    def __init__(self, context, request):
        super(ManageResultsView, self).__init__(context, request)
        self.icon = "{}/{}".format(
            self.portal_url,
            "/++resource++bika.lims.images/worksheet_big.png"
        )

        self.layout_displaylist = WORKSHEET_LAYOUT_OPTIONS

    def __call__(self):
        # TODO: Refactor Worfklow
        grant = self.is_edit_allowed() and self.is_manage_allowed()
        if not grant:
            redirect_url = api.get_url(self.context)
            return self.request.response.redirect(redirect_url)
        # TODO: Refactor this function call
        showRejectionMessage(self.context)

        # Save the results layout
        rlayout = self.request.get("resultslayout", "")
        if rlayout and rlayout in WORKSHEET_LAYOUT_OPTIONS.keys() \
           and rlayout != self.context.getResultsLayout():
            self.context.setResultsLayout(rlayout)
            message = _("Changes saved.")
            self.context.plone_utils.addPortalMessage(message, "info")

        # Classic/Transposed View Switch
        if self.context.getResultsLayout() == "1":
            view = "analyses_classic_view"
            self.Analyses = api.get_view(
                view, context=self.context, request=self.request)
        else:
            # TODO: Refactor transposed view to new listing
            self.Analyses = AnalysesTransposedView(self.context, self.request)

        self.analystname = self.context.getAnalystName()
        self.instrumenttitle = self.get_instrument_title()

        # Check if the instruments used are valid
        self.checkInstrumentsValidity()

        return self.template()

    def get_analysts(self):
        """Returns Analysts
        """
        roles = ["Manager", "LabManager", "Analyst"]
        return getUsers(self.context, roles)

    @view.memoize
    def get_instrument_title(self):
        """Return the current instrument title
        """
        instrument = self.context.getInstrument()
        if not instrument:
            return ""
        return api.get_title(instrument)

    @view.memoize
    def is_edit_allowed(self):
        """Check if edit is allowed
        """
        checkPermission = self.context.portal_membership.checkPermission
        return checkPermission(EditWorksheet, self.context)

    @view.memoize
    def is_manage_allowed(self):
        """Check if manage is allowed
        """
        checkPermission = self.context.portal_membership.checkPermission
        return checkPermission(ManageWorksheets, self.context)

    @view.memoize
    def is_assignment_allowed(self):
        """Check if analyst assignment is allowed
        """
        if not self.is_manage_allowed():
            return False
        review_state = api.get_workflow_status_of(self.context)
        edit_states = ["open", "attachment_due", "to_be_verified"]
        return review_state in edit_states

    def getInstruments(self):
        # TODO: Return only the allowed instruments for at least one contained
        # analysis
        bsc = getToolByName(self, 'bika_setup_catalog')
        items = [('', '')] + [(o.UID, o.Title) for o in
                              bsc(portal_type='Instrument',
                                  inactive_state='active')]
        o = self.context.getInstrument()
        if o and o.UID() not in [i[0] for i in items]:
            items.append((o.UID(), o.Title()))
        items.sort(lambda x, y: cmp(x[1].lower(), y[1].lower()))
        return DisplayList(list(items))

    def get_wide_interims(self):
        """Returns a dictionary with the analyses services from the current
        worksheet which have at least one interim with 'Wide' attribute set to
        true and state 'sample_received'

        The structure of the returned dictionary is the following:
        <Analysis_keyword>: {
            'analysis': <Analysis_name>,
            'keyword': <Analysis_keyword>,
            'interims': {
                <Interim_keyword>: {
                    'value': <Interim_default_value>,
                    'keyword': <Interim_key>,
                    'title': <Interim_title>
                }
            }
        }
        """
        outdict = {}
        allowed_states = ["sample_received"]
        for analysis in self._getAnalyses():
            if api.get_workflow_status_of(analysis) not in allowed_states:
                continue

            if analysis.getKeyword() in outdict.keys():
                continue

            calculation = analysis.getCalculation()
            if not calculation:
                continue

            andict = {
                "analysis": analysis.Title(),
                "keyword": analysis.getKeyword(),
                "interims": {}
            }

            # Analysis Service interim defaults
            for field in analysis.getInterimFields():
                if field.get("wide", False):
                    andict["interims"][field["keyword"]] = field

            # Interims from calculation
            for field in calculation.getInterimFields():
                if field["keyword"] not in andict["interims"].keys() \
                   and field.get("wide", False):
                    andict["interims"][field["keyword"]] = field

            if andict["interims"]:
                outdict[analysis.getKeyword()] = andict
        return outdict

    def checkInstrumentsValidity(self):

        """Checks the validity of the instruments used in the Analyses If an
        analysis with an invalid instrument (out-of-date or with calibration
        tests failed) is found, a warn message will be displayed.
        """
        invalid = []
        ans = self._getAnalyses()
        for an in ans:
            valid = an.isInstrumentValid()
            if not valid:
                instrument = an.getInstrument()
                inv = "%s (%s)" % (
                    safe_unicode(an.Title()), safe_unicode(instrument.Title()))
                if inv not in invalid:
                    invalid.append(inv)
        if len(invalid) > 0:
            message = _("Some analyses use out-of-date or uncalibrated "
                        "instruments. Results edition not allowed")
            message = "%s: %s" % (message, (", ".join(invalid)))
            self.context.plone_utils.addPortalMessage(message, "warn")

    def _getAnalyses(self):
        """
        This function returns a list with the analyses related to the worksheet
        and filtered by the current selected department in the department
        porlet.
        @returna list of analyses objects.
        """
        ans = [a for a in self.context.getAnalyses() if self._isItemAllowed(a)]
        return ans

    def _isItemAllowed(self, obj):
        """
        It checks if the analysis service can be added to the list depending
        on the department filter. If the analysis service is not assigned to a
        department, show it.
        If department filtering is disabled in bika_setup, will return True.
        @Obj: it is an analysis object.
        @return: boolean
        """
        if not self.context.bika_setup.getAllowDepartmentFiltering():
            return True
        # Gettin the department from analysis service
        serv_dep = obj.getDepartment()
        result = True
        if serv_dep:
            # Getting the cookie value
            cookie_dep_uid = self.request.get('filter_by_department_info', '')
            # Comparing departments' UIDs
            result = True if serv_dep.UID() in\
                cookie_dep_uid.split(',') else False
        return result


1		# -- coding: utf-8 --
2		#
3		# This file is part of SENAITE.CORE
4		#
5		# Copyright 2018 by it's authors.
6		# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.
7
8		from bika.lims import api
9		from bika.lims import bikaMessageFactory as _
10		from bika.lims.browser import BrowserView
11		from bika.lims.browser.worksheet.tools import showRejectionMessage
12		from bika.lims.config import WORKSHEET_LAYOUT_OPTIONS
13		from bika.lims.utils import getUsers
14		from plone.app.layout.globals.interfaces import IViewView
15		from Products.Archetypes.public import DisplayList
16		from Products.CMFCore.utils import getToolByName
17		from Products.CMFPlone.utils import safe_unicode
18		from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
19		from zope.interface import implements
20		from plone.memoize import view
21		from bika.lims.permissions import EditWorksheet
22		from bika.lims.permissions import ManageWorksheets
23		from bika.lims.browser.worksheet.views import AnalysesTransposedView
24
25
26		class ManageResultsView(BrowserView):
27		"""Worksheet Manage Results View
28		"""
29		implements(IViewView)
30		template = ViewPageTemplateFile("../templates/results.pt")
31
32		def __init__(self, context, request):
33		super(ManageResultsView, self).__init__(context, request)
34		self.icon = "{}/{}".format(
35		self.portal_url,
36		"/++resource++bika.lims.images/worksheet_big.png"
37		)
38
39		self.layout_displaylist = WORKSHEET_LAYOUT_OPTIONS
40
41		def __call__(self):
42		# TODO: Refactor Worfklow
43		grant = self.is_edit_allowed() and self.is_manage_allowed()
44		if not grant:
45		redirect_url = api.get_url(self.context)
46		return self.request.response.redirect(redirect_url)
47		# TODO: Refactor this function call
48		showRejectionMessage(self.context)
49
50		# Save the results layout
51		rlayout = self.request.get("resultslayout", "")
52		if rlayout and rlayout in WORKSHEET_LAYOUT_OPTIONS.keys() \
53		and rlayout != self.context.getResultsLayout():
54		self.context.setResultsLayout(rlayout)
55		message = _("Changes saved.")
56		self.context.plone_utils.addPortalMessage(message, "info")
57
58		# Classic/Transposed View Switch
59		if self.context.getResultsLayout() == "1":
60		view = "analyses_classic_view"
61		self.Analyses = api.get_view(
62		view, context=self.context, request=self.request)
63		else:
64		# TODO: Refactor transposed view to new listing
65		self.Analyses = AnalysesTransposedView(self.context, self.request)
66
67		self.analystname = self.context.getAnalystName()
68		self.instrumenttitle = self.get_instrument_title()
69
70		# Check if the instruments used are valid
71		self.checkInstrumentsValidity()
72
73		return self.template()
74
75		def get_analysts(self):
76		"""Returns Analysts
77		"""
78		roles = ["Manager", "LabManager", "Analyst"]
79		return getUsers(self.context, roles)
80
81		@view.memoize
82		def get_instrument_title(self):
83		"""Return the current instrument title
84		"""
85		instrument = self.context.getInstrument()
86		if not instrument:
87		return ""
88		return api.get_title(instrument)
89
90		@view.memoize
91		def is_edit_allowed(self):
92		"""Check if edit is allowed
93		"""
94		checkPermission = self.context.portal_membership.checkPermission
95		return checkPermission(EditWorksheet, self.context)
96
97		@view.memoize
98		def is_manage_allowed(self):
99		"""Check if manage is allowed
100		"""
101		checkPermission = self.context.portal_membership.checkPermission
102		return checkPermission(ManageWorksheets, self.context)
103
104		@view.memoize
105		def is_assignment_allowed(self):
106		"""Check if analyst assignment is allowed
107		"""
108		if not self.is_manage_allowed():
109		return False
110		review_state = api.get_workflow_status_of(self.context)
111		edit_states = ["open", "attachment_due", "to_be_verified"]
112		return review_state in edit_states
113
114		def getInstruments(self):
115		# TODO: Return only the allowed instruments for at least one contained
116		# analysis
117		bsc = getToolByName(self, 'bika_setup_catalog')
118		items = [('', '')] + [(o.UID, o.Title) for o in
119		bsc(portal_type='Instrument',
120		inactive_state='active')]
121		o = self.context.getInstrument()
122		if o and o.UID() not in [i[0] for i in items]:
123		items.append((o.UID(), o.Title()))
124		items.sort(lambda x, y: cmp(x[1].lower(), y[1].lower()))
125		return DisplayList(list(items))
126
127		def get_wide_interims(self):
128		"""Returns a dictionary with the analyses services from the current
129		worksheet which have at least one interim with 'Wide' attribute set to
130		true and state 'sample_received'
131
132		The structure of the returned dictionary is the following:
133		<Analysis_keyword>: {
134		'analysis': <Analysis_name>,
135		'keyword': <Analysis_keyword>,
136		'interims': {
137		<Interim_keyword>: {
138		'value': <Interim_default_value>,
139		'keyword': <Interim_key>,
140		'title': <Interim_title>
141		}
142		}
143		}
144		"""
145		outdict = {}
146		allowed_states = ["sample_received"]
147		for analysis in self._getAnalyses():
148		if api.get_workflow_status_of(analysis) not in allowed_states:
149		continue
150
151		if analysis.getKeyword() in outdict.keys():
152		continue
153
154		calculation = analysis.getCalculation()
155		if not calculation:
156		continue
157
158		andict = {
159		"analysis": analysis.Title(),
160		"keyword": analysis.getKeyword(),
161		"interims": {}
162		}
163
164		# Analysis Service interim defaults
165		for field in analysis.getInterimFields():
166		if field.get("wide", False):
167		andict["interims"][field["keyword"]] = field
168
169		# Interims from calculation
170		for field in calculation.getInterimFields():
171		if field["keyword"] not in andict["interims"].keys() \
172		and field.get("wide", False):
173		andict["interims"][field["keyword"]] = field
174
175		if andict["interims"]:
176		outdict[analysis.getKeyword()] = andict
177		return outdict
178
179	View Code Duplication	def checkInstrumentsValidity(self):
		0 ignored issues – show Duplication introduced 2017-12-01 14:39 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
180		"""Checks the validity of the instruments used in the Analyses If an
181		analysis with an invalid instrument (out-of-date or with calibration
182		tests failed) is found, a warn message will be displayed.
183		"""
184		invalid = []
185		ans = self._getAnalyses()
186		for an in ans:
187		valid = an.isInstrumentValid()
188		if not valid:
189		instrument = an.getInstrument()
190		inv = "%s (%s)" % (
191		safe_unicode(an.Title()), safe_unicode(instrument.Title()))
192		if inv not in invalid:
193		invalid.append(inv)
194		if len(invalid) > 0:
195		message = _("Some analyses use out-of-date or uncalibrated "
196		"instruments. Results edition not allowed")
197		message = "%s: %s" % (message, (", ".join(invalid)))
198		self.context.plone_utils.addPortalMessage(message, "warn")
199
200		def _getAnalyses(self):
201		"""
202		This function returns a list with the analyses related to the worksheet
203		and filtered by the current selected department in the department
204		porlet.
205		@returna list of analyses objects.
206		"""
207		ans = [a for a in self.context.getAnalyses() if self._isItemAllowed(a)]
208		return ans
209
210		def _isItemAllowed(self, obj):
211		"""
212		It checks if the analysis service can be added to the list depending
213		on the department filter. If the analysis service is not assigned to a
214		department, show it.
215		If department filtering is disabled in bika_setup, will return True.
216		@Obj: it is an analysis object.
217		@return: boolean
218		"""
219		if not self.context.bika_setup.getAllowDepartmentFiltering():
220		return True
221		# Gettin the department from analysis service
222		serv_dep = obj.getDepartment()
223		result = True
224		if serv_dep:
225		# Getting the cookie value
226		cookie_dep_uid = self.request.get('filter_by_department_info', '')
227		# Comparing departments' UIDs
228		result = True if serv_dep.UID() in\
229		cookie_dep_uid.split(',') else False
230		return result
231

senaite / senaite.core

Push — master ( 8785a0...af2087 )

ManageResultsView.getInstruments() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like