添加 PRCXI 耗材管理 Web 应用 (labware_manager)

CLAUDE.md

@@ -24,17 +24,28 @@ unilab --skip_env_check                   # skip auto-install of dependencies
 unilab --visual rviz|web|disable          # visualization mode
 unilab --is_slave                         # run as slave node
 unilab --restart_mode                     # auto-restart on config changes (supervisor/child process)
+unilab --external_devices_only            # only load external device packages
+unilab --extra_resource                   # load extra lab_ prefixed labware resources
 
 # Workflow upload subcommand
 unilab workflow_upload -f <workflow.json> -n <name> --tags tag1 tag2
 
+# Labware Manager (standalone web UI for PRCXI labware CRUD, port 8010)
+python -m unilabos.labware_manager
+
 # Tests
 pytest tests/                              # all tests
 pytest tests/resources/test_resourcetreeset.py  # single test file
 pytest tests/resources/test_resourcetreeset.py::TestClassName::test_method  # single test
 
 # CI check (matches .github/workflows/ci-check.yml)
 python -m unilabos --check_mode --skip_env_check
+
+# If registry YAML/Python files changed, regenerate before committing:
+python -m unilabos --complete_registry
+
+# Documentation build
+cd docs && python -m sphinx -b html . _build/html -v
 ```
 
 ## Architecture
@@ -45,7 +56,22 @@ python -m unilabos --check_mode --skip_env_check
 
 ### Core Layers
 
-**Registry** (`unilabos/registry/`): Singleton `Registry` class discovers and catalogs all device types, resource types, and communication devices. Two registration mechanisms: YAML definitions in `registry/devices/*.yaml` and Python decorators (`@device`, `@action`, `@resource` in `registry/decorators.py`). AST scanning discovers decorated classes without importing them. Class paths resolved to Python classes via `utils/import_manager.py`.
+**Registry** (`unilabos/registry/`): Singleton `Registry` class discovers and catalogs all device types, resource types, and communication devices. Two registration mechanisms:
+1. **YAML definitions** in `registry/devices/*.yaml` and `registry/resources/` (backward-compatible)
+2. **Python decorators** (`@device`, `@action`, `@resource` in `registry/decorators.py`) — preferred for new code
+
+AST scanning (`ast_registry_scanner.py`) discovers decorated classes without importing them, so `--check_mode` works without hardware dependencies. Class paths resolved to Python classes at runtime via `utils/import_manager.py`.
+
+Decorator usage pattern:
+```python
+from unilabos.registry.decorators import device, action, resource
+from unilabos.registry.decorators import InputHandle, OutputHandle, HardwareInterface
+
+@device(id="my_device.v1", category=["category_name"], handles=[...])
+class MyDevice:
+    @action(action_type=SomeActionType)
+    def do_something(self): ...
+```
 
 **Resource Tracking** (`unilabos/resources/resource_tracker.py`): Pydantic-based `ResourceDict` → `ResourceDictInstance` → `ResourceTreeSet` hierarchy. `ResourceTreeSet` is the canonical in-memory representation of all devices and resources. Graph I/O in `resources/graphio.py` reads JSON/GraphML device topology files into `nx.Graph` + `ResourceTreeSet`.
 
@@ -61,6 +87,8 @@ python -m unilabos --check_mode --skip_env_check
 
 **Web/API** (`unilabos/app/web/`): FastAPI server with REST API (`api.py`), Jinja2 templates (`pages.py`), HTTP client (`client.py`). Default port 8002.
 
+**Labware Manager** (`unilabos/labware_manager/`): Standalone FastAPI web app (port 8010) for PRCXI labware CRUD. Pydantic models in `models.py`, JSON database in `labware_db.json`. Supports importing from existing Python/YAML (`importer.py`), code generation (`codegen.py`), and YAML generation (`yaml_gen.py`). Web UI with SVG visualization (`static/labware_viz.js`), dynamic form handling (`static/form_handler.js`), and Jinja2 templates.
+
 ### Configuration System
 
 - **Config classes** in `unilabos/config/config.py`: `BasicConfig`, `WSConfig`, `HTTPConfig`, `ROSConfig` — class-level attributes, loaded from Python `.py` config files (see `config/example_config.py`)
@@ -88,6 +116,7 @@ Example device graphs and experiment configs are in `unilabos/test/experiments/`
 - CLI argument dashes auto-converted to underscores for consistency
 - No linter/formatter configuration enforced (no ruff, black, flake8, mypy configs)
 - Documentation built with Sphinx (Chinese language, `sphinx_rtd_theme`, `myst_parser`)
+- CI runs on Windows (`windows-latest`); if registry files change, run `python -m unilabos --complete_registry` locally before committing
 
 ## Licensing
 

unilabos/labware_manager/__init__.py

@@ -0,0 +1 @@
+# PRCXI 耗材管理 Web 应用

unilabos/labware_manager/__main__.py

@@ -0,0 +1,4 @@
+"""启动入口: python -m unilabos.labware_manager"""
+from unilabos.labware_manager.app import main
+
+main()

unilabos/labware_manager/app.py

@@ -0,0 +1,196 @@
+"""FastAPI 应用 + CRUD API + 启动入口。
+
+用法: python -m unilabos.labware_manager.app
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import List, Optional
+
+from fastapi import FastAPI, HTTPException, Query, Request
+from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.templating import Jinja2Templates
+
+from unilabos.labware_manager.models import LabwareDB, LabwareItem
+
+_HERE = Path(__file__).resolve().parent
+_DB_PATH = _HERE / "labware_db.json"
+
+app = FastAPI(title="PRCXI 耗材管理", version="1.0")
+
+# 静态文件 + 模板
+app.mount("/static", StaticFiles(directory=str(_HERE / "static")), name="static")
+templates = Jinja2Templates(directory=str(_HERE / "templates"))
+
+
+# ---------- DB 读写 ----------
+
+def _load_db() -> LabwareDB:
+    if not _DB_PATH.exists():
+        return LabwareDB()
+    with open(_DB_PATH, "r", encoding="utf-8") as f:
+        return LabwareDB(**json.load(f))
+
+
+def _save_db(db: LabwareDB) -> None:
+    with open(_DB_PATH, "w", encoding="utf-8") as f:
+        json.dump(db.model_dump(), f, ensure_ascii=False, indent=2)
+
+
+# ---------- 页面路由 ----------
+
+@app.get("/", response_class=HTMLResponse)
+async def index_page(request: Request):
+    db = _load_db()
+    # 按 type 分组
+    groups = {}
+    for item in db.items:
+        groups.setdefault(item.type, []).append(item)
+    return templates.TemplateResponse("index.html", {
+        "request": request,
+        "groups": groups,
+        "total": len(db.items),
+    })
+
+
+@app.get("/labware/new", response_class=HTMLResponse)
+async def new_page(request: Request, type: str = "plate"):
+    return templates.TemplateResponse("edit.html", {
+        "request": request,
+        "item": None,
+        "labware_type": type,
+        "is_new": True,
+    })
+
+
+@app.get("/labware/{item_id}", response_class=HTMLResponse)
+async def detail_page(request: Request, item_id: str):
+    db = _load_db()
+    item = _find_item(db, item_id)
+    if not item:
+        raise HTTPException(404, "耗材不存在")
+    return templates.TemplateResponse("detail.html", {
+        "request": request,
+        "item": item,
+    })
+
+
+@app.get("/labware/{item_id}/edit", response_class=HTMLResponse)
+async def edit_page(request: Request, item_id: str):
+    db = _load_db()
+    item = _find_item(db, item_id)
+    if not item:
+        raise HTTPException(404, "耗材不存在")
+    return templates.TemplateResponse("edit.html", {
+        "request": request,
+        "item": item,
+        "labware_type": item.type,
+        "is_new": False,
+    })
+
+
+# ---------- API 端点 ----------
+
+@app.get("/api/labware")
+async def api_list_labware():
+    db = _load_db()
+    return {"items": [item.model_dump() for item in db.items]}
+
+
+@app.post("/api/labware")
+async def api_create_labware(request: Request):
+    data = await request.json()
+    db = _load_db()
+    item = LabwareItem(**data)
+    # 确保 id 唯一
+    existing_ids = {it.id for it in db.items}
+    while item.id in existing_ids:
+        import uuid
+        item.id = uuid.uuid4().hex[:8]
+    db.items.append(item)
+    _save_db(db)
+    return {"status": "ok", "id": item.id}
+
+
+@app.put("/api/labware/{item_id}")
+async def api_update_labware(item_id: str, request: Request):
+    data = await request.json()
+    db = _load_db()
+    for i, it in enumerate(db.items):
+        if it.id == item_id or it.function_name == item_id:
+            updated = LabwareItem(**{**it.model_dump(), **data, "id": it.id})
+            db.items[i] = updated
+            _save_db(db)
+            return {"status": "ok", "id": it.id}
+    raise HTTPException(404, "耗材不存在")
+
+
+@app.delete("/api/labware/{item_id}")
+async def api_delete_labware(item_id: str):
+    db = _load_db()
+    original_len = len(db.items)
+    db.items = [it for it in db.items if it.id != item_id and it.function_name != item_id]
+    if len(db.items) == original_len:
+        raise HTTPException(404, "耗材不存在")
+    _save_db(db)
+    return {"status": "ok"}
+
+
+@app.post("/api/generate-code")
+async def api_generate_code(request: Request):
+    body = await request.json() if await request.body() else {}
+    test_mode = body.get("test_mode", True)
+    db = _load_db()
+    if not db.items:
+        raise HTTPException(400, "数据库为空，请先导入")
+
+    from unilabos.labware_manager.codegen import generate_code
+    from unilabos.labware_manager.yaml_gen import generate_yaml
+
+    py_path = generate_code(db, test_mode=test_mode)
+    yaml_paths = generate_yaml(db, test_mode=test_mode)
+
+    return {
+        "status": "ok",
+        "python_file": str(py_path),
+        "yaml_files": [str(p) for p in yaml_paths],
+        "test_mode": test_mode,
+    }
+
+
+@app.post("/api/import-from-code")
+async def api_import_from_code():
+    from unilabos.labware_manager.importer import import_from_code, save_db
+    db = import_from_code()
+    save_db(db)
+    return {
+        "status": "ok",
+        "count": len(db.items),
+        "items": [{"function_name": it.function_name, "type": it.type} for it in db.items],
+    }
+
+
+# ---------- 辅助函数 ----------
+
+def _find_item(db: LabwareDB, item_id: str) -> Optional[LabwareItem]:
+    for item in db.items:
+        if item.id == item_id or item.function_name == item_id:
+            return item
+    return None
+
+
+# ---------- 启动入口 ----------
+
+def main():
+    import uvicorn
+    port = int(os.environ.get("LABWARE_PORT", "8010"))
+    print(f"PRCXI 耗材管理 → http://localhost:{port}")
+    uvicorn.run(app, host="0.0.0.0", port=port)
+
+
+if __name__ == "__main__":
+    main()