《手札·开源篇》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, fields

class 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 json

class 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.model
    def 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 json

class SkfWebhookController(http.Controller):
    @http.route('/skf/webhook/maintenance_alert', type='json', auth='public')
    def handle_maintenance_alert(self, **kwargs):
        data = http.request.jsonrequest
        equipment = 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数据训练机器学习模型，预测设备故障时间。

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