150 lines
5.9 KiB
Python
150 lines
5.9 KiB
Python
"""Download CSV and validate the nine PRBS identification parameters."""
|
|
|
|
import csv
|
|
import io
|
|
import math
|
|
|
|
|
|
REQUIRED_FIELDS = {
|
|
"q_in_val", "dt", "n_order", "t_c", "levels",
|
|
"dead_area", "xa_full", "V_val", "repeat",
|
|
}
|
|
|
|
|
|
def validate_identification_config(config) -> dict:
|
|
"""Validate a parsed config mapping and normalize numeric values."""
|
|
if not isinstance(config, dict):
|
|
raise ValueError("辨识配置必须是参数映射")
|
|
actual = set(config)
|
|
if actual != REQUIRED_FIELDS:
|
|
missing = sorted(REQUIRED_FIELDS - actual)
|
|
extra = sorted(actual - REQUIRED_FIELDS)
|
|
raise ValueError(f"辨识配置字段错误，缺少={missing}，多余={extra}")
|
|
|
|
scalar_fields = {
|
|
"q_in_val", "dt", "t_c", "dead_area", "xa_full", "V_val"
|
|
}
|
|
for field in scalar_fields:
|
|
value = config[field]
|
|
if (isinstance(value, bool) or not isinstance(value, (int, float))
|
|
or not math.isfinite(float(value))):
|
|
raise ValueError(f"辨识参数 {field} 必须是有限数字")
|
|
|
|
for field in ("n_order", "repeat"):
|
|
value = config[field]
|
|
if isinstance(value, bool) or not isinstance(value, int):
|
|
raise ValueError(f"辨识参数 {field} 必须是整数")
|
|
|
|
levels = config["levels"]
|
|
if not isinstance(levels, list) or len(levels) < 2:
|
|
raise ValueError("levels 必须是至少包含 2 项的数组")
|
|
if len(levels) & (len(levels) - 1):
|
|
raise ValueError("levels 长度必须是 2 的整数次幂")
|
|
normalized_levels = []
|
|
for value in levels:
|
|
if (isinstance(value, bool) or not isinstance(value, (int, float))
|
|
or not math.isfinite(float(value)) or not 0 <= value <= 100):
|
|
raise ValueError("levels 中的开度必须是 0 到 100 的有限数字")
|
|
normalized_levels.append(float(value))
|
|
|
|
if config["q_in_val"] < 0:
|
|
raise ValueError("q_in_val 不能小于 0")
|
|
if config["dt"] <= 0 or config["t_c"] <= 0:
|
|
raise ValueError("dt 和 t_c 必须大于 0")
|
|
if config["t_c"] < config["dt"]:
|
|
raise ValueError("t_c 必须大于等于 dt，确保每个码元至少采样一次")
|
|
if config["n_order"] < 2:
|
|
raise ValueError("n_order 必须大于等于 2")
|
|
if config["repeat"] <= 0:
|
|
raise ValueError("repeat 必须是正整数")
|
|
if config["dead_area"] < 0 or config["xa_full"] <= config["dead_area"]:
|
|
raise ValueError("必须满足 0 <= dead_area < xa_full")
|
|
if config["xa_full"] < 1000:
|
|
raise ValueError("xa_full 不能小于前置行程扫描上限 1000")
|
|
if config["V_val"] <= 0:
|
|
raise ValueError("V_val 必须大于 0")
|
|
|
|
return {
|
|
"q_in_val": float(config["q_in_val"]),
|
|
"dt": float(config["dt"]),
|
|
"n_order": config["n_order"],
|
|
"t_c": float(config["t_c"]),
|
|
"levels": normalized_levels,
|
|
"dead_area": float(config["dead_area"]),
|
|
"xa_full": float(config["xa_full"]),
|
|
"V_val": float(config["V_val"]),
|
|
"repeat": config["repeat"],
|
|
}
|
|
|
|
|
|
def parse_identification_config_csv(csv_text: str) -> dict:
|
|
"""Parse a two-column CSV into the validated identification config.
|
|
|
|
The CSV must use ``parameter,value`` as its header. ``levels`` is one
|
|
quoted comma-separated value, for example ``"10,20,30,40"``.
|
|
"""
|
|
try:
|
|
reader = csv.DictReader(io.StringIO(csv_text))
|
|
fieldnames = [name.strip() for name in (reader.fieldnames or [])]
|
|
if fieldnames != ["parameter", "value"]:
|
|
raise ValueError("CSV 表头必须为 parameter,value")
|
|
|
|
raw = {}
|
|
for row in reader:
|
|
if None in row:
|
|
raise ValueError("CSV 每行只能包含 parameter 和 value 两列")
|
|
parameter = (row.get("parameter") or "").strip()
|
|
value = (row.get("value") or "").strip()
|
|
if not parameter:
|
|
raise ValueError("CSV 存在空参数名")
|
|
if parameter in raw:
|
|
raise ValueError(f"CSV 参数重复: {parameter}")
|
|
raw[parameter] = value
|
|
except csv.Error as exc:
|
|
raise ValueError(f"辨识配置 CSV 格式错误: {exc}") from exc
|
|
|
|
actual = set(raw)
|
|
if actual != REQUIRED_FIELDS:
|
|
missing = sorted(REQUIRED_FIELDS - actual)
|
|
extra = sorted(actual - REQUIRED_FIELDS)
|
|
raise ValueError(f"辨识配置字段错误，缺少={missing}，多余={extra}")
|
|
|
|
try:
|
|
levels = [float(value.strip()) for value in raw["levels"].split(",")]
|
|
config = {
|
|
"q_in_val": float(raw["q_in_val"]),
|
|
"dt": float(raw["dt"]),
|
|
"n_order": int(raw["n_order"]),
|
|
"t_c": float(raw["t_c"]),
|
|
"levels": levels,
|
|
"dead_area": float(raw["dead_area"]),
|
|
"xa_full": float(raw["xa_full"]),
|
|
"V_val": float(raw["V_val"]),
|
|
"repeat": int(raw["repeat"]),
|
|
}
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"辨识配置 CSV 参数值无效: {exc}") from exc
|
|
return validate_identification_config(config)
|
|
|
|
|
|
def download_identification_config(timeout=20) -> dict:
|
|
"""Download the current customer's CSV config through the cloud server."""
|
|
import requests
|
|
from api import device_post
|
|
|
|
try:
|
|
result = device_post({
|
|
"type": "getIdentificationConfig",
|
|
}, timeout=timeout)
|
|
except Exception as exc:
|
|
raise ValueError(f"连接云服务器辨识配置服务失败: {exc}") from exc
|
|
|
|
if not result.get("success"):
|
|
raise ValueError(result.get("errMsg", "云服务器未返回辨识配置"))
|
|
try:
|
|
config_response = requests.get(result["url"], timeout=timeout)
|
|
config_response.raise_for_status()
|
|
return parse_identification_config_csv(config_response.text)
|
|
except (KeyError, ValueError, requests.RequestException) as exc:
|
|
raise ValueError(f"下载或解析辨识配置 CSV 失败: {exc}") from exc
|