Merge pull request #33 from InfiniTensor/feat/read_mongodb_data

feat: dashboard reads data from MongoDB

Author: Chamberlain0w0

.pre-commit-config.yaml

@@ -9,6 +9,10 @@ repos:
     rev: 6.1.0
     hooks:
     -   id: flake8
-        # Use .flake8 for configuration (separate file to avoid TOML parsing issues)
-        # --exit-zero: show warnings but don't fail pre-commit
-        args: ['--config', '.flake8', '--exit-zero']
+        # Match GitHub CI: check for syntax errors and undefined names
+        # E9: Runtime errors (IndentationError, SyntaxError)
+        # F63: Assertion errors
+        # F7: Various errors
+        # F82: Undefined names
+        args: ['--select=E9,F63,F7,F82', '--show-source', '--statistics']
+

dashboard/app.py

@@ -13,7 +13,8 @@
 sys.path.append(str(project_root))
 
 from components.header import render_header
-from utils.data_loader import InfiniMetricsDataLoader, load_summary_file
+from utils.data_loader import InfiniMetricsDataLoader
+from common import show_data_source_info
 
 # Page configuration
 st.set_page_config(
@@ -28,6 +29,8 @@
     st.session_state.data_loader = InfiniMetricsDataLoader()
 if "selected_accelerators" not in st.session_state:
     st.session_state.selected_accelerators = []
+if "use_mongodb" not in st.session_state:
+    st.session_state.use_mongodb = False
 
 
 def main():
@@ -40,11 +43,34 @@ def main():
     with st.sidebar:
         st.markdown("## ⚙️ 设置")
 
+        # Data source selection
+        use_mongodb = st.toggle(
+            "使用 MongoDB",
+            value=st.session_state.use_mongodb,
+            help="切换到 MongoDB 数据源（需要 MongoDB 服务运行中）",
+        )
+
+        if use_mongodb != st.session_state.use_mongodb:
+            st.session_state.use_mongodb = use_mongodb
+            if use_mongodb:
+                st.session_state.data_loader = InfiniMetricsDataLoader(
+                    use_mongodb=True, fallback_to_files=True
+                )
+            else:
+                st.session_state.data_loader = InfiniMetricsDataLoader()
+
+        # Show current data source
+        show_data_source_info(style="sidebar")
+
+        st.markdown("---")
+
         results_dir = st.text_input(
-            "测试结果目录", value="./test_output", help="包含 JSON/CSV 测试结果的目录"
+            "测试结果目录", value="../output", help="包含 JSON/CSV 测试结果的目录"
         )
 
-        if results_dir != str(st.session_state.data_loader.results_dir):
+        if not use_mongodb and results_dir != str(
+            st.session_state.data_loader.results_dir
+        ):
             st.session_state.data_loader = InfiniMetricsDataLoader(results_dir)
 
         auto_refresh = st.toggle("自动刷新", value=False)
@@ -102,8 +128,9 @@ def render_dashboard(run_id_filter: str):
         ">
             <strong>InfiniMetrics Dashboard</strong> 用于统一展示
             <strong>通信（NCCL / 集合通信）</strong>、
-            <strong>推理（Direct / Service）</strong>、
-            <strong>算子（核心算子性能）</strong>
+            <strong>推理（直接推理 / 服务性能）</strong>、
+            <strong>算子（核心算子性能）</strong>、
+            <strong>硬件（内存带宽 / 缓存性能）</strong>
             等 AI 加速卡性能测试结果。
             <br/>
             测试框架输出 <code>JSON</code>（环境 / 配置 / 标量指标） +
@@ -230,10 +257,10 @@ def _latest(lst):
         st.dataframe(df, use_container_width=True, hide_index=True)
 
         # ========== Dispatcher summary ==========
-        summaries = load_summary_file()
+        summaries = st.session_state.data_loader.load_summaries()
 
         if not summaries:
-            st.info("No dispatcher_summary file found")
+            st.info("未找到 Dispatcher 汇总记录")
             return
 
         st.markdown("### 🧾 Dispatcher 汇总记录")

dashboard/common.py

@@ -21,8 +21,35 @@ def init_page(page_title: str, page_icon: str):
     # Page configuration
     st.set_page_config(page_title=page_title, page_icon=page_icon, layout="wide")
 
-    # Initialize DataLoader
+    # Initialize use_mongodb setting if not exists
+    if "use_mongodb" not in st.session_state:
+        st.session_state.use_mongodb = False
+
+    # Initialize DataLoader (respect MongoDB setting)
     if "data_loader" not in st.session_state:
         from utils.data_loader import InfiniMetricsDataLoader
 
-        st.session_state.data_loader = InfiniMetricsDataLoader()
+        st.session_state.data_loader = InfiniMetricsDataLoader(
+            use_mongodb=st.session_state.use_mongodb,
+            fallback_to_files=True,
+        )
+
+
+def show_data_source_info(style: str = "caption"):
+    """
+    Display current data source info (MongoDB or file system).
+
+    Args:
+        style: Display style - "caption" for pages, "sidebar" for main app sidebar
+    """
+    dl = st.session_state.data_loader
+    if dl.source_type == "mongodb":
+        if style == "sidebar":
+            st.success("🟢 数据源: MongoDB")
+        else:
+            st.caption("数据源: MongoDB")
+    else:
+        if style == "sidebar":
+            st.info(f"📁 数据源: 文件系统 ({dl.results_dir})")
+        else:
+            st.caption(f"数据源: 文件系统 ({dl.results_dir})")

dashboard/pages/communication.py

@@ -4,7 +4,7 @@
 import streamlit as st
 import pandas as pd
 
-from common import init_page
+from common import init_page, show_data_source_info
 from components.header import render_header
 from utils.data_loader import get_friendly_size
 from utils.metrics import extract_core_metrics
@@ -17,14 +17,16 @@
     create_summary_table_infer,
 )
 
-init_page("推理测试分析 | InfiniMetrics", "🔗")
+init_page("通信测试分析 | InfiniMetrics", "🔗")
 
 
 def main():
     """Main function for communication tests page."""
     render_header()
     st.markdown("## 🔗 通信性能测试分析")
 
+    show_data_source_info()
+
     try:
         # Load communication test results
         comm_runs = st.session_state.data_loader.list_test_runs("comm")
@@ -117,7 +119,9 @@ def main():
         for name in selected_indices:
             idx = run_options[name]
             run_info = filtered_runs[idx]
-            result = st.session_state.data_loader.load_test_result(run_info["path"])
+            # Use path for file source, run_id for MongoDB
+            identifier = run_info.get("path") or run_info.get("run_id")
+            result = st.session_state.data_loader.load_test_result(identifier)
             run_info["data"] = result
             selected_runs.append(run_info)
 

dashboard/pages/inference.py

@@ -4,9 +4,8 @@
 import streamlit as st
 import pandas as pd
 
-from common import init_page
+from common import init_page, show_data_source_info
 from components.header import render_header
-from utils.data_loader import InfiniMetricsDataLoader, get_friendly_size
 from utils.visualizations import (
     plot_timeseries_auto,
     create_summary_table_infer,
@@ -19,8 +18,9 @@ def main():
     render_header()
     st.markdown("## 🚀 推理性能测试分析")
 
-    dl = st.session_state.data_loader
-    runs = dl.list_test_runs("infer")
+    show_data_source_info()
+
+    runs = st.session_state.data_loader.list_test_runs("infer")
 
     if not runs:
         st.info("未找到推理测试结果（testcase 需以 infer.* 开头）。")
@@ -91,7 +91,9 @@ def _mode_of(r):
     selected_runs = []
     for k in selected:
         ri = filtered[options[k]]
-        data = dl.load_test_result(ri["path"])
+        # Use path for file source, run_id for MongoDB
+        identifier = ri.get("path") or ri.get("run_id")
+        data = st.session_state.data_loader.load_test_result(identifier)
         ri = dict(ri)
         ri["data"] = data
         selected_runs.append(ri)

dashboard/pages/operator.py

@@ -4,7 +4,7 @@
 import streamlit as st
 import pandas as pd
 
-from common import init_page
+from common import init_page, show_data_source_info
 from components.header import render_header
 from utils.visualizations import (
     create_summary_table_ops,
@@ -18,24 +18,18 @@ def main():
     render_header()
     st.markdown("## ⚡ 算子测试分析")
 
-    dl = st.session_state.data_loader
+    show_data_source_info()
 
-    runs = dl.list_test_runs()  # Load all test runs first
-    # Identify operator runs by checking "operators" in path or testcase starting with operator/ops
+    runs = st.session_state.data_loader.list_test_runs()
+    # Identify operator runs by testcase starting with operator/ops
     ops_runs = []
     for r in runs:
-        p = str(r.get("path", ""))
         tc = (r.get("testcase") or "").lower()
-        if (
-            ("/operators/" in p.replace("\\", "/"))
-            or tc.startswith("operator")
-            or tc.startswith("operators")
-            or tc.startswith("ops")
-        ):
+        if tc.startswith(("operator", "operators", "ops")):
             ops_runs.append(r)
 
     if not ops_runs:
-        st.info("未找到算子测试结果（请确认 JSON 在 test_output/operators/ 下）。")
+        st.info("未找到算子测试结果（请确认 JSON 在 output/operators 目录下）。")
         return
 
     with st.sidebar:
@@ -62,7 +56,9 @@ def main():
     selected_runs = []
     for k in selected:
         ri = filtered[options[k]]
-        data = dl.load_test_result(ri["path"])
+        # Use path for file source, run_id for MongoDB
+        identifier = ri.get("path") or ri.get("run_id")
+        data = st.session_state.data_loader.load_test_result(identifier)
         ri = dict(ri)
         ri["data"] = data
         selected_runs.append(ri)