509 lines
20 KiB
Python
509 lines
20 KiB
Python
# identification.py
|
|
"""辨识与容积测量管理器。
|
|
|
|
纯业务逻辑，无 UI 依赖。通过回调与 UI 层通信。
|
|
"""
|
|
|
|
import io
|
|
import time
|
|
import threading
|
|
import datetime
|
|
import json
|
|
import traceback
|
|
import requests
|
|
from collections import deque
|
|
|
|
from get_V import measure_volume
|
|
from ind_collector import collect_data_with_prbs
|
|
|
|
from api import device_post, the_folder
|
|
|
|
|
|
class IdentificationManager:
|
|
"""管理系统辨识与容积测量任务"""
|
|
|
|
def __init__(self):
|
|
self._identifying = False
|
|
self._task_thread = None
|
|
self._on_log = None
|
|
self._on_sample = None # 采样回调: (valve_cmd, pressure)
|
|
self._on_volume_result = None # 容积结果回调: (volume_L: float)
|
|
self._on_identification_upload = None
|
|
|
|
# ---- 回调设置 ----
|
|
def set_log_callback(self, callback):
|
|
"""设置日志回调"""
|
|
self._on_log = callback
|
|
|
|
def set_sample_callback(self, callback):
|
|
"""设置采样时段 UI 更新回调: callback(valve_cmd, pressure)"""
|
|
self._on_sample = callback
|
|
|
|
def set_volume_result_callback(self, callback):
|
|
"""设置容积测量结果回调: callback(volume_L: float)"""
|
|
self._on_volume_result = callback
|
|
|
|
def set_identification_upload_callback(self, callback):
|
|
"""设置辨识 CSV 上传结果回调: callback(success, filename, error)"""
|
|
self._on_identification_upload = callback
|
|
|
|
def log(self, message):
|
|
if self._on_log:
|
|
self._on_log(message)
|
|
|
|
@property
|
|
def is_running(self) -> bool:
|
|
"""当前是否正在辨识/测量中"""
|
|
thread_alive = (
|
|
self._task_thread is not None and self._task_thread.is_alive()
|
|
)
|
|
return self._identifying or thread_alive
|
|
|
|
def _upload_to_server(self, content, filename: str, folder: str) -> bool:
|
|
"""向 ReinLoop 云服务器申请上传地址并上传文本或字节数据。
|
|
|
|
返回 True 表示上传成功，False 表示失败（已内部记 log）。
|
|
"""
|
|
# Step 1: 向业务服务器申请一次性上传地址（不传文件内容）
|
|
try:
|
|
result = device_post({
|
|
"type": "uploadDataFile",
|
|
"fileName": filename,
|
|
"folder": folder,
|
|
}, timeout=30)
|
|
except Exception as e:
|
|
self.log(f"向云服务器申请上传地址异常: {e}")
|
|
return False
|
|
|
|
if not result.get("success"):
|
|
self.log(f"申请服务器上传地址失败: {result.get('errMsg', result)}")
|
|
return False
|
|
|
|
meta = result.get("uploadMetadata")
|
|
if not meta or "url" not in meta:
|
|
self.log("云服务器未返回有效的上传地址")
|
|
return False
|
|
|
|
# Step 2: multipart 上传到云服务器提供的一次性地址
|
|
try:
|
|
content_bytes = (
|
|
content if isinstance(content, bytes)
|
|
else str(content).encode("utf-8")
|
|
)
|
|
files = {"file": (filename, io.BytesIO(content_bytes))}
|
|
upload_resp = requests.post(meta["url"], files=files, timeout=60)
|
|
|
|
if upload_resp.status_code in [200, 204]:
|
|
return True
|
|
else:
|
|
self.log(f"云服务器上传失败，状态码: {upload_resp.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
self.log(f"云服务器上传异常: {e}")
|
|
return False
|
|
|
|
def _run_initial_travel_scan(self, conn_mgr):
|
|
"""Scan 1000..0 and upload each travel's stable pressure as JSON.
|
|
|
|
This is an independent pre-scan. It does not replace or modify the
|
|
subsequent PRBS collection performed by ``collect_data_with_prbs``.
|
|
"""
|
|
if getattr(conn_mgr, "simulation", False):
|
|
stable_pressure_records = [
|
|
{"distance": 1000, "pressure": 400},
|
|
{"distance": 900, "pressure": 361},
|
|
{"distance": 800, "pressure": 322},
|
|
{"distance": 700, "pressure": 283},
|
|
{"distance": 600, "pressure": 244},
|
|
{"distance": 500, "pressure": 205},
|
|
{"distance": 400, "pressure": 166},
|
|
{"distance": 300, "pressure": 127},
|
|
{"distance": 200, "pressure": 88},
|
|
{"distance": 100, "pressure": 49},
|
|
{"distance": 0, "pressure": 10},
|
|
]
|
|
timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
filename = f"travel_stability_pressures_{timestamp}.json"
|
|
payload = {"stable_pressures": stable_pressure_records}
|
|
uploaded = self._upload_to_server(
|
|
json.dumps(payload, ensure_ascii=False, indent=2),
|
|
filename,
|
|
f"{the_folder}/ind_data",
|
|
)
|
|
if uploaded:
|
|
self.log("行程稳态压力 JSON 上传成功，继续执行 PRBS 辨识")
|
|
else:
|
|
self.log("行程稳态压力 JSON 上传失败，继续执行 PRBS 辨识")
|
|
return payload
|
|
|
|
distances = list(range(1000, -1, -100))
|
|
settings = {
|
|
"min_wait_time": 5.0,
|
|
"sample_interval": 0.1,
|
|
"stable_window": 5.0,
|
|
"pressure_tolerance": 0.5,
|
|
"slope_tolerance": 0.05,
|
|
"stable_duration": 3.0,
|
|
"max_wait_time": 60.0,
|
|
}
|
|
stable_pressure_records = []
|
|
stopped = False
|
|
|
|
def slope(points):
|
|
mean_t = sum(point[0] for point in points) / len(points)
|
|
mean_p = sum(point[1] for point in points) / len(points)
|
|
denominator = sum((point[0] - mean_t) ** 2 for point in points)
|
|
if denominator == 0:
|
|
return 0.0
|
|
return sum(
|
|
(point[0] - mean_t) * (point[1] - mean_p)
|
|
for point in points
|
|
) / denominator
|
|
|
|
try:
|
|
for distance in distances:
|
|
if not self._identifying:
|
|
stopped = True
|
|
break
|
|
if not conn_mgr.set_motor_position(float(distance)):
|
|
self.log(f"行程 {distance} 写入失败")
|
|
continue
|
|
|
|
self.log(f"行程 {distance} 已写入，等待压力稳态")
|
|
stage_start = time.monotonic()
|
|
window = deque()
|
|
stable_since = None
|
|
stable_pressure = None
|
|
pressure_range = None
|
|
pressure_slope = None
|
|
|
|
while time.monotonic() - stage_start < settings["max_wait_time"]:
|
|
if not self._identifying:
|
|
stopped = True
|
|
break
|
|
|
|
sample_start = time.monotonic()
|
|
elapsed = sample_start - stage_start
|
|
pressure = conn_mgr.read_pressure()
|
|
if pressure is not None:
|
|
pressure = float(pressure)
|
|
if self._on_sample:
|
|
# Do not expose the confidential travel command.
|
|
self._on_sample(None, pressure)
|
|
|
|
if elapsed >= settings["min_wait_time"]:
|
|
window.append([elapsed, pressure])
|
|
cutoff = elapsed - settings["stable_window"]
|
|
while window and window[0][0] < cutoff:
|
|
window.popleft()
|
|
|
|
window_span = (
|
|
window[-1][0] - window[0][0]
|
|
if len(window) > 1 else 0
|
|
)
|
|
if window_span >= (
|
|
settings["stable_window"] -
|
|
settings["sample_interval"] * 1.5):
|
|
pressures = [point[1] for point in window]
|
|
pressure_range = max(pressures) - min(pressures)
|
|
pressure_slope = slope(window)
|
|
stable_now = (
|
|
pressure_range <= settings["pressure_tolerance"] and
|
|
abs(pressure_slope) <= settings["slope_tolerance"]
|
|
)
|
|
if stable_now:
|
|
if stable_since is None:
|
|
stable_since = sample_start
|
|
elif sample_start - stable_since >= settings["stable_duration"]:
|
|
stable_pressure = sum(pressures) / len(pressures)
|
|
break
|
|
else:
|
|
stable_since = None
|
|
|
|
remaining = (
|
|
settings["sample_interval"] -
|
|
(time.monotonic() - sample_start)
|
|
)
|
|
if remaining > 0:
|
|
time.sleep(remaining)
|
|
|
|
if stopped:
|
|
break
|
|
if stable_pressure is None:
|
|
self.log(f"行程 {distance} 在 60 秒内未达到稳态")
|
|
continue
|
|
|
|
stable_pressure_records.append({
|
|
"distance": distance,
|
|
"pressure": float(stable_pressure),
|
|
})
|
|
self.log(
|
|
f"行程 {distance} 达到稳态，压力 {stable_pressure:.3f} kPa"
|
|
)
|
|
finally:
|
|
# The requested sequence ends at zero; also return there on stop.
|
|
conn_mgr.set_motor_position(0)
|
|
|
|
timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
filename = f"travel_stability_pressures_{timestamp}.json"
|
|
payload = {"stable_pressures": stable_pressure_records}
|
|
uploaded = self._upload_to_server(
|
|
json.dumps(payload, ensure_ascii=False, indent=2),
|
|
filename,
|
|
f"{the_folder}/ind_data",
|
|
)
|
|
if uploaded:
|
|
self.log("行程稳态压力 JSON 上传成功，继续执行 PRBS 辨识")
|
|
else:
|
|
self.log("行程稳态压力 JSON 上传失败，继续执行 PRBS 辨识")
|
|
return payload
|
|
|
|
# ---- 系统辨识 ----
|
|
def start_identification(self, *,
|
|
conn_mgr,
|
|
running_flag_check,
|
|
q_in_val: float, dt: float,
|
|
n_order: int, t_c: float,
|
|
levels: list, dead_area: float,
|
|
xa_full: float, V_val: float,
|
|
repeat: int = 2):
|
|
"""启动辨识数据采集（在后台线程中运行）
|
|
|
|
Args:
|
|
conn_mgr: ConnectionManager 实例
|
|
running_flag_check: 检查是否应该停止的可调用对象, 返回 bool
|
|
q_in_val: 流量 (L/min)
|
|
dt: 控制周期
|
|
n_order: 阶数
|
|
t_c: 周期 (s)
|
|
levels: 序列 (阀门开度列表)
|
|
dead_area: 死区
|
|
xa_full: 总限幅
|
|
V_val: 容积 (L)
|
|
repeat: 整段复合序列重复次数，默认 2
|
|
"""
|
|
if running_flag_check():
|
|
self.log("错误：请先停止控制再进行辨识")
|
|
return False
|
|
|
|
if self.is_running:
|
|
self.log("辨识正在进行中，请等待完成")
|
|
return False
|
|
|
|
if not conn_mgr or not conn_mgr.is_connected():
|
|
self.log("错误：请先连接设备")
|
|
return False
|
|
|
|
self._identifying = True
|
|
self.log("开始辨识数据采集...")
|
|
|
|
def _on_sample_point(t, u_cmd, p):
|
|
if self._on_sample:
|
|
self._on_sample(u_cmd, p)
|
|
|
|
def collect_thread():
|
|
try:
|
|
# Independent pre-scan. The PRBS call below is intentionally
|
|
# left unchanged and starts after the travel scan completes.
|
|
self._run_initial_travel_scan(conn_mgr)
|
|
if not self._identifying:
|
|
return
|
|
|
|
result = collect_data_with_prbs(
|
|
conn_mgr,
|
|
q_in_val=q_in_val, dt=dt,
|
|
n_order=n_order, t_c=t_c,
|
|
levels=levels, dead_area=dead_area,
|
|
xa_full=xa_full,
|
|
V_val=V_val,
|
|
should_stop=lambda: not self._identifying,
|
|
log=self.log,
|
|
on_sample=_on_sample_point,
|
|
repeat=repeat,
|
|
)
|
|
|
|
if result.get('success'):
|
|
csv_data = result.get("csv_data")
|
|
csv_filename = result.get("filename")
|
|
if not csv_data or not csv_filename:
|
|
error = "辨识采集结果缺少 CSV 数据或文件名"
|
|
self.log(error)
|
|
if self._on_identification_upload:
|
|
self._on_identification_upload(False, None, error)
|
|
elif self._upload_to_server(
|
|
csv_data, csv_filename, f"{the_folder}/ind_data"):
|
|
self.log("辨识数据上传成功")
|
|
if self._on_identification_upload:
|
|
self._on_identification_upload(
|
|
True, csv_filename, None
|
|
)
|
|
else:
|
|
self.log("辨识数据上传失败")
|
|
if self._on_identification_upload:
|
|
self._on_identification_upload(
|
|
False, csv_filename, "辨识 CSV 上传失败"
|
|
)
|
|
else:
|
|
self.log("辨识未采集到数据")
|
|
if self._on_identification_upload:
|
|
self._on_identification_upload(
|
|
False, None, "辨识未采集到数据"
|
|
)
|
|
|
|
except Exception as e:
|
|
self.log(f"辨识数据采集详细错误: {traceback.format_exc()}")
|
|
self.log(f"辨识数据采集失败: {e}")
|
|
if self._on_identification_upload:
|
|
self._on_identification_upload(False, None, str(e))
|
|
finally:
|
|
self._identifying = False
|
|
# self.log("辨识结束")
|
|
|
|
thread = threading.Thread(target=collect_thread, daemon=True)
|
|
self._task_thread = thread
|
|
thread.start()
|
|
return True
|
|
|
|
# ---- 容积测量 ----
|
|
def start_volume_measurement(self, *,
|
|
conn_mgr,
|
|
running_flag_check,
|
|
q_in_val: float, dt: float,
|
|
p_max: float, fit_low: float,
|
|
fit_high: float, T_delta: float,
|
|
xa_full: float = 1000,
|
|
num_runs: int = 3):
|
|
"""启动容积测量（在后台线程中运行）"""
|
|
if running_flag_check():
|
|
self.log("错误：请先停止控制再进行测试")
|
|
return False
|
|
|
|
if self.is_running:
|
|
self.log("测试正在进行中，请等待完成")
|
|
return False
|
|
|
|
if not conn_mgr or not conn_mgr.is_connected():
|
|
self.log("错误：请先连接设备")
|
|
return False
|
|
|
|
self._identifying = True
|
|
self.log("开始测量容积...")
|
|
|
|
def _on_vol_sample(t, p):
|
|
if self._on_sample:
|
|
self._on_sample(None, p)
|
|
|
|
def volume_thread():
|
|
all_results = [] # 存储每次成功的结果
|
|
|
|
try:
|
|
for run_idx in range(num_runs):
|
|
if not self._identifying:
|
|
break
|
|
|
|
print(f"--- 第 {run_idx + 1}/{num_runs} 次测量 ---")
|
|
|
|
# 非首次测量前，等待压力回落
|
|
if run_idx > 0:
|
|
print("等待压力回落...")
|
|
wait_start = time.time()
|
|
while time.time() - wait_start < 60: # 最多等 60 秒
|
|
p = conn_mgr.read_pressure()
|
|
if p is not None and p < fit_low:
|
|
print(f"压力已回落至 {p:.1f} kPa，等待 10 秒稳定...")
|
|
time.sleep(10)
|
|
break
|
|
time.sleep(1)
|
|
else:
|
|
print("等待压力回落超时，跳过剩余测量")
|
|
break
|
|
|
|
result = measure_volume(
|
|
conn_mgr,
|
|
q_in_slm=q_in_val,
|
|
dt=dt,
|
|
xa=xa_full,
|
|
p_max=p_max,
|
|
fit_low=fit_low,
|
|
fit_high=fit_high,
|
|
T_delta=T_delta,
|
|
should_stop=lambda: not self._identifying,
|
|
log=self.log,
|
|
on_sample=_on_vol_sample,
|
|
)
|
|
|
|
if result.get('success'):
|
|
all_results.append(result)
|
|
print(f"第 {run_idx + 1} 次测量成功，V = {result['volume_L']:.4f} L")
|
|
else:
|
|
print(f"第 {run_idx + 1} 次测量失败")
|
|
|
|
# ---- 汇总 ----
|
|
if all_results:
|
|
n = len(all_results)
|
|
|
|
# 平均关键参数
|
|
avg_vol = sum(r['volume_L'] for r in all_results) / n
|
|
avg_slope = sum(r['slope'] for r in all_results) / n
|
|
avg_intercept = sum(r['intercept'] for r in all_results) / n
|
|
avg_c1 = sum(r['c1'] for r in all_results) / n
|
|
|
|
# 构建上传数据
|
|
timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
individual_runs = []
|
|
for i, r in enumerate(all_results):
|
|
individual_runs.append({
|
|
"run": i + 1,
|
|
"volume_L": r['volume_L'],
|
|
"slope": r['slope'],
|
|
"intercept": r['intercept'],
|
|
"c1": r['c1'],
|
|
"valid_points": r['valid_points'],
|
|
"record_time": r.get('record_time', []),
|
|
"p_actual": r.get('p_actual', []),
|
|
})
|
|
|
|
full_data = {
|
|
"num_runs_total": num_runs,
|
|
"num_runs_successful": n,
|
|
"averaged": {
|
|
"volume_L": avg_vol,
|
|
"slope": avg_slope,
|
|
"intercept": avg_intercept,
|
|
"c1": avg_c1,
|
|
},
|
|
"individual_runs": individual_runs,
|
|
"q_in_slm": all_results[0]['payload_data'].get('q_in_slm'),
|
|
"T_delta": T_delta,
|
|
}
|
|
json_str = json.dumps(full_data, indent=2, ensure_ascii=False)
|
|
filename = f"volume_test_avg{avg_vol:.2f}L_{timestamp}.json"
|
|
|
|
if self._upload_to_server(json_str, filename, f"{the_folder}/V_config"):
|
|
self.log("体积测量数据上传成功")
|
|
self.log(f"成功：{n}/{num_runs}，平均等效体积 V = {avg_vol:.4f} L")
|
|
|
|
else:
|
|
self.log("体积测量数据上传失败")
|
|
|
|
if self._on_volume_result:
|
|
self._on_volume_result(avg_vol)
|
|
else:
|
|
self.log("所有测量均失败：有效数据点不足，无法计算体积")
|
|
|
|
except Exception as e:
|
|
self.log(f"容积测量详细错误: {traceback.format_exc()}")
|
|
self.log(f"容积测量失败: {e}")
|
|
finally:
|
|
self._identifying = False
|
|
# self.log("测量结束")
|
|
|
|
thread = threading.Thread(target=volume_thread, daemon=True)
|
|
self._task_thread = thread
|
|
thread.start()
|
|
return True
|
|
|
|
def stop(self):
|
|
"""停止当前辨识/测量任务"""
|
|
self._identifying = False
|