louis/ReinLoopTest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update panel

Browse Source
This commit is contained in:
rangang
2026-07-30 11:40:00 +08:00
parent 4312cb878c
commit e12134e40d
16 changed files with 3327 additions and 2775 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ReinLoop/core/data_collector.py
+234 -199
View File
@@ -1,199 +1,234 @@
# data_collector.py
"""数据采集器：管理 Episode 数据记录与上上传。
纯业务逻辑，无 UI 依赖。通过回调与 UI 层通信。
"""
import io
import json
import pickle
import datetime
import threading
import requests
from api import base_url, data_record_url, the_folder
class DataCollector:
"""管理控制过程中的 Episode 数据采集与保存"""
def __init__(self):
self.episode_data_raw = [] # 所有已完成的 Episode
self.current_episode = None # 当前正在记录的 Episode
self.last_target_record = None
self._on_log = None
def set_log_callback(self, callback):
"""设置日志回调"""
self._on_log = callback
def log(self, message):
if self._on_log:
self._on_log(message)
def _upload_to_cos(self, data_bytes: bytes, filename: str, folder: str) -> bool:
"""通过云函数获取直传凭证，再将数据直传到腾讯云 COS。"""
try:
resp = requests.post(data_record_url, json={
"type": "uploadDataFile",
"fileName": filename,
"folder": folder,
}, timeout=30)
result = resp.json()
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 or "authorization" not in meta:
self.log("云端未返回有效的上传元数据")
return False
try:
form_data = {
"key": meta["cosFileId"],
"Signature": meta["authorization"],
"x-cos-security-token": meta["token"],
"x-cos-meta-fileid": meta["fileId"],
}
files = {"file": (filename, io.BytesIO(data_bytes))}
cos_resp = requests.post(meta["url"], data=form_data, files=files, timeout=60)
if cos_resp.status_code in [200, 204]:
return True
else:
self.log(f"COS 直传失败，状态码: {cos_resp.status_code}")
return False
except Exception as e:
self.log(f"COS 直传异常: {e}")
return False
def reset(self):
"""重置所有采集状态（控制启动时调用）"""
self.episode_data_raw = []
self.current_episode = None
self.last_target_record = None
def record_step(self, cycle_count: int, current_pressure: float,
target_pressure: float, valve_opening: float,
kp: float, ki: float, kd: float,
q_in: float, v: float):
"""记录一个控制周期的数据点
Args:
cycle_count: 控制周期计数
current_pressure: 当前压力
target_pressure: 目标压力
valve_opening: 阀门开度
kp, ki, kd: PID 参数
q_in: 流量
v: 容积
"""
# 目标压力变化时自动切分 Episode
if self.current_episode is None or target_pressure != self.last_target_record:
if self.current_episode is not None:
self.episode_data_raw.append(self.current_episode)
self.log(f"Episode 结束，已记录 {len(self.current_episode['pressures'])} 个点")
self.current_episode = {
'pid': [float(kp), float(ki), float(kd)],
'target_pressure': target_pressure,
'Q_in': q_in,
'V': v,
'steps': [],
'pressures': [],
'errors': [],
'valves': []
}
self.last_target_record = target_pressure
# 记录当前步数据
error = -(target_pressure - current_pressure)
self.current_episode['steps'].append(cycle_count)
self.current_episode['pressures'].append(current_pressure)
self.current_episode['errors'].append(error)
self.current_episode['valves'].append(float(valve_opening))
def finalize_and_upload(self, flow: float, vol: float):
"""停止控制时：闭合最后一个 Episode，分片上传到云存储。
单文件超过 5MB 时自动拆分为多个分片，
同时上传一个 manifest.json 记录所有分片信息。
Args:
flow: 流量值 (用于文件名/路径)
vol: 容积值 (用于文件名/路径)
"""
# 闭合最后一个 Episode
if self.current_episode and len(self.current_episode['pressures']) > 0:
self.episode_data_raw.append(self.current_episode)
self.current_episode = None
if not self.episode_data_raw:
return
try:
timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
base_folder = f"{the_folder}/data_record/data_{flow}SLM_{vol}L"
# 拆成每片尽量不超过 5MB 的 episode 分组
MAX_CHUNK_BYTES = 5 * 1024 * 1024 # 5MB
chunks = [] # [(chunk_index, episodes_subset)]
current_chunk = []
for ep in self.episode_data_raw:
current_chunk.append(ep)
if pickle.dumps(current_chunk).__len__() >= MAX_CHUNK_BYTES:
# 当前片已满，回退一个 episode 后保存
current_chunk.pop()
chunks.append(current_chunk)
current_chunk = [ep]
if current_chunk:
chunks.append(current_chunk)
total_chunks = len(chunks)
self.log(f"控制数据共 {len(self.episode_data_raw)} 个 Episode，"
f"拆为 {total_chunks} 个分片上传")
def upload_all():
part_files = []
for idx, chunk_eps in enumerate(chunks):
data_bytes = pickle.dumps(chunk_eps)
size_kb = len(data_bytes) / 1024
part_filename = f'episode_raw_data_{timestamp}_part{idx + 1}of{total_chunks}.pkl'
self.log(f" 上传分片 {idx + 1}/{total_chunks} ({size_kb:.0f} KB)...")
if self._upload_to_cos(data_bytes, part_filename, base_folder):
part_files.append(part_filename)
else:
self.log(f" 分片 {idx + 1} 上传失败")
# 上传 manifest
manifest = {
"timestamp": timestamp,
"total_chunks": total_chunks,
"uploaded_chunks": len(part_files),
"part_files": part_files,
"total_episodes": len(self.episode_data_raw),
"flow": flow,
"volume": vol,
}
manifest_str = json.dumps(manifest, indent=2, ensure_ascii=False)
manifest_bytes = manifest_str.encode('utf-8')
manifest_filename = f'episode_raw_data_{timestamp}_manifest.json'
self._upload_to_cos(manifest_bytes, manifest_filename, base_folder)
if len(part_files) == total_chunks:
self.log(f"控制数据上传成功 ({total_chunks} 个分片)")
else:
self.log(f"控制数据部分上传失败 ({len(part_files)}/{total_chunks})")
threading.Thread(target=upload_all, daemon=True).start()
except Exception as e:
self.log(f"保存收集数据时发生错误: {e}")
finally:
self.episode_data_raw = []
# data_collector.py
"""数据采集器：管理 Episode 数据记录与上上传。
纯业务逻辑，无 UI 依赖。通过回调与 UI 层通信。
"""
import io
import json
import pickle
import datetime
import threading
import requests
from api import base_url, data_record_url, the_folder
class DataCollector:
"""管理控制过程中的 Episode 数据采集与保存"""
def __init__(self):
self.episode_data_raw = [] # 所有已完成的 Episode
self.current_episode = None # 当前正在记录的 Episode
self.last_target_record = None
self._on_log = None
self._on_upload_complete = None
def set_log_callback(self, callback):
"""设置日志回调"""
self._on_log = callback
def set_upload_complete_callback(self, callback):
"""设置控制数据上传完成回调。
callback(success, manifest, error) 会在后台上传线程中调用。成功时
manifest 是已上传的清单字典；失败时 error 为可展示的错误信息。
"""
self._on_upload_complete = callback
def log(self, message):
if self._on_log:
self._on_log(message)
def _notify_upload_complete(self, success, manifest=None, error=None):
if self._on_upload_complete:
self._on_upload_complete(success, manifest, error)
def _upload_to_server(self, data_bytes: bytes, filename: str, folder: str) -> bool:
"""向 ReinLoop 云服务器申请上传地址并上传控制数据。"""
try:
resp = requests.post(data_record_url, json={
"type": "uploadDataFile",
"fileName": filename,
"folder": folder,
}, timeout=30)
result = resp.json()
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
try:
files = {"file": (filename, io.BytesIO(data_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 reset(self):
"""重置所有采集状态（控制启动时调用）"""
self.episode_data_raw = []
self.current_episode = None
self.last_target_record = None
def record_step(self, cycle_count: int, current_pressure: float,
target_pressure: float, valve_opening: float,
kp: float, ki: float, kd: float,
q_in: float, v: float):
"""记录一个控制周期的数据点
Args:
cycle_count: 控制周期计数
current_pressure: 当前压力
target_pressure: 目标压力
valve_opening: 阀门开度
kp, ki, kd: PID 参数
q_in: 流量
v: 容积
"""
# 目标压力变化时自动切分 Episode
if self.current_episode is None or target_pressure != self.last_target_record:
if self.current_episode is not None:
self.episode_data_raw.append(self.current_episode)
self.log(f"Episode 结束，已记录 {len(self.current_episode['pressures'])} 个点")
self.current_episode = {
'pid': [float(kp), float(ki), float(kd)],
'target_pressure': target_pressure,
'Q_in': q_in,
'V': v,
'steps': [],
'pressures': [],
'errors': [],
'valves': []
}
self.last_target_record = target_pressure
# 记录当前步数据
error = -(target_pressure - current_pressure)
self.current_episode['steps'].append(cycle_count)
self.current_episode['pressures'].append(current_pressure)
self.current_episode['errors'].append(error)
self.current_episode['valves'].append(float(valve_opening))
def finalize_and_upload(self, flow: float, vol: float):
"""停止控制时：闭合最后一个 Episode，分片上传到云存储。
单文件超过 5MB 时自动拆分为多个分片，
同时上传一个 manifest.json 记录所有分片信息。
Args:
flow: 流量值 (用于文件名/路径)
vol: 容积值 (用于文件名/路径)
"""
# 闭合最后一个 Episode
if self.current_episode and len(self.current_episode['pressures']) > 0:
self.episode_data_raw.append(self.current_episode)
self.current_episode = None
if not self.episode_data_raw:
return
# 上传在线程中继续执行，因此必须持有本轮数据快照。否则 finally
# 清空缓存后，异步线程生成的 manifest 会错误地显示 0 个 Episode。
episodes = list(self.episode_data_raw)
try:
timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S_%f')
base_folder = f"{the_folder}/data_record/data_{flow}SLM_{vol}L"
# 拆成每片尽量不超过 5MB 的 episode 分组
MAX_CHUNK_BYTES = 5 * 1024 * 1024 # 5MB
chunks = [] # [(chunk_index, episodes_subset)]
current_chunk = []
for ep in episodes:
current_chunk.append(ep)
if pickle.dumps(current_chunk).__len__() >= MAX_CHUNK_BYTES:
# 当前片已满，回退一个 episode 后保存
current_chunk.pop()
# 单个 Episode 也可能超过 5 MB；此时仍上传该 Episode，
# 而不是产生一个无内容的空分片。
if current_chunk:
chunks.append(current_chunk)
current_chunk = [ep]
if current_chunk:
chunks.append(current_chunk)
total_chunks = len(chunks)
self.log(f"控制数据共 {len(episodes)} 个 Episode，"
f"拆为 {total_chunks} 个分片上传")
def upload_all():
part_files = []
part_metadata = []
for idx, chunk_eps in enumerate(chunks):
data_bytes = pickle.dumps(chunk_eps)
size_kb = len(data_bytes) / 1024
part_filename = f'episode_raw_data_{timestamp}_part{idx + 1}of{total_chunks}.pkl'
self.log(f" 上传分片 {idx + 1}/{total_chunks} ({size_kb:.0f} KB)...")
if self._upload_to_server(data_bytes, part_filename, base_folder):
part_files.append(part_filename)
part_metadata.append({
"file_name": part_filename,
"episode_count": len(chunk_eps),
"size_bytes": len(data_bytes),
})
else:
self.log(f" 分片 {idx + 1} 上传失败")
# 上传 manifest
manifest = {
"schema_version": 1,
"data_type": "control_episode",
"run_id": timestamp,
"timestamp": timestamp,
"folder": base_folder,
"total_chunks": total_chunks,
"uploaded_chunks": len(part_files),
"part_files": part_files,
"parts": part_metadata,
"total_episodes": len(episodes),
"flow": flow,
"volume": vol,
}
manifest_str = json.dumps(manifest, indent=2, ensure_ascii=False)
manifest_bytes = manifest_str.encode('utf-8')
manifest_filename = f'episode_raw_data_{timestamp}_manifest.json'
manifest_uploaded = self._upload_to_server(
manifest_bytes, manifest_filename, base_folder
)
if len(part_files) == total_chunks and manifest_uploaded:
self.log(f"控制数据上传成功 ({total_chunks} 个分片)")
self._notify_upload_complete(True, manifest, None)
else:
error = (
"控制数据清单上传失败"
if not manifest_uploaded
else f"控制数据部分上传失败 ({len(part_files)}/{total_chunks})"
)
self.log(error)
self._notify_upload_complete(False, manifest, error)
threading.Thread(target=upload_all, daemon=True).start()
except Exception as e:
self.log(f"保存收集数据时发生错误: {e}")
self._notify_upload_complete(False, None, str(e))
finally:
self.episode_data_raw = []
ReinLoop/core/identification.py
+254 -260
View File
@@ -1,27 +1,27 @@
# identification.py
"""辨识与容积测量管理器。
纯业务逻辑，无 UI 依赖。通过回调与 UI 层通信。
"""
import io
import time
import threading
import datetime
import json
# 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 base_url, data_record_url, the_folder
class IdentificationManager:
"""管理系统辨识与容积测量任务"""
from get_V import measure_volume
from ind_collector import collect_data_with_prbs
from api import base_url, data_record_url, the_folder
class IdentificationManager:
"""管理系统辨识与容积测量任务"""
def __init__(self):
self._identifying = False
self._task_thread = None
@@ -29,16 +29,16 @@ class IdentificationManager:
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_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
@@ -46,67 +46,61 @@ class IdentificationManager:
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 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_cos(self, content, filename: str, folder: str) -> bool:
"""通过云函数获取直传凭证，再将文本或字节数据直传到 COS。
返回 True 表示上传成功，False 表示失败（已内部记 log）。
"""
# Step 1: 向云函数申请直传凭证（不传文件内容）
try:
resp = requests.post(data_record_url, json={
"type": "uploadDataFile",
"fileName": filename,
"folder": folder,
}, timeout=30)
result = resp.json()
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 or "authorization" not in meta:
self.log("云端未返回有效的上传元数据")
return False
# Step 2: 直传到 COS
try:
form_data = {
"key": meta["cosFileId"],
"Signature": meta["authorization"],
"x-cos-security-token": meta["token"],
"x-cos-meta-fileid": meta["fileId"],
}
def _upload_to_server(self, content, filename: str, folder: str) -> bool:
"""向 ReinLoop 云服务器申请上传地址并上传文本或字节数据。
返回 True 表示上传成功，False 表示失败（已内部记 log）。
"""
# Step 1: 向业务服务器申请一次性上传地址（不传文件内容）
try:
resp = requests.post(data_record_url, json={
"type": "uploadDataFile",
"fileName": filename,
"folder": folder,
}, timeout=30)
result = resp.json()
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))}
cos_resp = requests.post(meta["url"], data=form_data, files=files, timeout=60)
if cos_resp.status_code in [200, 204]:
return True
else:
self.log(f"COS 直传失败，状态码: {cos_resp.status_code}")
return False
except Exception as e:
self.log(f"COS 直传异常: {e}")
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):
@@ -226,7 +220,7 @@ class IdentificationManager:
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_cos(
uploaded = self._upload_to_server(
json.dumps(payload, ensure_ascii=False, indent=2),
filename,
f"{the_folder}/ind_data",
@@ -236,31 +230,31 @@ class IdentificationManager:
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
"""
# ---- 系统辨识 ----
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
@@ -272,14 +266,14 @@ class IdentificationManager:
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)
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
@@ -289,18 +283,18 @@ class IdentificationManager:
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,
)
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")
@@ -309,7 +303,7 @@ class IdentificationManager:
self.log(error)
if self._on_identification_upload:
self._on_identification_upload(False, None, error)
elif self._upload_to_cos(
elif self._upload_to_server(
csv_data, csv_filename, f"{the_folder}/ind_data"):
self.log("辨识数据上传成功")
if self._on_identification_upload:
@@ -334,25 +328,25 @@ class IdentificationManager:
self.log(f"辨识数据采集失败: {e}")
if self._on_identification_upload:
self._on_identification_upload(False, None, str(e))
finally:
self._identifying = False
# self.log("辨识结束")
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):
"""启动容积测量（在后台线程中运行）"""
# ---- 容积测量 ----
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
@@ -364,124 +358,124 @@ class IdentificationManager:
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_cos(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("测量结束")
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
def stop(self):
"""停止当前辨识/测量任务"""
self._identifying = False
ReinLoop/core/identification_config.py
+3 -3
View File
@@ -128,7 +128,7 @@ def parse_identification_config_csv(csv_text: str) -> dict:
def download_identification_config(timeout=20) -> dict:
"""Download the current customer's CSV config through the cloud function."""
"""Download the current customer's CSV config through the cloud server."""
import requests
from api import data_record_url, the_folder
@@ -140,10 +140,10 @@ def download_identification_config(timeout=20) -> dict:
response.raise_for_status()
result = response.json()
except Exception as exc:
raise ValueError(f"连接云端辨识配置服务失败: {exc}") from exc
raise ValueError(f"连接云服务器辨识配置服务失败: {exc}") from exc
if not result.get("success"):
raise ValueError(result.get("errMsg", "云端未返回辨识配置"))
raise ValueError(result.get("errMsg", "云服务器未返回辨识配置"))
try:
config_response = requests.get(result["url"], timeout=timeout)
config_response.raise_for_status()