Files
Odoo-Modules/fusion_plating/fusion_plating_certificates/models/fp_certificate.py
gsinghpal 0d85063b5e feat(numbering): wire CoC/RCV/DLV/PU into parent-numbered mixin + rename counters
Per-model counter fields on sale.order renamed to x_fc_pn_*_count
to avoid collision with pre-existing compute fields of the same
short name in bridge_mrp / receiving / configurator (silent
compute-override was suppressing the storage). 4 child models
(fp.certificate, fp.receiving, fusion.plating.delivery,
fusion.plating.pickup.request) now derive names as PFX-<parent>
with -NN suffix from the 2nd onward.

fusion.plating.pickup.request gains a sale_order_id field
(optional) so pickups created against an SO get parent-derived
names, while standalone pickups (pre-SO) fall back to PU/YYYY/NNNN.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 13:30:37 -04:00

555 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
# Part of the Fusion Plating product family.
import logging
from odoo import api, fields, models, _
from odoo.exceptions import UserError
_logger = logging.getLogger(__name__)
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', 'fp.parent.numbered.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')
# Phase 6 (Sub 11) — production_id retired (MRP module gone).
# Certificates link via sale_order_id + portal_job_id natively.
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).',
)
# ===== Sub 12c — chronological CoC opt-in ===============================
body_style = fields.Selection(
[
('classic', 'Classic (recipe-order)'),
('chronological', 'Chronological (chain-of-custody)'),
],
string='CoC Body Style', default='classic',
help='Chronological walks fp.job.step.move records in time order '
'with measurement sub-tables per move, matching Steelhead\'s '
'CoC PDF layout. Classic uses the existing recipe-order body.',
)
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('sale_order_id')
def _compute_batch_ids(self):
# Phase 6 (Sub 11) — walks fp.job via SO instead of mrp.production.
Batch = self.env.get('fusion.plating.batch')
Bath = self.env['fusion.plating.bath']
Job = self.env.get('fp.job')
empty_batch = self.env['fusion.plating.batch']
for rec in self:
if Batch is None or Job is None or not rec.sale_order_id:
rec.batch_ids = empty_batch
rec.batch_count = 0
rec.bath_ids = Bath
continue
jobs = Job.search([('sale_order_id', '=', rec.sale_order_id.id)])
if not jobs:
rec.batch_ids = empty_batch
rec.batch_count = 0
rec.bath_ids = Bath
continue
if 'x_fc_job_id' in Batch._fields:
batches = Batch.search([('x_fc_job_id', 'in', jobs.ids)])
else:
batches = empty_batch
rec.batch_ids = batches
rec.batch_count = len(batches)
rec.bath_ids = batches.mapped('bath_id')
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
# ----- Parent-numbered mixin hooks -------------------------------------
def _fp_parent_sale_order(self):
return self.sale_order_id
def _fp_name_prefix(self):
return 'CoC'
def _fp_parent_counter_field(self):
return 'x_fc_pn_cert_count'
# ----- Create: parent-derived name (fallback to legacy sequence) -------
@api.model_create_multi
def create(self, vals_list):
SaleOrder = self.env['sale.order']
for vals in vals_list:
# Spec-limit auto-fill (existing behaviour, preserved).
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)
# Defer naming: let the record exist so the mixin can write
# name via raw SQL, then fall back to the legacy sequence if
# no parent SO is reachable.
if not vals.get('name'):
vals['name'] = 'New'
records = super().create(vals_list)
for rec in records:
if rec.name and rec.name != 'New':
continue
if not rec._fp_assign_parent_name():
seq = self.env['ir.sequence'].next_by_code('fp.certificate') or 'New'
self.env.cr.execute(
"UPDATE fp_certificate SET name = %s WHERE id = %s",
(seq, rec.id),
)
rec.invalidate_recordset(['name'])
return records
# ----- 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})
# Aerospace / Nadcap customers: actual thickness readings
# must be on file BEFORE the cert is issued. The flag lives
# on the partner so commercial customers aren't blocked.
if (rec.partner_id
and 'x_fc_strict_thickness_required' in rec.partner_id._fields
and rec.partner_id.x_fc_strict_thickness_required
and rec.certificate_type == 'coc'):
if not rec.thickness_reading_ids:
raise UserError(_(
'Cannot issue CoC "%(name)s" — customer "%(cust)s" '
'requires actual thickness readings on every CoC '
'(Nadcap / aerospace).\n\nLog Fischerscope readings '
'against the job for SO %(so)s via the Tablet Station '
'before issuing.'
) % {
'name': rec.name or rec.display_name,
'cust': rec.partner_id.name,
'so': rec.sale_order_id.name if rec.sale_order_id else '?',
})
rec.state = 'issued'
# Generate the CoC PDF and attach it so action_send_to_customer
# has something to email. Without this the workflow goes:
# Issue → Send → opens composer with no attachment → operator
# closes confused. Best-effort: if the report renders, attach;
# if it fails, log + continue (cert is still issued).
try:
rec._fp_render_and_attach_pdf()
except Exception as e:
_logger.warning(
'Cert %s: PDF render failed: %s', rec.name, e,
)
rec.message_post(body=_('Certificate issued.'))
def _fp_render_and_attach_pdf(self):
"""Render the CoC PDF via the bound report action, OPTIONALLY
merge the Fischerscope thickness report PDF (uploaded by the
QC tablet operator) as page 2, and attach the result.
Without the merge, a customer who specs "CoC must include the
XRF report" gets two separate PDFs to chase down. AS9100 wants
the supporting evidence inline with the cert.
Tries the EN-language CoC report first, falls back to the
generic action_report_coc. Idempotent — skips if attachment_id
is already set. PDF merge is best-effort: corrupt Fischerscope
upload or missing pypdf falls back to CoC-only with a warning.
"""
import base64
import io
self.ensure_one()
if self.attachment_id:
return self.attachment_id
report = (
self.env.ref(
'fusion_plating_reports.action_report_coc_en',
raise_if_not_found=False,
)
or self.env.ref(
'fusion_plating_reports.action_report_coc',
raise_if_not_found=False,
)
)
if not report:
_logger.warning(
'Cert %s: no CoC report action found, cannot render PDF',
self.name,
)
return False
coc_pdf_bytes, _content_type = report._render_qweb_pdf(
report.report_name, res_ids=self.ids,
)
# Try to append the Fischerscope thickness-report PDF as page 2.
merged_bytes = self._fp_merge_thickness_into_pdf(coc_pdf_bytes)
final_pdf = merged_bytes or coc_pdf_bytes
att = self.env['ir.attachment'].sudo().create({
'name': '%s.pdf' % (self.name or 'certificate'),
'type': 'binary',
'datas': base64.b64encode(final_pdf),
'mimetype': 'application/pdf',
'res_model': self._name,
'res_id': self.id,
})
self.attachment_id = att.id
return att
def _fp_merge_thickness_into_pdf(self, coc_pdf_bytes):
"""Look up the linked QC check, find its thickness_report_pdf_id
(Fischerscope / XDAL 600 XRF export), and return a merged PDF
with the CoC first + Fischerscope appended as page 2+.
Returns None when:
- cert isn't a CoC, or
- no fp.job linked, or
- no fp.quality.check on the job has a PDF uploaded, or
- pypdf / PyPDF2 not installed, or
- either PDF fails to parse.
Caller falls back to CoC-only when None is returned.
"""
import io
import base64 as _b64
self.ensure_one()
if self.certificate_type != 'coc':
return None
# Find the linked job. fp.certificate has either x_fc_job_id
# (preferred — added by fusion_plating_jobs) or job_id (older).
job = False
if 'x_fc_job_id' in self._fields:
job = self.x_fc_job_id
if not job and 'job_id' in self._fields:
job = self.job_id
if not job:
return None
# Find a passed QC on this job with an uploaded Fischerscope PDF.
# Prefer state=passed; fall through to any with a PDF.
QC = self.env.get('fusion.plating.quality.check')
if QC is None:
return None
qc = QC.sudo().search([
('job_id', '=', job.id),
('state', '=', 'passed'),
('thickness_report_pdf_id', '!=', False),
], order='completed_at desc', limit=1)
if not qc:
qc = QC.sudo().search([
('job_id', '=', job.id),
('thickness_report_pdf_id', '!=', False),
], order='create_date desc', limit=1)
if not qc or not qc.thickness_report_pdf_id:
return None
fischer_bytes = _b64.b64decode(
qc.thickness_report_pdf_id.datas or b''
)
if not fischer_bytes:
return None
# Merge — pypdf is the modern name; PyPDF2 still works on older
# Odoo bundles. Either is fine.
try:
from pypdf import PdfWriter
writer_cls = PdfWriter
use_append = True
except ImportError:
try:
from PyPDF2 import PdfMerger
writer_cls = PdfMerger
use_append = False
except ImportError:
_logger.warning(
'Cert %s: neither pypdf nor PyPDF2 installed, '
'cannot append Fischerscope PDF to CoC.',
self.name,
)
return None
try:
if use_append:
# pypdf 3.x — PdfWriter.append() handles bytes/streams
writer = writer_cls()
writer.append(io.BytesIO(coc_pdf_bytes))
writer.append(io.BytesIO(fischer_bytes))
out = io.BytesIO()
writer.write(out)
merged = out.getvalue()
else:
# PyPDF2 — PdfMerger.append + write
merger = writer_cls()
merger.append(io.BytesIO(coc_pdf_bytes))
merger.append(io.BytesIO(fischer_bytes))
out = io.BytesIO()
merger.write(out)
merger.close()
merged = out.getvalue()
except Exception:
_logger.exception(
'PDF merge failed for cert %s — Fischerscope PDF may '
'be corrupt / encrypted / malformed. Falling back to '
'CoC-only.', self.name,
)
return None
self.message_post(body=_(
'Fischerscope thickness report from QC %s appended to CoC PDF.'
) % qc.name)
return merged
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,
}