kalenislims/lims_quality_control/lims.py

# This file is part of lims_quality_control module for Tryton.
# The COPYRIGHT file at the top level of this repository contains
# the full copyright notices and license terms.
from datetime import datetime
from trytond.model import fields
from trytond.pyson import Eval, Equal
from trytond.transaction import Transaction
from trytond.pool import Pool, PoolMeta
from trytond.exceptions import UserError
from trytond.i18n import gettext
from trytond.modules.lims_interface.interface import FUNCTIONS
from .function import custom_functions

__all__ = ['Configuration', 'Method', 'Analysis', 'Typification',
    'NotebookLine', 'EntryDetailAnalysis', 'ResultReport',
    'NotebookLoadResultsManualLine', 'NotebookLoadResultsManual']

_PROOF_TYPES = [
    ('qualitative', 'Qualitative'),
    ('quantitative', 'Quantitative')
    ]
FUNCTIONS.update(custom_functions)


class Configuration(metaclass=PoolMeta):
    __name__ = 'lims.configuration'

    qc_fraction_type = fields.Many2One('lims.fraction.type',
        'QC fraction type')


class Method(metaclass=PoolMeta):
    __name__ = 'lims.lab.method'
    specification = fields.Text('Specification')


class Analysis(metaclass=PoolMeta):
    __name__ = 'lims.analysis'
    quality_type = fields.Selection(_PROOF_TYPES, 'Quality Type',
        required=True)
    quality_uom = fields.Many2One('product.uom', 'Quality UoM',
        domain=[('category.lims_only_available', '=', True)],
        states={
            'invisible': ~Equal(Eval('quality_type'), 'quantitative'),
            'required': Equal(Eval('quality_type'), 'quantitative'),
            }, depends=['quality_type'])
    quality_possible_values = fields.One2Many('lims.quality.qualitative.value',
        'analysis', 'Possible Values',
        states={
            'invisible': ~Equal(Eval('quality_type'), 'qualitative'),
            'required': Equal(Eval('quality_type'), 'qualitative'),
            }, depends=['quality_type'])

    @staticmethod
    def default_quality_type():
        return 'quantitative'


class Typification(metaclass=PoolMeta):
    __name__ = 'lims.typification'
    _history = True

    specification = fields.Text('Specification')
    quality = fields.Boolean('Quality')
    quality_template = fields.Many2One('lims.quality.template',
        'Quality Template')
    quality_type = fields.Function(fields.Selection(_PROOF_TYPES,
        'Quality Type', states={'invisible': True}),
        'on_change_with_quality_type')
    valid_value = fields.Many2One('lims.quality.qualitative.value',
        'Valid Value',
        states={
            'invisible': ~Equal(Eval('quality_type'), 'qualitative'),
            'required': Equal(Eval('quality_type'), 'qualitative'),
            },
        domain=[('id', 'in', Eval('valid_value_domain'))],
        depends=['valid_value_domain', 'quality_type'])
    valid_value_domain = fields.Function(fields.Many2Many(
        'lims.quality.qualitative.value',
        None, None, 'Valid Value domain',
        states={'invisible': True}), 'on_change_with_valid_value_domain')
    quality_test_report = fields.Boolean('Quality Test Report')
    quality_order = fields.Integer('Quality Order')
    quality_min = fields.Float('Min',
        digits=(16, Eval('limit_digits', 2)), depends=['limit_digits'])
    quality_max = fields.Float('Max',
        digits=(16, Eval('limit_digits', 2)), depends=['limit_digits'])

    @classmethod
    def __setup__(cls):
        super(Typification, cls).__setup__()
        cls._sql_constraints = []

    @classmethod
    def __register__(cls, module_name):
        super(Typification, cls).__register__(module_name)
        table = cls.__table_handler__(module_name)

        table.drop_constraint('product_matrix_analysis_method_uniq')

    @fields.depends('analysis')
    def on_change_with_valid_value_domain(self, name=None):
        values = []
        if self.analysis and self.analysis.quality_possible_values:
            for value in self.analysis.quality_possible_values:
                values.append(value.id)
        return values

    @fields.depends('analysis')
    def on_change_with_quality_type(self, name=None):
        if self.analysis:
            return self.analysis.quality_type

    @fields.depends('analysis')
    def on_change_analysis(self):
        if self.analysis:
            if self.analysis.quality_type == 'quantitative':
                self.start_uom = self.analysis.quality_uom
                self.end_uom = self.analysis.quality_uom

    @fields.depends('method')
    def on_change_method(self):
        if self.method:
            self.specification = self.method.specification

    def check_default(self):
        if self.quality:
            return
        if self.by_default:
            typifications = self.search([
                ('product_type', '=', self.product_type.id),
                ('matrix', '=', self.matrix.id),
                ('analysis', '=', self.analysis.id),
                ('valid', '=', True),
                ('by_default', '=', True),
                ('id', '!=', self.id),
                ])
            if typifications:
                raise UserError(gettext('lims.msg_default_typification'))
        else:
            if self.valid:
                typifications = self.search([
                    ('product_type', '=', self.product_type.id),
                    ('matrix', '=', self.matrix.id),
                    ('analysis', '=', self.analysis.id),
                    ('valid', '=', True),
                    ('id', '!=', self.id),
                    ])
                if not typifications:
                    raise UserError(
                        gettext('lims.msg_not_default_typification'))

    @classmethod
    def create(cls, vlist):
        Template = Pool().get('lims.quality.template')

        vlist = [x.copy() for x in vlist]
        for values in vlist:
            if values.get('quality_template'):
                template, = Template.browse([values.get('quality_template')])
                values['product_type'] = template.product.product_type.id
                values['matrix'] = template.product.matrix.id
        return super(Typification, cls).create(vlist)


class NotebookLine(metaclass=PoolMeta):
    __name__ = 'lims.notebook.line'
    typification = fields.Many2One('lims.typification', 'Typification')
    quality_test = fields.Many2One('lims.quality.test', 'Quality Test',
        select=True)
    test_value = fields.Many2One('lims.quality.qualitative.value',
        'Test Value',
        states={
            'readonly': True,
            })
    qualitative_value = fields.Many2One('lims.quality.qualitative.value',
        'Qualitative Value',
        domain=[
            ('analysis', '=', Eval('analysis')),
            ], depends=['analysis'])
    success = fields.Function(fields.Boolean('Success',
        depends=['success_icon']), 'get_success')
    success_icon = fields.Function(fields.Char('Success Icon',
        depends=['success']), 'get_success_icon')
    quality_min = fields.Float('Min',
        digits=(16, Eval('decimals', 2)), depends=['decimals'])
    quality_max = fields.Float('Max',
        digits=(16, Eval('decimals', 2)), depends=['decimals'])
    quality_test_report = fields.Boolean('Quality Test Report')

    @staticmethod
    def default_quality_test_report():
        return True

    @classmethod
    def get_success(self, lines, name):
        res = {}
        for line in lines:
            res[line.id] = False
            if line.analysis.quality_type == 'quantitative':
                if line.result:
                    value = float(line.result)
                    quality_min = line.quality_min
                    if not isinstance(quality_min, (int, float)):
                        quality_min = float('-inf')
                    quality_max = line.quality_max
                    if not isinstance(quality_max, (int, float)):
                        quality_max = float('inf')
                    if (value >= quality_min and value <= quality_max):
                        res[line.id] = True
            else:
                if line.qualitative_value == line.test_value:
                    res[line.id] = True
        return res

    def get_success_icon(self, name):
        if not self.accepted:
            return 'lims-white'
        if self.success:
            return 'lims-green'
        return 'lims-red'

    @fields.depends('qualitative_value')
    def on_change_qualitative_value(self):
        if self.qualitative_value:
            self.literal_result = self.qualitative_value.name

    @classmethod
    def fields_view_get(cls, view_id=None, view_type='form'):
        pool = Pool()
        User = pool.get('res.user')
        Config = pool.get('lims.configuration')
        UiView = pool.get('ir.ui.view')

        result = super(NotebookLine, cls).fields_view_get(view_id=view_id,
            view_type=view_type)

        # All Notebook Lines view
        if view_id and UiView(view_id).name == 'notebook_line_all_list':
            return result

        notebook_view = User(Transaction().user).notebook_view
        if not notebook_view:
            notebook_view = Config(1).default_notebook_view
            if not notebook_view:
                return result

        if view_type == 'tree':
            xml = '<?xml version="1.0"?>\n' \
                '<tree editable="bottom">\n'
            fields = set()
            for column in notebook_view.columns:
                fields.add(column.field.name)
                attrs = []
                if column.field.name == 'analysis':
                    attrs.append('icon="icon"')
                if column.field.name == 'success':
                    attrs.append('icon="success_icon"')
                if column.field.name in ('acceptance_date', 'annulment_date'):
                    attrs.append('widget="date"')
                xml += ('<field name="%s" %s/>\n'
                    % (column.field.name, ' '.join(attrs)))
                for depend in getattr(cls, column.field.name).depends:
                    fields.add(depend)
            for field in ('report_date', 'result', 'converted_result',
                    'result_modifier', 'converted_result_modifier',
                    'literal_result', 'backup', 'verification', 'uncertainty',
                    'accepted', 'acceptance_date', 'end_date', 'report',
                    'annulled', 'annulment_date', 'icon'):
                fields.add(field)
            xml += '</tree>'
            result['arch'] = xml
            result['fields'] = cls.fields_get(fields_names=list(fields))
        return result


class EntryDetailAnalysis(metaclass=PoolMeta):
    __name__ = 'lims.entry.detail.analysis'

    @classmethod
    def create_notebook_lines(cls, details, fraction):
        cursor = Transaction().connection.cursor()
        pool = Pool()
        Typification = pool.get('lims.typification')
        Method = pool.get('lims.lab.method')
        WaitingTime = pool.get('lims.lab.method.results_waiting')
        AnalysisLaboratory = pool.get('lims.analysis-laboratory')
        ProductType = pool.get('lims.product.type')
        Notebook = pool.get('lims.notebook')
        Company = pool.get('company.company')

        def _str_value(val=None):
            return str(val) if val is not None else None

        lines_create = []

        template_id = None
        if Transaction().context.get('template'):
            template_id = Transaction().context.get('template')
        for detail in details:
            query = 'SELECT default_repetitions, ' \
                    'initial_concentration, final_concentration, start_uom, ' \
                    'end_uom, detection_limit, quantification_limit, ' \
                    'lower_limit, upper_limit, calc_decimals, report, id ' \
                'FROM "' + Typification._table + '" ' \
                'WHERE product_type = %s ' \
                    'AND matrix = %s ' \
                    'AND analysis = %s ' \
                    'AND method = %s ' \
                    'AND valid'

            if template_id:
                query += ' AND quality_template = %s'
                cursor.execute(query,
                    (fraction.product_type.id, fraction.matrix.id,
                        detail.analysis.id, detail.method.id, template_id))
            else:
                cursor.execute(query,
                    (fraction.product_type.id, fraction.matrix.id,
                        detail.analysis.id, detail.method.id))
            typifications = cursor.fetchall()
            typification = (typifications[0] if len(typifications) == 1
                else None)
            if typification:
                repetitions = typification[0]
                initial_concentration = _str_value(typification[1])
                final_concentration = _str_value(typification[2])
                initial_unit = typification[3] or None
                final_unit = typification[4] or None
                detection_limit = _str_value(typification[5])
                quantification_limit = _str_value(typification[6])
                lower_limit = _str_value(typification[7])
                upper_limit = _str_value(typification[8])
                decimals = typification[9]
                report = typification[10]
            else:
                repetitions = 0
                initial_concentration = None
                final_concentration = None
                initial_unit = None
                final_unit = None
                detection_limit = None
                quantification_limit = None
                lower_limit = None
                upper_limit = None
                decimals = 2
                report = False

            results_estimated_waiting = None
            cursor.execute('SELECT results_estimated_waiting '
                'FROM "' + WaitingTime._table + '" '
                'WHERE method = %s '
                    'AND party = %s',
                (detail.method.id, detail.party.id))
            res = cursor.fetchone()
            if res:
                results_estimated_waiting = res[0]
            else:
                cursor.execute('SELECT results_estimated_waiting '
                    'FROM "' + Method._table + '" '
                    'WHERE id = %s', (detail.method.id,))
                res = cursor.fetchone()
                if res:
                    results_estimated_waiting = res[0]

            department = None
            cursor.execute('SELECT department '
                'FROM "' + AnalysisLaboratory._table + '" '
                'WHERE analysis = %s '
                    'AND laboratory = %s',
                    (detail.analysis.id, detail.laboratory.id))
            res = cursor.fetchone()
            if res and res[0]:
                department = res[0]
            else:
                cursor.execute('SELECT department '
                    'FROM "' + ProductType._table + '" '
                    'WHERE id = %s', (fraction.product_type.id,))
                res = cursor.fetchone()
                if res and res[0]:
                    department = res[0]

            for i in range(0, repetitions + 1):
                notebook_line = {
                    'analysis_detail': detail.id,
                    'service': detail.service.id,
                    'analysis': detail.analysis.id,
                    'analysis_origin': detail.analysis_origin,
                    'repetition': i,
                    'laboratory': detail.laboratory.id,
                    'method': detail.method.id,
                    'device': detail.device.id if detail.device else None,
                    'initial_concentration': initial_concentration,
                    'final_concentration': final_concentration,
                    'initial_unit': initial_unit,
                    'final_unit': final_unit,
                    'detection_limit': detection_limit,
                    'quantification_limit': quantification_limit,
                    'decimals': decimals,
                    'report': report,
                    'results_estimated_waiting': results_estimated_waiting,
                    'department': department,
                    }
                if template_id:
                    quality_typification = Typification(typification[11])
                    notebook_line['typification'] = quality_typification.id
                    notebook_line['test_value'] = \
                        quality_typification.valid_value.id \
                        if quality_typification.valid_value else None
                    notebook_line['quality_test'] = Transaction().context.get(
                        'test')
                    notebook_line['quality_min'] = \
                        quality_typification.quality_min
                    notebook_line['quality_max'] = \
                        quality_typification.quality_max
                    notebook_line['quality_test_report'] = \
                        quality_typification.quality_test_report
                lines_create.append(notebook_line)

        if not lines_create:
            companies = Company.search([])
            if fraction.party.id not in [c.party.id for c in companies]:
                raise UserError(gettext(
                    'lims.msg_not_services', fraction=fraction.rec_name))

        with Transaction().set_user(0):
            notebook = Notebook.search([
                ('fraction', '=', fraction.id),
                ])
            Notebook.write(notebook, {
                'lines': [('create', lines_create)],
                })


class ResultReport(metaclass=PoolMeta):
    __name__ = 'lims.result_report'


    @classmethod
    def get_reference(cls, range_type, notebook_line, language,
            report_section):
        res = super(ResultReport, cls).get_reference(
            range_type, notebook_line, language, report_section)
        if res:
            return res

        res = ''
        if notebook_line.quality_min:
            with Transaction().set_context(language=language):
                resf = float(notebook_line.quality_min)
                resd = abs(resf) - abs(int(resf))
                if resd > 0:
                    res1 = str(round(notebook_line.quality_min, 2))
                else:
                    res1 = str(int(notebook_line.quality_min))
                res = gettext('lims.msg_caa_min', min=res1)

        if notebook_line.quality_max:
            if res:
                res += ' - '
            with Transaction().set_context(language=language):
                resf = float(notebook_line.quality_max)
                resd = abs(resf) - abs(int(resf))
                if resd > 0:
                    res1 = str(round(notebook_line.quality_max, 2))
                else:
                    res1 = str(int(notebook_line.quality_max))

                res += gettext('lims.msg_caa_max', max=res1)
        return res


class NotebookLoadResultsManualLine(metaclass=PoolMeta):
    __name__ = 'lims.notebook.load_results_manual.line'
    qualitative_value = fields.Many2One('lims.quality.qualitative.value',
        'Qualitative Value',
        domain=[
            ('analysis', '=', Eval('analysis')),
            ], depends=['analysis'])
    analysis = fields.Function(fields.Many2One('lims.analysis', 'Analysis'),
        'get_analysis')

    def get_analysis(self, name):
        return self.line.analysis.id if self.line else None

    @fields.depends('result', 'literal_result', 'result_modifier', 'end_date',
        'qualitative_value')
    def on_change_with_end_date(self):
        pool = Pool()
        Date = pool.get('ir.date')
        if self.end_date:
            return self.end_date
        if (self.result or self.literal_result or self.qualitative_value or
                self.result_modifier not in ('eq', 'low')):
            return Date.today()
        return None

    @fields.depends('qualitative_value')
    def on_change_qualitative_value(self):
        if self.qualitative_value:
            self.literal_result = self.qualitative_value.name


class NotebookLoadResultsManual(metaclass=PoolMeta):
    __name__ = 'lims.notebook.load_results_manual'

    def transition_confirm_(self):
        pool = Pool()
        NotebookLoadResultsManualLine = pool.get(
            'lims.notebook.load_results_manual.line')
        NotebookLine = pool.get('lims.notebook.line')
        LabProfessionalMethod = pool.get('lims.lab.professional.method')
        LabProfessionalMethodRequalification = pool.get(
            'lims.lab.professional.method.requalification')
        Date = pool.get('ir.date')

        # Write Results to Notebook lines
        actions = NotebookLoadResultsManualLine.search([
            ('session_id', '=', self._session_id),
            ])

        for data in actions:
            notebook_line = NotebookLine(data.line.id)
            if not notebook_line:
                continue
            notebook_line_write = {
                'result': data.result,
                'qualitative_value': data.qualitative_value,
                'result_modifier': data.result_modifier,
                'end_date': data.end_date,
                'chromatogram': data.chromatogram,
                'initial_unit': (data.initial_unit.id if
                    data.initial_unit else None),
                'comments': data.comments,
                'literal_result': data.literal_result,
                'converted_result': None,
                'converted_result_modifier': 'eq',
                'backup': None,
                'verification': None,
                'uncertainty': None,
                }
            if data.result_modifier == 'na':
                notebook_line_write['annulled'] = True
                notebook_line_write['annulment_date'] = datetime.now()
                notebook_line_write['report'] = False
            professionals = [{'professional': self.result.professional.id}]
            notebook_line_write['professionals'] = (
                [('delete', [p.id for p in notebook_line.professionals])] +
                [('create', professionals)])
            NotebookLine.write([notebook_line], notebook_line_write)

        # Write Supervisors to Notebook lines
        supervisor_lines = {}
        if hasattr(self.sit2, 'supervisor'):
            supervisor_lines[self.sit2.supervisor.id] = [
                l.id for l in self.sit2.lines]
        for prof_id, lines in supervisor_lines.items():
            notebook_lines = NotebookLine.search([
                ('id', 'in', lines),
                ])
            if notebook_lines:
                professionals = [{'professional': prof_id}]
                notebook_line_write = {
                    'professionals': [('create', professionals)],
                    }
                NotebookLine.write(notebook_lines, notebook_line_write)

        # Write the execution of method
        all_prof = {}
        key = (self.result.professional.id, self.result.method.id)
        all_prof[key] = []
        if hasattr(self.sit2, 'supervisor'):
            for detail in self.sit2.lines:
                key = (self.sit2.supervisor.id, detail.method.id)
                if key not in all_prof:
                    all_prof[key] = []
                key = (self.result.professional.id, detail.method.id)
                if self.sit2.supervisor.id not in all_prof[key]:
                    all_prof[key].append(self.sit2.supervisor.id)

        today = Date.today()
        for key, sup in all_prof.items():
            professional_method, = LabProfessionalMethod.search([
                ('professional', '=', key[0]),
                ('method', '=', key[1]),
                ('type', '=', 'analytical'),
                ])
            if professional_method.state == 'training':
                history = LabProfessionalMethodRequalification.search([
                    ('professional_method', '=', professional_method.id),
                    ('type', '=', 'training'),
                    ])
                if history:
                    prev_supervisors = [s.supervisor.id for s in
                        history[0].supervisors]
                    supervisors = [{'supervisor': s} for s in sup
                        if s not in prev_supervisors]
                    LabProfessionalMethodRequalification.write(history, {
                        'last_execution_date': today,
                        'supervisors': [('create', supervisors)],
                        })
                else:
                    supervisors = [{'supervisor': s} for s in sup]
                    to_create = [{
                        'professional_method': professional_method.id,
                        'type': 'training',
                        'date': today,
                        'last_execution_date': today,
                        'supervisors': [('create', supervisors)],
                        }]
                    LabProfessionalMethodRequalification.create(to_create)

            elif professional_method.state == 'qualified':
                history = LabProfessionalMethodRequalification.search([
                    ('professional_method', '=', professional_method.id),
                    ('type', '=', 'qualification'),
                    ])
                if history:
                    LabProfessionalMethodRequalification.write(history, {
                        'last_execution_date': today,
                        })
                else:
                    to_create = [{
                        'professional_method': professional_method.id,
                        'type': 'qualification',
                        'date': today,
                        'last_execution_date': today,
                        }]
                    LabProfessionalMethodRequalification.create(to_create)

            else:
                history = LabProfessionalMethodRequalification.search([
                    ('professional_method', '=', professional_method.id),
                    ('type', '=', 'requalification'),
                    ])
                if history:
                    LabProfessionalMethodRequalification.write(history, {
                        'last_execution_date': today,
                        })
                else:
                    to_create = [{
                        'professional_method': professional_method.id,
                        'type': 'requalification',
                        'date': today,
                        'last_execution_date': today,
                        }]
                    LabProfessionalMethodRequalification.create(to_create)

        return 'end'