# -*- coding: utf-8 -*- # Copyright 2026 Nexa Systems Inc. # License OPL-1 (Odoo Proprietary License v1.0) import logging from odoo import http, fields from odoo.http import request _logger = logging.getLogger(__name__) class SensorController(http.Controller): """JSON-RPC endpoint for IoT devices to push sensor readings.""" @http.route( '/fp/sensor/measure', type='jsonrpc', auth='user', methods=['POST'], ) def sensor_measure(self, uuid=None, value=None, value_text=None, value_bool=None, effective_at=None, comment=None): """Record a measurement from an IoT device or external API. Args: uuid: Sensor UUID (required) value: Numeric reading (for NUMBER sensors) value_text: Text reading (for TEXT sensors) value_bool: Boolean reading (for BOOLEAN sensors) effective_at: ISO datetime string (optional, defaults to now) comment: Optional note Returns: dict with ok=True and measurement_id on success """ if not uuid: return {'ok': False, 'error': 'uuid is required'} sensor = request.env['fp.sensor'].sudo().search( [('uuid', '=', uuid)], limit=1, ) if not sensor: return {'ok': False, 'error': f'No sensor with UUID {uuid}'} vals = { 'sensor_id': sensor.id, 'source': 'api', 'creator_id': request.env.uid, } if effective_at: vals['effective_at'] = effective_at if comment: vals['comment'] = comment mtype = sensor.measurement_type if mtype == 'number': if value is None: return {'ok': False, 'error': 'value is required for NUMBER sensors'} vals['value'] = float(value) elif mtype == 'text': if value_text is None: return {'ok': False, 'error': 'value_text is required for TEXT sensors'} vals['value_text'] = str(value_text) elif mtype == 'boolean': if value_bool is None: return {'ok': False, 'error': 'value_bool is required for BOOLEAN sensors'} vals['value_bool'] = bool(value_bool) measurement = request.env['fp.sensor.measurement'].sudo().create(vals) _logger.info( 'Sensor %s (%s): recorded measurement %s via API', sensor.name, uuid, measurement.name, ) return {'ok': True, 'measurement_id': measurement.id}