ultrafiltration_model/test_plc_update.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
PLC指令测试工具
用于测试PLC参数更新请求的实际发送
"""

import requests
import json
import hashlib
import time


def load_config(config_file='config.json'):
    """加载配置文件"""
    with open(config_file, 'r', encoding='utf-8') as f:
        return json.load(f)


def generate_md5_signature(record_data, secret, timestamp):
    """生成MD5签名"""
    cal_str = f"{record_data}{secret}{timestamp}"
    cal_md5 = hashlib.md5(cal_str.encode('utf-8')).hexdigest()
    return cal_md5.upper()


def send_plc_update_test(device_name, item, old_value, new_value, command_type):
    """
    发送PLC参数更新测试
    
    参数:
        device_name: 设备名称
        item: 参数项名称
        old_value: 当前值
        new_value: 目标值
        command_type: 命令类型
        
    返回:
        是否发送成功
    """
    config = load_config()
    
    PLC_URL = config['api']['base_url'] + config['api']['plc_endpoint']
    PROJECT_ID = config['scada']['project_id']
    SCADA_SECRET = config['scada']['secret']
    
    timestamp = int(time.time())
    
    record_data = json.dumps({
        "project_id": PROJECT_ID,
        "item": item,
        "old_value": old_value,
        "new_value": new_value,
        "command_type": command_type
    }, separators=(',', ':'))
    
    signature = generate_md5_signature(record_data, SCADA_SECRET, timestamp)
    url = f"{PLC_URL}?sign={signature}&timestamp={timestamp}"
    
    payload = [{
        "project_id": PROJECT_ID,
        "item": item,
        "old_value": old_value,
        "new_value": new_value,
        "command_type": command_type
    }]
    
    print(f"PLC测试")
    print(f"设备 {device_name}")
    print(f"参数 {item}")
    print(f"旧值 {old_value}")
    print(f"新值 {new_value}")
    print(f"类型 {command_type}")
    print(f"时间戳 {timestamp}")
    print(f"URL {url}")
    print(f"请求体 {json.dumps(payload, indent=2, ensure_ascii=False)}")
    print(f"签名数据 {record_data}")
    print(f"签名 {signature}")
    print("-" * 50)
    
    try:
        headers = {"Content-Type": "application/json"}
        response = requests.post(url, headers=headers, json=payload, timeout=15)
        
        print(f"响应状态码 {response.status_code}")
        print(f"响应头 {dict(response.headers)}")
        
        try:
            response_json = response.json()
            print(f"响应JSON {json.dumps(response_json, indent=2, ensure_ascii=False)}")
        except:
            print(f"响应文本 {response.text}")
            
        response.raise_for_status()
        print("请求发送成功")
        return True
        
    except requests.exceptions.RequestException as e:
        print(f"请求失败 {e}")
        return False
    except Exception as e:
        print(f"未知错误 {e}")
        return False


if __name__ == "__main__":
    # 测试配置
    device_name = "UF2"
    item = "C.M.UF2_DB@time_production"
    old_value = "3800"
    new_value = "3801"
    command_type = 1
    
    print("开始PLC指令测试")
    success = send_plc_update_test(device_name, item, old_value, new_value, command_type)
    
    if success:
        print("\n测试完成 请检查PLC系统")
    else:
        print("\n测试失败 请检查网络和配置")