wangsli/acoustic_algo_scheduler
1
0
派生 0
代码 工单 合并请求 工作流 软件包 项目 发布 百科 活动

feat: v3.5 - Multi-language acoustic algorithm scheduler

浏览代码 
- Core: Python/C++/MATLAB tri-protocol bridge architecture
- Validation: 5-type real-time field, type, dimension, completeness, and compatibility checking
- UI: Premium Deep Sea theme with visual pipeline preview
- Modules: Sea Trial Exclusive + Exception Hub + Algorithm Management
- Tests: Full coverage with 5 automated test cases
这个提交包含在：
wangsli
2026-01-21 07:41:12 +08:00
当前提交 11a2d4a4b9
共修改 25 个文件，包含 2528 行新增和 0 行删除
下载 Patch 文件 下载 Diff 文件 展开所有文件 折叠所有文件
main.py
+624
查看文件
@@ -0,0 +1,624 @@
import streamlit as st
import numpy as np
import sys
import os
# Ensure src is in path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from src.models import WaterSoundData
from src.bridges import PythonBridge
from src.scheduler import CoreScheduler
from src.algo_loader import load_algo_func
st.set_page_config(page_title="水声算法全协议集成与调度平台", layout="wide")
from src.algo_manager import AlgorithmManager
from src.template_manager import TemplateManager
ALGO_DB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data/algorithms.json")
TEMPLATE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data/templates")
def set_design_system():
st.markdown("""
<style>
* {
/* Keep offline-friendly font stack (Inter if installed). */
font-family: Inter, ui-sans-serif, system-ui, -apple-system, "Segoe UI", Roboto, Helvetica, Arial, "Noto Sans", "Liberation Sans", sans-serif;
}
/* Sidebar Styling */
[data-testid="stSidebar"] {
background-image: linear-gradient(180deg, #023e8a 0%, #03045e 100%);
color: white;
}
[data-testid="stSidebar"] * {
color: white !important;
}
/* Metric Cards */
[data-testid="stMetricValue"] {
font-size: 2.5rem !important;
font-weight: 700 !important;
color: #00b4d8 !important;
}
/* Glassmorphism Containers (Approximation in Streamlit) */
.stExpander {
border: 1px solid rgba(255, 255, 255, 0.1) !important;
border-radius: 12px !important;
box-shadow: 0 4px 6px -1px rgba(0, 0, 0, 0.1), 0 2px 4px -1px rgba(0, 0, 0, 0.06) !important;
background: rgba(255, 255, 255, 0.02) !important;
}
/* Custom Badges */
.algo-badge {
padding: 4px 12px;
border-radius: 20px;
font-size: 0.8rem;
font-weight: 600;
color: white;
display: inline-block;
margin-bottom: 10px;
}
.badge-python { background-color: #3776AB; }
.badge-cpp { background-color: #7C3AED; }
.badge-matlab { background-color: #ED8B00; }
/* Hero Section */
.hero {
padding: 2rem;
border-radius: 15px;
background: linear-gradient(90deg, #0077b6 0%, #0096c7 100%);
color: white;
margin-bottom: 2rem;
box-shadow: 0 10px 15px -3px rgba(0, 0, 0, 0.1);
}
/* Buttons */
.stButton>button {
border-radius: 8px !important;
border: none !important;
background-color: #0077b6 !important;
color: white !important;
transition: all 0.3s ease !important;
}
.stButton>button:hover {
background-color: #00b4d8 !important;
transform: translateY(-2px);
box-shadow: 0 4px 6px -1px rgba(0, 0, 0, 0.2);
}
</style>
""", unsafe_allow_html=True)
def main():
set_design_system()
st.sidebar.markdown("<h1 style='text-align: center; color: white;'>🌊 HYD-SCHED</h1>", unsafe_allow_html=True)
st.sidebar.markdown("<p style='text-align: center; opacity: 0.8;'>海音达算法智能调度平台 v3.5</p>", unsafe_allow_html=True)
st.sidebar.divider()
app_mode = st.sidebar.selectbox("功能菜单",
["首页", "算法库管理", "接口配置", "算法调度", "海试专属", "异常中心"])
if app_mode == "首页":
show_home()
elif app_mode == "算法库管理":
show_algo_library()
elif app_mode == "接口配置":
show_interface_config()
elif app_mode == "算法调度":
show_scheduling()
elif app_mode == "海试专属":
show_sea_trial()
elif app_mode == "异常中心":
show_exceptions()
else:
st.info(f"模块 '{app_mode}' 正在开发中...")
# Keep sidebar spacing without relying on non-Streamlit APIs.
st.sidebar.markdown("<div style='height: 100px;'></div>", unsafe_allow_html=True)
st.sidebar.markdown("""
<div style="position: fixed; bottom: 20px; width: 200px; padding: 10px; border-top: 1px solid rgba(255,255,255,0.1);">
<p style="font-size: 0.7rem; opacity: 0.6; margin: 0;">平台状态: 🟢 已就绪</p>
<p style="font-size: 0.7rem; opacity: 0.6; margin: 0;">环境检测: ✅ 2.0.4</p>
</div>
""", unsafe_allow_html=True)
def show_home():
st.markdown("""
<div class="hero">
<h1>🌊 海音达智能调度处理系统</h1>
<p style="font-size: 1.2rem; opacity: 0.9;">
全协议集成 · 五维实时校验 · 跨语言桥接驱动
</p>
<hr style="border-color: rgba(255,255,255,0.2);">
<div style="display: flex; gap: 20px;">
<span>✅ Python Native</span>
<span>⚡ C++ Performance</span>
<span>📊 MATLAB Legacy</span>
</div>
</div>
""", unsafe_allow_html=True)
col1, col2 = st.columns(2)
with col1:
st.subheader("🚀 核心能力")
st.markdown("""
- **全协议兼容**: 支持 Python/C++/MATLAB/Octave 原生算法无缝调用。
- **工业级校验**: 5维字段、类型、空间、完整性与环境兼容性实时监测。
- **动态资产化**: 算法元数据驱动，支持 I/O 映射模板化管理。
""")
with col2:
st.subheader("⚓ 应用场景")
st.markdown("""
- **海试一键巡检**: 自动化质量巡检链路，实时出具测评分析报表。
- **多源异构调度**: 将不同语言编写的声呐算法快速编排入链。
- **异常深度追溯**: 聚合校验、执行与环境日志，一键打包现场问题。
""")
def show_algo_library():
st.title("📚 算法库管理")
from src.env_detect import collect_env_info, get_available_algo_types
manager = AlgorithmManager(ALGO_DB)
algos = manager.load_algorithms()
env = collect_env_info()
available_types = get_available_algo_types()
# 环境状态概览
with st.expander("🖥️ 当前环境能力", expanded=False):
col1, col2, col3 = st.columns(3)
col1.metric("Python", "✅ 可用")
col2.metric("MATLAB/Octave", "✅ 可用" if "matlab" in available_types else "❌ 不可用")
col3.metric("C++ 编译器", "✅ 可用" if "cpp" in available_types else "❌ 不可用")
st.json(env)
st.subheader("已注册算法")
# 筛选器
filter_type = st.selectbox("按类型筛选", ["全部"] + available_types)
for algo in algos:
algo_type = algo.get("type", "python")
if filter_type != "全部" and algo_type != filter_type:
continue
# 判断是否可用
is_available = algo_type in available_types
status_icon = "🔵" if is_available else "⚪"
badge_class = f"badge-{algo_type}"
with st.expander(f"{status_icon} {algo['name']} (v{algo['version']})"):
st.markdown(f'<div class="algo-badge {badge_class}">{algo_type.upper()}</div>', unsafe_allow_html=True)
st.write(f"**描述**: {algo.get('description', 'N/A')}")
if algo_type == "python":
st.write(f"**入口**: `{algo.get('module', 'N/A')}.{algo.get('function', 'N/A')}`")
elif algo_type == "cpp":
st.write(f"**入口**: `{algo.get('lib_path', 'N/A')}::{algo.get('function', 'N/A')}`")
elif algo_type == "matlab":
st.write(f"**入口**: `{algo.get('m_path', 'N/A')}::{algo.get('function', 'N/A')}`")
else:
st.write(f"**入口**: `{algo.get('function', 'N/A')}`")
st.write("**输入参数**:")
for k, v in (algo.get("inputs") or {}).items():
st.write(f" - `{k}`: {v.get('type', '?')} → `{v.get('field', k)}`")
st.write("**输出参数**:")
for k, v in (algo.get("outputs") or {}).items():
st.write(f" - `{k}`: {v.get('type', '?')}")
if not is_available:
st.warning(f"⚠️ 当前环境不支持 {algo_type} 类型算法，请安装相应依赖。")
def show_interface_config():
st.title("🔧 接口配置")
template_manager = TemplateManager(TEMPLATE_DIR)
manager = AlgorithmManager(ALGO_DB)
algos = manager.load_algorithms()
st.subheader("创建/编辑接口模板")
algo_names = [a["name"] for a in algos]
selected_algo = st.selectbox("选择算法", algo_names) if algo_names else None
if selected_algo:
algo = manager.get_algorithm(selected_algo)
st.write("**输入映射配置**")
wsd_fields = ["raw_data", "sample_rate", "collect_time", "sea_env",
"clean_data", "denoised_data", "stft_data", "feat_data",
"final_result", "core_index", "vis_data"]
input_mapping = {}
for param, spec in (algo.get("inputs") or {}).items():
default_field = (spec or {}).get("field", param)
selected_field = st.selectbox(
f"`{param}` 映射到",
wsd_fields,
index=wsd_fields.index(default_field) if default_field in wsd_fields else 0,
key=f"in_{param}"
)
input_mapping[param] = selected_field
st.write("**输出映射配置**")
output_mapping = {}
for field, spec in (algo.get("outputs") or {}).items():
idx = (spec or {}).get("index", 0)
selected_field = st.selectbox(
f"输出 `{field}` 写入",
wsd_fields,
index=wsd_fields.index(field) if field in wsd_fields else 0,
key=f"out_{field}"
)
output_mapping[str(idx)] = selected_field
st.write("**预处理算子（可选）**")
preprocessing = []
if st.checkbox("启用归一化 (normalize)"):
preprocessing.append({"op": "normalize", "field": list(input_mapping.values())[0] if input_mapping else "raw_data", "params": {"min": 0, "max": 1}})
if st.checkbox("启用缩放 (scale)"):
factor = st.number_input("缩放因子", value=1.0, step=0.1)
preprocessing.append({"op": "scale", "field": list(input_mapping.values())[0] if input_mapping else "raw_data", "params": {"factor": factor}})
template_name = st.text_input("模板名称", value=selected_algo)
if st.button("保存模板"):
template_data = {
"algorithm": selected_algo,
"input_mapping": input_mapping,
"output_mapping": output_mapping,
"preprocessing": preprocessing,
}
template_manager.save_template(template_name, template_data)
st.success(f"✅ 模板 '{template_name}' 已保存！")
st.divider()
st.subheader("已保存模板")
templates = template_manager.list_templates()
if templates:
for tpl_name in templates:
with st.expander(f"📄 {tpl_name}"):
tpl = template_manager.load_template(tpl_name)
st.json(tpl)
else:
st.info("暂无已保存模板。")
def show_scheduling():
st.title("⚙️ 算法链路编排与调度")
manager = AlgorithmManager(ALGO_DB)
available_algos = manager.load_algorithms()
template_manager = TemplateManager(TEMPLATE_DIR)
if 'pipeline' not in st.session_state:
st.session_state.pipeline = []
col1, col2 = st.columns([1, 1])
# Visual Pipeline flow
if st.session_state.pipeline:
st.write("**📦 运行流水线预览**")
flow_html = '<div style="display: flex; align-items: center; gap: 10px; overflow-x: auto; padding: 10px;">'
for i, algo in enumerate(st.session_state.pipeline):
algo_type = algo.get("type", "python")
badge_class = f"badge-{algo_type}"
flow_html += f'''
<div style="min-width: 120px; padding: 10px; border: 1px solid #0077b6; border-radius: 8px; text-align: center; background: rgba(0,119,182,0.05);">
<div style="font-size: 0.7rem; opacity: 0.7;">STEP {i+1}</div>
<div style="font-weight: 600; font-size: 0.9rem;">{algo["name"]}</div>
<div class="algo-badge {badge_class}" style="margin: 5px 0 0 0; font-size: 0.6rem;">{algo_type.upper()}</div>
</div>
'''
if i < len(st.session_state.pipeline) - 1:
flow_html += '<div style="font-size: 1.5rem; color: #0077b6;">➡</div>'
flow_html += '</div>'
st.markdown(flow_html, unsafe_allow_html=True)
st.divider()
with col1:
st.subheader("1. 链路编排")
if 'pipeline' not in st.session_state:
st.session_state.pipeline = []
algo_names = sorted({a["name"] for a in available_algos}) if available_algos else []
if not algo_names:
st.warning("算法库为空：请先在 algorithms.json 注册算法元数据。")
return
selected_algo_name = st.selectbox("选择算法添加至链路", algo_names)
if st.button("添加至链路"):
algo_meta = manager.get_algorithm(selected_algo_name)
st.session_state.pipeline.append(algo_meta)
st.rerun()
# 链路列表与移动操作
for i, algo in enumerate(st.session_state.pipeline):
cols = st.columns([0.6, 0.1, 0.1, 0.2])
cols[0].write(f"{i+1}. **{algo['name']}** (v{algo['version']})")
if cols[1].button("↑", key=f"up_{i}") and i > 0:
st.session_state.pipeline[i], st.session_state.pipeline[i-1] = st.session_state.pipeline[i-1], st.session_state.pipeline[i]
st.rerun()
if cols[2].button("↓", key=f"down_{i}") and i < len(st.session_state.pipeline) - 1:
st.session_state.pipeline[i], st.session_state.pipeline[i+1] = st.session_state.pipeline[i+1], st.session_state.pipeline[i]
st.rerun()
if cols[3].button("移除", key=f"rm_{i}"):
st.session_state.pipeline.pop(i)
st.rerun()
with col2:
st.subheader("2. 数据与校验")
sample_rate = st.number_input("采样率 (Hz)", min_value=1.0, value=1000.0, step=100.0)
uploaded_file = st.file_uploader("上传待处理数据 (.npy)", type=["npy"])
auto_apply_template = st.checkbox("自动应用同名接口模板（若存在）", value=True)
if uploaded_file and st.session_state.pipeline:
data = np.load(uploaded_file)
wsd = WaterSoundData(raw_data=data, sample_rate=float(sample_rate))
# 自动映射 (Phase 2 简化版：直接使用元数据中的映射)
scheduler = CoreScheduler()
bridges = []
for algo in st.session_state.pipeline:
tpl = None
if auto_apply_template:
try:
tpl = template_manager.load_template(algo["name"])
except FileNotFoundError:
tpl = None
if tpl:
in_map = tpl.get("input_mapping") or {}
out_map = tpl.get("output_mapping") or {}
preprocessing = tpl.get("preprocessing") or []
else:
in_map = {
param: (spec or {}).get("field") or param
for param, spec in (algo.get("inputs") or {}).items()
}
out_map = {
str((spec or {}).get("index", 0)): field
for field, spec in (algo.get("outputs") or {}).items()
}
preprocessing = []
from src.bridges import PythonBridge, CppBridge
algo_type = algo.get("type", "python")
if algo_type == "python":
try:
algo_func = load_algo_func(algo.get("module", ""), algo.get("function", ""))
except Exception as e:
st.error(f"算法加载失败：{algo.get('name', '<unknown>')} - {e}")
st.stop()
bridge = PythonBridge(
algo["name"],
algo_func,
in_map,
out_map,
preprocessing=preprocessing,
requirements=algo.get("inputs") or {},
output_specs=algo.get("outputs") or {},
algo_type=algo_type,
)
elif algo_type == "cpp":
bridge = CppBridge(
algo["name"],
algo.get("lib_path", ""),
algo.get("function", ""),
in_map,
out_map,
preprocessing=preprocessing,
requirements=algo.get("inputs") or {},
output_specs=algo.get("outputs") or {},
)
elif algo_type == "matlab":
from src.bridges import MatlabBridge
bridge = MatlabBridge(
algo["name"],
algo.get("m_path", ""),
algo.get("function", ""),
in_map,
out_map,
preprocessing=preprocessing,
requirements=algo.get("inputs") or {},
output_specs=algo.get("outputs") or {},
)
else:
st.warning(f"暂不支持的算法类型: {algo_type}")
continue
bridges.append(bridge)
scheduler.set_flow(bridges)
if st.button("🔍 链路全局校验"):
issues = scheduler.validate_flow(wsd)
st.session_state.last_validation_issues = issues
if not issues:
st.success("✅ 校验通过：所有字段、类型与维度匹配。")
else:
for issue in issues:
st.error(f"❌ {issue['algo']}: {issue['message']}")
st.info(issue['suggestion'])
if st.button("🚀 一键执行调度"):
issues = scheduler.validate_flow(wsd)
st.session_state.last_validation_issues = issues
if issues:
st.error("⚠️ 校验未通过：已拦截执行，请先修复问题。")
for issue in issues:
st.error(f"❌ {issue['algo']}: {issue['message']}")
st.info(issue['suggestion'])
st.stop()
with st.spinner("流水线执行中..."):
result_wsd = scheduler.execute(wsd)
st.session_state.last_run = {
"status": result_wsd.status,
"error_info": result_wsd.error_info,
"log": result_wsd.log,
}
if result_wsd.status == "finished":
st.success("🎉 执行成功")
else:
st.error(f"⚠️ 执行中断: {result_wsd.error_info}")
with st.expander("实时执行日志"):
st.table(result_wsd.log)
def show_sea_trial():
st.title("⛴️ 海试专属一键检测")
st.info("内置『一键质量检测』模板：去直流 -> 带通滤波 -> 质量巡检 -> SNR估计")
sample_rate = st.number_input("采样率 (Hz)", min_value=1.0, value=1000.0, step=100.0)
uploaded_file = st.file_uploader("上传海试原始数据 (.npy)", type=["npy"])
if uploaded_file:
data = np.load(uploaded_file)
wsd = WaterSoundData(raw_data=data, sample_rate=float(sample_rate))
# 预设模板执行
if st.button("开始一键巡检"):
# 这里的逻辑可以抽象为从 TemplateManager 加载一个名叫 "sea_trial_default" 的模板
# 这里硬编码模拟一镜到底
try:
# 1. 加载函数
from src.sample_algorithms import dc_removal_and_bandpass, quality_check, snr_estimation
# 2. 构造链路
b1 = PythonBridge("去直流+带通", dc_removal_and_bandpass,
{"raw_data": "raw_data", "fs": "sample_rate"}, {"0": "clean_data"})
b2 = PythonBridge("质量巡检", quality_check,
{"raw_data": "raw_data"}, {"0": "core_index"})
b3 = PythonBridge("SNR估计", snr_estimation,
{"raw_data": "clean_data"}, {"0": "snr_value"})
scheduler = CoreScheduler()
scheduler.set_flow([b1, b2, b3])
with st.spinner("正在巡检中..."):
result = scheduler.execute(wsd)
st.session_state.last_run = {
"status": result.status,
"error_info": result.error_info,
"log": result.log,
}
if result.status == "finished":
st.success("✅ 巡检完成")
# 展示简报 (使用更高级的展示方式)
snr_value = getattr(result, "snr_value", None)
st.markdown("""
<div style="background: rgba(0,119,182,0.05); padding: 20px; border-radius: 12px; border: 1px solid rgba(0,119,182,0.2); margin-bottom: 20px;">
<h3 style="margin-top: 0; color: #0077b6;">📊 巡检简报</h3>
</div>
""", unsafe_allow_html=True)
c1, c2, c3 = st.columns(3)
with c1:
snr_str = f"{snr_value:.2f} dB" if isinstance(snr_value, (int, float)) and np.isfinite(snr_value) else "N/A"
st.metric("信噪比 (SNR)", snr_str)
with c2:
st.metric("最大幅值 (Peak)", f"{result.core_index.get('max_abs_val',0):.2f}")
with c3:
st.metric("质量评估分", f"{result.core_index.get('quality_score',0):.1f}")
# 导出报告
import csv
import io
timestamp = result.log[0].get("timestamp") if result.log else ""
row = {
"timestamp": timestamp,
"sample_rate_hz": float(sample_rate),
"snr_db": snr_value,
"saturation": result.core_index.get("saturation"),
"dropouts": result.core_index.get("dropouts"),
"max_abs_val": result.core_index.get("max_abs_val"),
"quality_score": result.core_index.get("quality_score"),
}
output = io.StringIO()
writer = csv.DictWriter(output, fieldnames=list(row.keys()))
writer.writeheader()
writer.writerow(row)
st.download_button("下载 CSV 报告", output.getvalue(), "sea_trial_report.csv", "text/csv")
else:
st.error(f"巡检中断：{result.error_info}")
except Exception as e:
st.error(f"系统错误：{e}")
def show_exceptions():
st.markdown("""
<div style="background: rgba(255,107,107,0.1); padding: 1.5rem; border-left: 5px solid #ff6b6b; border-radius: 8px; margin-bottom: 2rem;">
<h2 style="margin:0; color: #ff6b6b;">🚑 异常中心 (Exception Hub)</h2>
<p style="margin:0; opacity: 0.8;">聚合全链路校验与执行错误，支持一键导出排障包。</p>
</div>
""", unsafe_allow_html=True)
issues = st.session_state.get("last_validation_issues")
last_run = st.session_state.get("last_run")
if issues is None and last_run is None:
st.info("💡 暂无异常记录。所有组件目前运行平稳。")
return
col1, col2 = st.columns(2)
with col1:
st.subheader("🛡️ 静态校验追踪")
if issues is not None:
if not issues:
st.success("✅ 链路校验完美 (Zero Issues)")
else:
for issue in issues:
with st.container():
st.markdown(f"""
<div style="padding: 10px; border: 1px solid rgba(255,107,107,0.3); border-radius: 8px; margin-bottom: 10px;">
<strong style="color: #ff6b6b;">❌ {issue.get('algo', 'Global')}</strong><br/>
<span style="font-size: 0.9rem;">{issue.get('message', '')}</span>
<div style="background: rgba(0,0,0,0.05); padding: 5px; margin-top: 5px; font-size: 0.8rem; border-radius: 4px;">
💡 修复建议: {issue.get('suggestion', '')}
</div>
</div>
""", unsafe_allow_html=True)
with col2:
st.subheader("⚙️ 动态执行追踪")
if last_run is not None:
status = last_run.get('status')
color = "#51cf66" if status == "finished" else "#ff6b6b"
st.markdown(f'状态: <span style="color: {color}; font-weight: bold;">{status.upper()}</span>', unsafe_allow_html=True)
if last_run.get("error_info"):
st.error(f"运行时拦截: {last_run.get('error_info')}")
with st.expander("📝 详细追踪日志"):
st.table(last_run.get("log") or [])
st.divider()
c1, c2 = st.columns(2)
with c1:
with st.expander("🖥️ 环境自检快照"):
from src.env_detect import collect_env_info
st.json(collect_env_info())
with c2:
st.write("📦 **故障包存档**")
import json
payload = {
"validation_issues": issues,
"last_run": last_run,
}
st.download_button(
"下载排障 JSON 包",
json.dumps(payload, ensure_ascii=False, indent=2),
file_name="issue_bundle.json",
mime="application/json",
use_container_width=True
)
if __name__ == "__main__":
main()