《手札·开源篇》Odoo系统与SKF Observer Phoenix API双向对接方案

Odoo系统与SKF Observer Phoenix API双向对接方案

一、方案设计概述

（一）目标
实现Odoo设备维护模块与SKF Observer Phoenix的传感器数据双向同步：

1. Odoo→SKF：推送维护工单状态、设备档案信息。

2. SKF→Odoo：同步设备传感器数据（振动、温度、转速等）和预测性维护建议。

（二）技术架构

1. Odoo端：使用XML-RPC/JSON-RPC接口对外暴露API，并通过自定义模块处理数据逻辑。

2. SKF端：通过RESTful API获取传感器数据，并支持Webhook接收Odoo的工单状态更新。

3. 中间层：Python脚本作为调度器，定时拉取SKF数据并触发Odoo业务逻辑。

---

二、数据模型与接口定义

（一）Odoo数据模型（自定义模块）

# models/equipment.py
from odoo import models, fieldsclass Equipment(models.Model):_name = 'maintenance.equipment.skf'_description = 'SKF设备档案'name = fields.Char('设备名称', required=True)skf_id = fields.Char('SKF设备ID')  # SKF系统的设备唯一标识vibration_threshold = fields.Float('振动阈值（mm/s）')last_sync_time = fields.Datetime('最后同步时间')sensor_data = fields.One2many('skf.sensor.data', 'equipment_id', '传感器数据')class SkfSensorData(models.Model):_name = 'skf.sensor.data'timestamp = fields.Datetime('时间戳')vibration = fields.Float('振动值（mm/s）')temperature = fields.Float('温度（℃）')equipment_id = fields.Many2one('maintenance.equipment.skf', '设备')

（二）SKF API接口定义

1.数据拉取接口（SKF→Odoo）

• URL:https://api.skf.com/observer/v1/devices/{device_id}/sensor_data

• 方法:GET

• 参数:start_time,end_time

• 响应示例：

  {"device_id": "SKF-12345","data": [{"timestamp": "2025-02-10T14:30:00Z", "vibration": 2.5, "temperature": 45.2},{"timestamp": "2025-02-10T14:35:00Z", "vibration": 3.1, "temperature": 47.8}]}

2.工单状态推送接口（Odoo→SKF）

• URL:https://api.skf.com/observer/v1/workorders/{workorder_id}

• 方法:PATCH

• 请求体：

  {"status": "completed", "notes": "维护完成，更换轴承"}

---

三、双向对接代码实现

（一）Odoo→SKF：推送工单状态

# controllers/skf_api.py
from odoo import http
import requests
import jsonclass SkfController(http.Controller):@http.route('/skf/update_workorder', type='json', auth='user')def update_workorder(self, workorder_id, status, notes):# 获取工单数据workorder = http.request.env['maintenance.request'].browse(workorder_id)if not workorder.exists():return {'error': '工单不存在'}# 调用SKF API更新状态skf_api_key = http.request.env['ir.config_parameter'].get_param('skf.api_key')headers = {'Authorization': f'Bearer {skf_api_key}', 'Content-Type': 'application/json'}url = f'https://api.skf.com/observer/v1/workorders/{workorder.skf_workorder_id}'payload = {'status': status, 'notes': notes}try:response = requests.patch(url, headers=headers, data=json.dumps(payload))response.raise_for_status()workorder.write({'skf_sync_status': 'synced'})return {'success': True}except Exception as e:return {'error': str(e)}

（二）SKF→Odoo：定时同步传感器数据

# models/skf_sync.py
from odoo import models, fields, api
import requests
import logging
_logger = logging.getLogger(__name__)class SkfDataSync(models.Model):_name = 'skf.data.sync'@api.modeldef cron_sync_sensor_data(self):# 获取所有SKF设备equipments = self.env['maintenance.equipment.skf'].search([('skf_id', '!=', False)])skf_api_key = self.env['ir.config_parameter'].get_param('skf.api_key')for equipment in equipments:url = f'https://api.skf.com/observer/v1/devices/{equipment.skf_id}/sensor_data'params = {'start_time': equipment.last_sync_time.isoformat() if equipment.last_sync_time else '2025-01-01T00:00:00Z'}headers = {'Authorization': f'Bearer {skf_api_key}'}try:response = requests.get(url, headers=headers, params=params)data = response.json()for entry in data.get('data', []):self.env['skf.sensor.data'].create({'equipment_id': equipment.id,'timestamp': entry['timestamp'],'vibration': entry['vibration'],'temperature': entry['temperature']})equipment.last_sync_time = fields.Datetime.now()_logger.info(f"设备 {equipment.name} 数据同步成功")except Exception as e:_logger.error(f"同步失败: {str(e)}")

（三）SKF Webhook接收Odoo触发事件

# controllers/skf_webhook.py
from odoo import http
import jsonclass SkfWebhookController(http.Controller):@http.route('/skf/webhook/maintenance_alert', type='json', auth='public')def handle_maintenance_alert(self, **kwargs):data = http.request.jsonrequestequipment = http.request.env['maintenance.equipment.skf'].search([('skf_id', '=', data['device_id'])])if equipment:# 创建维护工单workorder = http.request.env['maintenance.request'].create({'name': f"预警维护：{equipment.name}",'equipment_id': equipment.id,'description': f"SKF系统预警：{data['alert_message']}",'priority': 'high'})return {'odoo_workorder_id': workorder.id}else:return {'error': '设备未找到'}

---

四、安全与配置

（一）API密钥管理
在Odoo的系统参数中存储SKF API密钥（skf.api_key），通过加密字段保护。

（二）HTTPS加密
所有API调用均通过HTTPS传输，SKF端需提供SSL证书。

（三）IP白名单
限制SKF API仅允许Odoo服务器的IP访问。

---

五、测试与部署

（一）单元测试

# 测试数据同步
def test_sensor_sync(self):equipment = self.env.ref('skf_integration.equipment_skf_1')equipment.last_sync_time = '2025-02-09 00:00:00'self.env['skf.data.sync'].cron_sync_sensor_data()self.assertEqual(len(equipment.sensor_data), 10, "应同步10条传感器数据")

（二）部署步骤

1. 安装Odoo自定义模块。

2. 配置定时任务（如每30分钟同步一次传感器数据）。

3. 在SKF Observer Phoenix中注册Odoo的Webhook URL。

---

六、真实数据示例

（一）SKF传感器数据

{"device_id": "SKF-12345","data": [{"timestamp": "2025-02-10T14:40:00Z", "vibration": 4.2, "temperature": 50.1}]
}

（二）Odoo工单推送

{"status": "in_progress", "notes": "开始更换轴承"}

---

七、扩展建议

（一）数据可视化
在Odoo中集成Grafana，实时展示设备健康状态。

（二）预测性维护
基于SKF数据训练机器学习模型，预测设备故障时间。

---

让转型不迷航——邹工转型手札