# -*- coding: utf-8 -*- # Copyright 2026 Nexa Systems Inc. # License OPL-1 (Odoo Proprietary License v1.0) # Part of the Fusion Plating product family. """Time-series of sensor readings. Every POST to /fp/iot/ingest (or every broadcast from the iot proxy) lands as a new row here. Kept intentionally append-only — we never update or delete readings, which makes this the compliance log for bath-temperature history. Auto-creates a fusion.plating.quality.hold when a reading falls outside the sensor's alert range AND the sensor has `alert_on_out_of_spec` enabled. The hold is created once per excursion (we don't spam a new hold for every reading during a sustained excursion) — tracked via the sensor's most-recent `last_reading_in_spec` flag. """ import logging from odoo import api, fields, models _logger = logging.getLogger(__name__) class FpTankReading(models.Model): _name = 'fp.tank.reading' _description = 'Fusion Plating — Tank Sensor Reading' _order = 'reading_at desc, id desc' _rec_name = 'display_name' sensor_id = fields.Many2one( 'fp.tank.sensor', string='Sensor', required=True, ondelete='cascade', index=True, ) # Denormalised for fast list views + kpi queries — auto-filled at # create time from sensor_id. Indexed for historical trending. tank_id = fields.Many2one( 'fusion.plating.tank', string='Tank', related='sensor_id.tank_id', store=True, index=True, ) bath_id = fields.Many2one( 'fusion.plating.bath', string='Bath', related='sensor_id.bath_id', store=True, ) parameter_id = fields.Many2one( 'fusion.plating.bath.parameter', string='Parameter', related='sensor_id.parameter_id', store=True, ) reading_at = fields.Datetime( string='Read At', required=True, default=fields.Datetime.now, index=True, ) value = fields.Float( string='Value (raw)', required=True, digits=(12, 4), help='Stored reading in the sensor\'s canonical unit (for ' 'temperature sensors this is always °C — the DS18B20 and ' 'every other temperature chip reports in Celsius natively; ' 'keeping storage canonical lets us switch display units ' 'per-company without re-migrating history).', ) unit = fields.Char( string='Unit (raw)', related='parameter_id.uom_display', store=True, ) # ------------------------------------------------------------------ # Display-aware fields — converted to the company's preferred unit # (res.company.x_fc_default_temp_uom = 'F' or 'C'). Only the list # and form views should show these; internal spec comparisons use # `value` so thresholds stay consistent across regions. # ------------------------------------------------------------------ display_value = fields.Float( string='Value', compute='_compute_display', digits=(12, 2), help='Reading in the company-preferred unit (Settings → Plating → ' 'Temperature Unit).', ) display_unit = fields.Char( string='Unit', compute='_compute_display', ) @api.depends('value', 'parameter_id', 'parameter_id.parameter_type', 'parameter_id.uom') def _compute_display(self): # Read once per compute call — env.company rarely changes mid-batch. pref = self.env.company.x_fc_default_temp_uom or 'C' for r in self: ptype = (r.parameter_id.parameter_type or '').lower() if ptype == 'temperature' and pref == 'F': r.display_value = r.value * 9.0 / 5.0 + 32.0 r.display_unit = '°F' else: r.display_value = r.value r.display_unit = r.parameter_id.uom_display or '' # ------------------------------------------------------------------ # Deviation from setpoint — signed Δ from the sensor's effective target # in the company-preferred unit. Zero if no setpoint defined. # ------------------------------------------------------------------ deviation_from_target = fields.Float( string='Δ from Setpoint', compute='_compute_deviation', digits=(12, 2), help='Signed difference between this reading and the sensor\'s ' 'setpoint, in the company-preferred unit. Positive = above ' 'setpoint, negative = below. Zero if no setpoint is defined.', ) deviation_band = fields.Selection( [ ('none', 'No setpoint'), ('on', 'On target (±1°)'), ('near', 'Near target (±3°)'), ('far', 'Far from target (>3°)'), ('out', 'Out of spec'), ], string='Band', compute='_compute_deviation', help='Coarse band for quick visual scanning in lists.', ) @api.depends('value', 'display_value', 'in_spec', 'sensor_id', 'sensor_id.target_value_override', 'sensor_id.parameter_id.target_value', 'sensor_id.parameter_id.parameter_type') def _compute_deviation(self): pref = self.env.company.x_fc_default_temp_uom or 'C' for r in self: sensor = r.sensor_id if not sensor: r.deviation_from_target = 0.0 r.deviation_band = 'none' continue raw_sp = sensor._get_setpoint() if not raw_sp: r.deviation_from_target = 0.0 r.deviation_band = 'none' continue # Convert setpoint + value to display unit for the delta so the # number shown matches what operators expect (°F if pref=F). ptype = (sensor.parameter_id.parameter_type or '').lower() if ptype == 'temperature' and pref == 'F': sp_disp = raw_sp * 9.0 / 5.0 + 32.0 else: sp_disp = raw_sp delta = r.display_value - sp_disp r.deviation_from_target = round(delta, 2) if not r.in_spec: r.deviation_band = 'out' elif abs(delta) <= 1.0: r.deviation_band = 'on' elif abs(delta) <= 3.0: r.deviation_band = 'near' else: r.deviation_band = 'far' source = fields.Selection( [ ('iot_proxy', 'IoT Proxy (Pi)'), ('http_ingest', 'HTTP Ingest (direct)'), ('manual', 'Manual Entry'), ], string='Source', default='http_ingest', required=True, ) in_spec = fields.Boolean( string='In Spec', readonly=True, help='Whether this reading fell within the sensor\'s alert range.', ) hold_id = fields.Many2one( 'fusion.plating.quality.hold', string='Resulting Hold', ondelete='set null', readonly=True, help='The quality hold auto-created by this reading, if any. ' 'Only the FIRST out-of-spec reading in an excursion creates ' 'a hold; subsequent readings during the same excursion do ' 'not duplicate.', ) display_name = fields.Char( string='Display', compute='_compute_display_name', store=True, ) @api.depends('sensor_id', 'display_value', 'display_unit', 'reading_at') def _compute_display_name(self): for r in self: sensor = r.sensor_id.name or 'sensor' at = fields.Datetime.to_string(r.reading_at) if r.reading_at else '' r.display_name = f'{sensor} — {r.display_value:.2f} {r.display_unit} @ {at}' # ------------------------------------------------------------------ # Create hook — evaluate against spec + raise a quality hold if we # just crossed INTO an out-of-spec state. # ------------------------------------------------------------------ @api.model_create_multi def create(self, vals_list): records = super().create(vals_list) for rec in records: try: rec._evaluate_spec() except Exception: # Never let alert-logic break the ingest path — the # reading itself is what matters for compliance. Log # and carry on. _logger.exception( 'fp.tank.reading alert eval failed for reading %s', rec.id, ) return records def _evaluate_spec(self): """Set `in_spec`, update sensor cache, raise hold if this reading is the first out-of-spec reading of a new excursion. """ self.ensure_one() sensor = self.sensor_id lo, hi = sensor._get_alert_range() # Zero-bounded checks: a 0 value means "no bound defined" ok_lo = (lo == 0.0) or (self.value >= lo) ok_hi = (hi == 0.0) or (self.value <= hi) in_spec = ok_lo and ok_hi self.in_spec = in_spec # Track excursion transitions on the sensor so we only fire ONE # hold per out-of-spec episode, not one per reading. previously_in_spec = sensor.last_reading_in_spec sensor.sudo().write({ 'last_reading_value': self.value, 'last_reading_at': self.reading_at, 'last_reading_in_spec': in_spec, }) # Crossed from in-spec → out-of-spec on this reading newly_excursion = (previously_in_spec and not in_spec) first_reading_and_bad = (sensor.reading_count == 1 and not in_spec) if (newly_excursion or first_reading_and_bad) and sensor.alert_on_out_of_spec: self._raise_quality_hold() def _raise_quality_hold(self): """Create a quality hold describing the out-of-spec reading.""" self.ensure_one() Hold = self.env.get('fusion.plating.quality.hold') if Hold is None: return # quality module not installed sensor = self.sensor_id lo, hi = sensor._get_alert_range() parts = [ f'Sensor {sensor.name!r} reading {self.value:.2f} ' f'{self.unit or ""} is out of spec.', f'Target range: {lo:.2f} .. {hi:.2f}.', ] if sensor.tank_id: parts.append(f'Tank: {sensor.tank_id.name}.') if sensor.bath_id: parts.append(f'Bath: {sensor.bath_id.name}.') description = ' '.join(parts) hold_vals = { 'hold_reason': 'out_of_spec', 'description': description, 'qty_on_hold': 1, # state defaults to 'on_hold' — leave it } # Attach facility + work-centre context if the tank has them, # so the hold is actionable from the shop floor (operator can # navigate back to the tank from the hold record). if sensor.tank_id: if 'facility_id' in Hold._fields: tank_facility = getattr(sensor.tank_id, 'facility_id', None) if tank_facility: hold_vals['facility_id'] = tank_facility.id if 'part_ref' in Hold._fields: hold_vals['part_ref'] = f'Tank {sensor.tank_id.name} bath' try: hold = Hold.sudo().create(hold_vals) self.hold_id = hold.id _logger.info( 'fp.tank.reading %s triggered quality hold %s (%.2f %s out of %.2f..%.2f)', self.id, hold.id, self.value, self.unit or '', lo, hi, ) except Exception: _logger.exception( 'Could not create quality hold for reading %s', self.id, )