Odoo-Modules/fusion_plating/fusion_plating_certificates/models/fp_certificate.py

# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
# Part of the Fusion Plating product family.

from odoo import api, fields, models, _
from odoo.exceptions import UserError


class FpCertificate(models.Model):
    """Unified certificate registry.

    Logs every quality document issued to customers: CoC, thickness
    reports, mill test reports, Nadcap certs, and customer-specific
    formats. Auto-created when reports are generated.
    """
    _name = 'fp.certificate'
    _description = 'Fusion Plating — Certificate'
    _inherit = ['mail.thread', 'mail.activity.mixin']
    _order = 'issue_date desc, id desc'

    name = fields.Char(string='Reference', readonly=True, copy=False, default='New')
    certificate_type = fields.Selection(
        [
            ('coc', 'Certificate of Conformance'),
            ('thickness_report', 'Thickness Report'),
            ('mill_test', 'Mill Test Report'),
            ('nadcap_cert', 'Nadcap Certificate'),
            ('customer_specific', 'Customer-Specific'),
        ],
        string='Type', required=True, default='coc', tracking=True,
    )
    partner_id = fields.Many2one(
        'res.partner', string='Customer', required=True, tracking=True,
        domain="[('customer_rank', '>', 0)]",
    )
    sale_order_id = fields.Many2one('sale.order', string='Sale Order')
    production_id = fields.Many2one('mrp.production', string='Manufacturing Order')
    portal_job_id = fields.Many2one('fusion.plating.portal.job', string='Portal Job')
    part_number = fields.Char(string='Part Number', help='Denormalized for fast search.')
    process_description = fields.Char(
        string='Process', help='e.g. "ELECTROLESS NICKEL PLATING PER AMS 2404"',
    )
    spec_reference = fields.Char(string='Spec Reference')
    po_number = fields.Char(string='Customer PO #')
    entech_wo_number = fields.Char(string='Entech WO #')
    quantity_shipped = fields.Integer(string='Qty Shipped')
    nc_quantity = fields.Integer(
        string='NC Qty',
        help='Non-conforming quantity — parts that failed inspection / rework.',
    )
    customer_job_no = fields.Char(
        string='Customer Job No.',
        help="Customer's internal job / traveler reference.",
    )
    contact_partner_id = fields.Many2one(
        'res.partner', string='Customer Contact',
        domain="[('parent_id', '=', partner_id)]",
        help="Specific contact person at the customer for this certificate. "
             'Their name, email, and phone are printed on the CoC.',
    )
    issued_by_id = fields.Many2one(
        'res.users', string='Issued By', default=lambda self: self.env.user,
    )
    certified_by_id = fields.Many2one(
        'res.users', string='Certified By', help='Signing authority (e.g. Quality Manager).',
    )
    issue_date = fields.Date(string='Issue Date', default=fields.Date.today, tracking=True)
    attachment_id = fields.Many2one('ir.attachment', string='Certificate PDF')
    thickness_reading_ids = fields.One2many(
        'fp.thickness.reading', 'certificate_id', string='Thickness Readings',
    )

    # ---- Material traceability (T2.3) ----
    batch_ids = fields.Many2many(
        'fusion.plating.batch', compute='_compute_batch_ids',
        string='Batches', help='All batches used for this MO.',
    )
    batch_count = fields.Integer(
        string='Batches', compute='_compute_batch_ids',
    )
    bath_ids = fields.Many2many(
        'fusion.plating.bath', compute='_compute_batch_ids',
        string='Baths Used',
    )

    @api.depends('production_id')
    def _compute_batch_ids(self):
        Batch = self.env.get('fusion.plating.batch')
        Bath = self.env['fusion.plating.bath']
        empty_batch = self.env['fusion.plating.batch']
        for rec in self:
            if Batch is not None and rec.production_id:
                batches = Batch.search([
                    ('production_id', '=', rec.production_id.id),
                ])
                rec.batch_ids = batches
                rec.batch_count = len(batches)
                rec.bath_ids = batches.mapped('bath_id')
            else:
                rec.batch_ids = empty_batch
                rec.batch_count = 0
                rec.bath_ids = Bath
    state = fields.Selection(
        [('draft', 'Draft'), ('issued', 'Issued'), ('voided', 'Voided')],
        string='Status', default='draft', tracking=True, required=True,
    )
    void_reason = fields.Text(string='Void Reason')
    notes = fields.Html(string='Notes')

    # ----- Computed stats from readings -------------------------------------
    reading_count = fields.Integer(
        string='Readings', compute='_compute_reading_stats', store=True,
    )
    mean_nip_mils = fields.Float(
        string='Mean NiP (mils)', compute='_compute_reading_stats',
        store=True, digits=(10, 4),
    )

    # ---- SPC (T2.1) ----
    spec_min_mils = fields.Float(
        string='Spec Min (mils)', digits=(10, 4),
        help='Lower specification limit pulled from the coating config. '
             'Override per certificate if needed.',
    )
    spec_max_mils = fields.Float(
        string='Spec Max (mils)', digits=(10, 4),
        help='Upper specification limit pulled from the coating config.',
    )
    std_dev_mils = fields.Float(
        string='Std Dev (mils)', compute='_compute_reading_stats',
        store=True, digits=(10, 4),
    )
    min_reading_mils = fields.Float(
        string='Min Reading', compute='_compute_reading_stats',
        store=True, digits=(10, 4),
    )
    max_reading_mils = fields.Float(
        string='Max Reading', compute='_compute_reading_stats',
        store=True, digits=(10, 4),
    )
    cpk = fields.Float(
        string='Cpk', compute='_compute_reading_stats', store=True, digits=(6, 3),
        help='Process capability index. <1.0 = incapable · 1.0-1.33 = marginal · '
             '≥1.33 = capable · ≥1.67 = excellent.',
    )
    cpk_status = fields.Selection(
        [('incapable', 'Incapable'),
         ('marginal', 'Marginal'),
         ('capable', 'Capable'),
         ('excellent', 'Excellent'),
         ('insufficient', 'Insufficient Data')],
        string='Cpk Status', compute='_compute_reading_stats', store=True,
    )
    trend_alert = fields.Selection(
        [('ok', 'OK'),
         ('warning', 'Trend Detected'),
         ('alert', 'Out of Control')],
        string='Trend Alert', compute='_compute_reading_stats', store=True,
        help='Western Electric rule 1 (any point beyond 3σ) or rule 4 '
             '(8 consecutive points on one side of centre).',
    )
    trend_explanation = fields.Char(
        string='Trend Note', compute='_compute_reading_stats', store=True,
    )

    @api.depends('thickness_reading_ids', 'thickness_reading_ids.nip_mils',
                 'spec_min_mils', 'spec_max_mils')
    def _compute_reading_stats(self):
        for rec in self:
            readings = rec.thickness_reading_ids
            rec.reading_count = len(readings)
            values = [r.nip_mils for r in readings if r.nip_mils]
            n = len(values)
            if n == 0:
                rec.mean_nip_mils = 0.0
                rec.std_dev_mils = 0.0
                rec.min_reading_mils = 0.0
                rec.max_reading_mils = 0.0
                rec.cpk = 0.0
                rec.cpk_status = 'insufficient'
                rec.trend_alert = 'ok'
                rec.trend_explanation = ''
                continue

            mean = sum(values) / n
            rec.mean_nip_mils = round(mean, 4)
            rec.min_reading_mils = round(min(values), 4)
            rec.max_reading_mils = round(max(values), 4)

            # Sample standard deviation (Bessel's correction)
            if n >= 2:
                sq_diff = sum((v - mean) ** 2 for v in values)
                sigma = (sq_diff / (n - 1)) ** 0.5
            else:
                sigma = 0.0
            rec.std_dev_mils = round(sigma, 4)

            # Cpk — requires spec limits + non-zero sigma
            usl = rec.spec_max_mils or 0.0
            lsl = rec.spec_min_mils or 0.0
            if n < 5 or sigma == 0 or (usl == 0 and lsl == 0):
                rec.cpk = 0.0
                rec.cpk_status = 'insufficient'
            else:
                if usl and lsl:
                    cpu = (usl - mean) / (3.0 * sigma)
                    cpl = (mean - lsl) / (3.0 * sigma)
                    cpk = min(cpu, cpl)
                elif usl:
                    cpk = (usl - mean) / (3.0 * sigma)
                else:
                    cpk = (mean - lsl) / (3.0 * sigma)
                rec.cpk = round(cpk, 3)
                if cpk < 1.0:
                    rec.cpk_status = 'incapable'
                elif cpk < 1.33:
                    rec.cpk_status = 'marginal'
                elif cpk < 1.67:
                    rec.cpk_status = 'capable'
                else:
                    rec.cpk_status = 'excellent'

            # Trend detection (Western Electric rules)
            # Rule 1: any point outside 3σ from mean
            # Rule 4: 8+ consecutive on one side of mean
            alert = 'ok'
            explanation = ''
            if sigma > 0:
                three_sigma = 3.0 * sigma
                for v in values:
                    if abs(v - mean) > three_sigma:
                        alert = 'alert'
                        explanation = 'Rule 1: reading beyond 3σ from mean'
                        break
            if alert == 'ok' and n >= 8:
                # Check last 8 readings for all-above or all-below
                last_eight = values[-8:]
                if all(v > mean for v in last_eight):
                    alert = 'warning'
                    explanation = 'Rule 4: 8 consecutive readings above mean'
                elif all(v < mean for v in last_eight):
                    alert = 'warning'
                    explanation = 'Rule 4: 8 consecutive readings below mean'
            rec.trend_alert = alert
            rec.trend_explanation = explanation

    # ----- Sequence + spec-limit auto-fill ---------------------------------
    @api.model_create_multi
    def create(self, vals_list):
        SaleOrder = self.env['sale.order']
        for vals in vals_list:
            if vals.get('name', 'New') == 'New':
                vals['name'] = self.env['ir.sequence'].next_by_code('fp.certificate') or 'New'
            # Pull thickness spec limits from coating config if not set
            already_set = vals.get('spec_min_mils') or vals.get('spec_max_mils')
            if not already_set and vals.get('sale_order_id'):
                so = SaleOrder.browse(vals['sale_order_id'])
                cfg = getattr(so, 'x_fc_coating_config_id', False)
                if cfg and cfg.thickness_uom == 'mils':
                    vals.setdefault('spec_min_mils', cfg.thickness_min or 0.0)
                    vals.setdefault('spec_max_mils', cfg.thickness_max or 0.0)
        return super().create(vals_list)

    # ----- State actions ----------------------------------------------------
    def action_issue(self):
        for rec in self:
            if rec.state != 'draft':
                raise UserError(_('Only draft certificates can be issued.'))
            # Spec reference is what the cert ATTESTS — without it the
            # cert is just a piece of paper. AS9100 / Nadcap require
            # naming the spec the work was performed to.
            if not rec.spec_reference:
                raise UserError(_(
                    'Cannot issue certificate "%(name)s" — no Spec '
                    'Reference set.\n\nFill the Spec Reference field '
                    '(e.g. "AMS 2404", "MIL-C-26074") so the cert '
                    'states which standard the work meets.'
                ) % {'name': rec.name or rec.display_name})
            rec.state = 'issued'
            rec.message_post(body=_('Certificate issued.'))

    def action_void(self):
        for rec in self:
            if rec.state != 'issued':
                raise UserError(_('Only issued certificates can be voided.'))
            if not rec.void_reason:
                raise UserError(_('Please enter a void reason before voiding.'))
            rec.state = 'voided'
            rec.message_post(body=_('Certificate voided. Reason: %s') % rec.void_reason)

    def action_view_traceability(self):
        """Show the batches (and their chemistry logs) that produced
        these parts — auditor's dream, customer's RMA friend."""
        self.ensure_one()
        return {
            'type': 'ir.actions.act_window',
            'name': _('Traceability — %s') % self.name,
            'res_model': 'fusion.plating.batch',
            'view_mode': 'list,form',
            'domain': [('id', 'in', self.batch_ids.ids)],
            'target': 'current',
        }

    def action_send_to_customer(self):
        """Open email composer with the certificate PDF attached."""
        self.ensure_one()
        template = self.env.ref('mail.email_compose_message_wizard_form', raise_if_not_found=False)
        ctx = {
            'default_model': 'fp.certificate',
            'default_res_ids': self.ids,
            'default_composition_mode': 'comment',
            'default_partner_ids': [self.partner_id.id] if self.partner_id else [],
        }
        if self.attachment_id:
            ctx['default_attachment_ids'] = [self.attachment_id.id]
        return {
            'type': 'ir.actions.act_window',
            'res_model': 'mail.compose.message',
            'view_mode': 'form',
            'target': 'new',
            'context': ctx,
        }