build.bika.lims.browser.analysisservice.JSONReadExtender.__init__() - Code Metrics - Inspection of "Audit log (#1324)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 804f45...1fa2c0 )

by Jordi

created 2019-04-22 21:47 UTC

JSONReadExtender.init() A

↳ Parent: build.bika.lims.browser.analysisservice

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	2
rs	10
c	0
b	0
f	0
cc	1
nop	2

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE.
#
# SENAITE.CORE is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright 2018-2019 by it's authors.
# Some rights reserved, see README and LICENSE.

import json

import plone
import plone.protect
from bika.lims import api
from bika.lims.api.security import check_permission
from bika.lims.browser import BrowserView
from bika.lims.config import POINTS_OF_CAPTURE
from bika.lims.content.analysisservice import getContainers
from bika.lims.interfaces import IAnalysisService
from bika.lims.interfaces import IJSONReadExtender
from bika.lims.jsonapi import get_include_fields
from bika.lims.jsonapi import load_field_values
from bika.lims.permissions import ViewLogTab
from bika.lims.utils import get_image
from magnitude import mg
from plone.memoize import view
from Products.CMFCore.utils import getToolByName
from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
from zope.component import adapts
from zope.i18n.locales import locales
from zope.interface import implements


class AnalysisServiceInfoView(BrowserView):
    """Show details of the Analysis Service
    """
    template = ViewPageTemplateFile("templates/analysisservice_info.pt")

    def __init__(self, context, request):
        super(AnalysisServiceInfoView, self).__init__(context, request)

    def __call__(self):
        # disable the editable border
        self.request.set("disable_border", 1)
        return self.template()

    @view.memoize
    def show_prices(self):
        """Checks if prices should be shown or not
        """
        setup = api.get_setup()
        return setup.getShowPrices()

    @view.memoize
    def get_currency_symbol(self):
        """Get the currency Symbol
        """
        locale = locales.getLocale('en')
        setup = api.get_setup()
        currency = setup.getCurrency()
        return locale.numbers.currencies[currency].symbol

    def get_icon_for(self, typename):
        image = "{}_big.png".format(typename)
        return get_image(
            image, width="20px", style="vertical-align: baseline;")

    @view.memoize
    def get_service(self):
        service_uid = self.request.form.get("service_uid")
        return api.get_object_by_uid(service_uid, None)

    @view.memoize
    def get_service_url(self):
        service = self.get_service()
        return api.get_url(service)

    @view.memoize
    def is_accredited(self):
        service = self.get_service()
        return service.getAccredited()

    def get_analysis_or_service(self):
        analysis = self.get_analysis()
        if analysis:
            return analysis
        return self.get_service()

    def get_methods(self):
        if not self.get_service():
            return None
        return self.get_service().getMethods()

    @view.memoize
    def get_analysis(self):
        analysis_uid = self.request.form.get("analysis_uid")
        return api.get_object_by_uid(analysis_uid, None)

    @view.memoize
    def get_calculation(self):
        if not self.get_analysis_or_service():
            return None
        return self.get_analysis_or_service().getCalculation()

    def get_dependent_services(self):
        if not self.get_calculation():
            return []
        return self.get_calculation().getDependentServices()

    def get_calculation_dependencies(self):
        if not self.get_calculation():
            return []
        return self.get_calculation().getCalculationDependencies()

    def can_view_logs_of(self, obj):
        """Checks if the current user is allowed to see the logs
        """
        return check_permission(ViewLogTab, obj)

    def analysis_log_view(self):
        """Get the log view of the requested analysis
        """
        service = self.get_analysis_or_service()
        if not self.can_view_logs_of(service):
            return None
        view = api.get_view("auditlog", context=service, request=self.request)
        view.update()
        view.before_render()
        return view


class ajaxGetContainers(BrowserView):
    """ajax Preservation/Container widget filter
    request values:
    - allow_blank: print blank value in return
    - show_container_types
    - show_containers
    - minvol: magnitude (string).
    """
    def __call__(self):
        plone.protect.CheckAuthenticator(self.request)

        allow_blank = self.request.get('allow_blank', False) == 'true'
        show_container_types = json.loads(self.request.get('show_container_types', 'true'))
        show_containers = json.loads(self.request.get('show_containers', 'true'))
        minvol = self.request.get("minvol", "0")
        try:
            minvol = minvol.split()
            minvol = mg(float(minvol[0]), " ".join(minvol[1:]))
        except:
            minvol = mg(0)

        containers = getContainers(
            self.context,
            minvol=minvol,
            allow_blank=allow_blank,
            show_containers=show_containers,
            show_container_types=show_container_types,
        )

        return json.dumps(containers)


class ajaxGetServiceInterimFields:
    "Tries to fall back to Calculation for defaults"

    def __init__(self, context, request):
        self.context = context
        self.request = request

    def __call__(self):
        plone.protect.CheckAuthenticator(self.request)
        plone.protect.PostOnly(self.request)
        bsc = getToolByName(self.context, 'bika_setup_catalog')
        service_url = self.request['service_url']
        service_id = service_url.split('/')[-1]
        services = bsc(portal_type='AnalysisService', id=service_id)
        if services:
            service = services[0].getObject()
            service_interims = service.getInterimFields()
        else:
            # portal_factory has no service
            return

        calc = service.getCalculation()
        if calc:
            calc_interims = calc.getInterimFields()
        else:
            calc_interims = []

        # overwrite existing fields in position
        for s_i in service_interims:
            placed = 0
            for c_i in calc_interims:
                if s_i['keyword'] == c_i['keyword']:
                    c_i['value'] = s_i['value']
                    placed = 1
                    break
            if placed:
                continue
            # otherwise, create new ones (last)
            calc_interims.append(s_i)

        return json.dumps(calc_interims)


class JSONReadExtender(object):
    """- Adds fields to Analysis Service:

    MethodInstruments - A dictionary of instruments:
        keys: Method UID
        values: list of instrument UIDs

    """

    implements(IJSONReadExtender)
    adapts(IAnalysisService)

    def __init__(self, context):
        self.context = context

    def service_info(self, service):
        ret = {
            "Category": service.getCategory().Title(),
            "Category_uid": service.getCategory().UID(),
            "Service": service.Title(),
            "Service_uid": service.UID(),
            "Keyword": service.getKeyword(),
            "PointOfCapture": service.getPointOfCapture(),
            "PointOfCapture_title": POINTS_OF_CAPTURE.getValue(service.getPointOfCapture()),
        }
        return ret

    def __call__(self, request, data):
        include_fields = get_include_fields(request)

        if not include_fields or "MethodInstruments" in include_fields:
            data["MethodInstruments"] = {}
            for method in self.context.getAvailableMethods():
                for instrument in method.getInstruments():
                    if method.UID() not in data["MethodInstruments"]:
                        data["MethodInstruments"][method.UID()] = []
                    data["MethodInstruments"][method.UID()].append(
                        load_field_values(instrument, include_fields=[]))


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE.
4			#
5			# SENAITE.CORE is free software: you can redistribute it and/or modify it under
6			# the terms of the GNU General Public License as published by the Free Software
7			# Foundation, version 2.
8			#
9			# This program is distributed in the hope that it will be useful, but WITHOUT
10			# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11			# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
12			# details.
13			#
14			# You should have received a copy of the GNU General Public License along with
15			# this program; if not, write to the Free Software Foundation, Inc., 51
16			# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17			#
18			# Copyright 2018-2019 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			import json
22
23			import plone
24			import plone.protect
25			from bika.lims import api
26			from bika.lims.api.security import check_permission
27			from bika.lims.browser import BrowserView
28			from bika.lims.config import POINTS_OF_CAPTURE
29			from bika.lims.content.analysisservice import getContainers
30			from bika.lims.interfaces import IAnalysisService
31			from bika.lims.interfaces import IJSONReadExtender
32			from bika.lims.jsonapi import get_include_fields
33			from bika.lims.jsonapi import load_field_values
34			from bika.lims.permissions import ViewLogTab
35			from bika.lims.utils import get_image
36			from magnitude import mg
37			from plone.memoize import view
38			from Products.CMFCore.utils import getToolByName
39			from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
40			from zope.component import adapts
41			from zope.i18n.locales import locales
42			from zope.interface import implements
43
44
45			class AnalysisServiceInfoView(BrowserView):
46			"""Show details of the Analysis Service
47			"""
48			template = ViewPageTemplateFile("templates/analysisservice_info.pt")
49
50			def __init__(self, context, request):
51			super(AnalysisServiceInfoView, self).__init__(context, request)
52
53			def __call__(self):
54			# disable the editable border
55			self.request.set("disable_border", 1)
56			return self.template()
57
58			@view.memoize
59			def show_prices(self):
60			"""Checks if prices should be shown or not
61			"""
62			setup = api.get_setup()
63			return setup.getShowPrices()
64
65			@view.memoize
66			def get_currency_symbol(self):
67			"""Get the currency Symbol
68			"""
69			locale = locales.getLocale('en')
70			setup = api.get_setup()
71			currency = setup.getCurrency()
72			return locale.numbers.currencies[currency].symbol
73
74			def get_icon_for(self, typename):
75			image = "{}_big.png".format(typename)
76			return get_image(
77			image, width="20px", style="vertical-align: baseline;")
78
79			@view.memoize
80			def get_service(self):
81			service_uid = self.request.form.get("service_uid")
82			return api.get_object_by_uid(service_uid, None)
83
84			@view.memoize
85			def get_service_url(self):
86			service = self.get_service()
87			return api.get_url(service)
88
89			@view.memoize
90			def is_accredited(self):
91			service = self.get_service()
92			return service.getAccredited()
93
94			def get_analysis_or_service(self):
95			analysis = self.get_analysis()
96			if analysis:
97			return analysis
98			return self.get_service()
99
100			def get_methods(self):
101			if not self.get_service():
102			return None
103			return self.get_service().getMethods()
104
105			@view.memoize
106			def get_analysis(self):
107			analysis_uid = self.request.form.get("analysis_uid")
108			return api.get_object_by_uid(analysis_uid, None)
109
110			@view.memoize
111			def get_calculation(self):
112			if not self.get_analysis_or_service():
113			return None
114			return self.get_analysis_or_service().getCalculation()
115
116			def get_dependent_services(self):
117			if not self.get_calculation():
118			return []
119			return self.get_calculation().getDependentServices()
120
121			def get_calculation_dependencies(self):
122			if not self.get_calculation():
123			return []
124			return self.get_calculation().getCalculationDependencies()
125
126			def can_view_logs_of(self, obj):
127			"""Checks if the current user is allowed to see the logs
128			"""
129			return check_permission(ViewLogTab, obj)
130
131			def analysis_log_view(self):
132			"""Get the log view of the requested analysis
133			"""
134			service = self.get_analysis_or_service()
135			if not self.can_view_logs_of(service):
136			return None
137			view = api.get_view("auditlog", context=service, request=self.request)
138			view.update()
139			view.before_render()
140			return view
141
142
143			class ajaxGetContainers(BrowserView):
144			"""ajax Preservation/Container widget filter
145			request values:
146			- allow_blank: print blank value in return
147			- show_container_types
148			- show_containers
149			- minvol: magnitude (string).
150			"""
151			def __call__(self):
152			plone.protect.CheckAuthenticator(self.request)
153
154			allow_blank = self.request.get('allow_blank', False) == 'true'
155			show_container_types = json.loads(self.request.get('show_container_types', 'true'))
156			show_containers = json.loads(self.request.get('show_containers', 'true'))
157			minvol = self.request.get("minvol", "0")
158			try:
159			minvol = minvol.split()
160			minvol = mg(float(minvol[0]), " ".join(minvol[1:]))
161			except:
162			minvol = mg(0)
163
164			containers = getContainers(
165			self.context,
166			minvol=minvol,
167			allow_blank=allow_blank,
168			show_containers=show_containers,
169			show_container_types=show_container_types,
170			)
171
172			return json.dumps(containers)
173
174
175			class ajaxGetServiceInterimFields:
176			"Tries to fall back to Calculation for defaults"
177
178			def __init__(self, context, request):
179			self.context = context
180			self.request = request
181
182			def __call__(self):
183			plone.protect.CheckAuthenticator(self.request)
184			plone.protect.PostOnly(self.request)
185			bsc = getToolByName(self.context, 'bika_setup_catalog')
186			service_url = self.request['service_url']
187			service_id = service_url.split('/')[-1]
188			services = bsc(portal_type='AnalysisService', id=service_id)
189			if services:
190			service = services[0].getObject()
191			service_interims = service.getInterimFields()
192			else:
193			# portal_factory has no service
194			return
195
196			calc = service.getCalculation()
197			if calc:
198			calc_interims = calc.getInterimFields()
199			else:
200			calc_interims = []
201
202			# overwrite existing fields in position
203			for s_i in service_interims:
204			placed = 0
205			for c_i in calc_interims:
206			if s_i['keyword'] == c_i['keyword']:
207			c_i['value'] = s_i['value']
208			placed = 1
209			break
210			if placed:
211			continue
212			# otherwise, create new ones (last)
213			calc_interims.append(s_i)
214
215			return json.dumps(calc_interims)
216
217
218			class JSONReadExtender(object):
219			"""- Adds fields to Analysis Service:
220
221			MethodInstruments - A dictionary of instruments:
222			keys: Method UID
223			values: list of instrument UIDs
224
225			"""
226
227			implements(IJSONReadExtender)
228			adapts(IAnalysisService)
229
230			def __init__(self, context):
231			self.context = context
232
233			def service_info(self, service):
234			ret = {
235			"Category": service.getCategory().Title(),
236			"Category_uid": service.getCategory().UID(),
237			"Service": service.Title(),
238			"Service_uid": service.UID(),
239			"Keyword": service.getKeyword(),
240			"PointOfCapture": service.getPointOfCapture(),
241			"PointOfCapture_title": POINTS_OF_CAPTURE.getValue(service.getPointOfCapture()),
242			}
243			return ret
244
245			def __call__(self, request, data):
246			include_fields = get_include_fields(request)
247
248			if not include_fields or "MethodInstruments" in include_fields:
249			data["MethodInstruments"] = {}
250			for method in self.context.getAvailableMethods():
251			for instrument in method.getInstruments():
252			if method.UID() not in data["MethodInstruments"]:
253			data["MethodInstruments"][method.UID()] = []
254			data["MethodInstruments"][method.UID()].append(
255			load_field_values(instrument, include_fields=[]))
256

senaite / senaite.core

Push — master ( 804f45...1fa2c0 )

JSONReadExtender.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

JSONReadExtender.init() A