225 lines
6.5 KiB
Python
225 lines
6.5 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Annotated, Any
|
|
|
|
import typer
|
|
|
|
from .apps import component_option_app, network_app, project_app
|
|
from .common import emit_api, runtime_context
|
|
from .core import CLIError, emit_success, request_bytes, request_json, require_network
|
|
|
|
|
|
@project_app.command("list")
|
|
def project_list(ctx: typer.Context) -> None:
|
|
emit_api(ctx, summary="读取项目列表成功", method="GET", path="/meta/projects", require_auth=True)
|
|
|
|
|
|
@project_app.command("info")
|
|
def project_info(ctx: typer.Context) -> None:
|
|
emit_api(
|
|
ctx,
|
|
summary="读取项目信息成功",
|
|
method="GET",
|
|
path="/meta/project",
|
|
require_auth=True,
|
|
require_project=True,
|
|
)
|
|
|
|
|
|
@project_app.command("db-health")
|
|
def project_db_health(ctx: typer.Context) -> None:
|
|
emit_api(
|
|
ctx,
|
|
summary="读取数据库健康状态成功",
|
|
method="GET",
|
|
path="/meta/db/health",
|
|
require_auth=True,
|
|
require_project=True,
|
|
)
|
|
|
|
|
|
@project_app.command("data")
|
|
def project_data(
|
|
ctx: typer.Context,
|
|
kind: Annotated[str, typer.Option("--kind", help="scada-info|scheme-list|burst-locate-result")],
|
|
) -> None:
|
|
kind_map = {
|
|
"scada-info": "/scada-info",
|
|
"scheme-list": "/scheme-list",
|
|
"burst-locate-result": "/burst-locate-result",
|
|
}
|
|
path = kind_map.get(kind)
|
|
if path is None:
|
|
raise CLIError(
|
|
"CLI 参数错误",
|
|
code="INVALID_KIND",
|
|
message="kind must be one of: scada-info, scheme-list, burst-locate-result",
|
|
exit_code=2,
|
|
)
|
|
emit_api(
|
|
ctx,
|
|
summary="读取项目数据成功",
|
|
method="GET",
|
|
path=path,
|
|
require_auth=True,
|
|
require_project=True,
|
|
)
|
|
|
|
|
|
@project_app.command("export-inp")
|
|
def project_export_inp(
|
|
ctx: typer.Context,
|
|
output: Annotated[Path, typer.Option("--output", help="本地输出路径")],
|
|
) -> None:
|
|
runtime = runtime_context(ctx)
|
|
network = require_network(runtime)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
temp_name = f"{output.stem}-{runtime.request_id}.inp"
|
|
_, duration_dump = request_json(
|
|
runtime,
|
|
method="GET",
|
|
path="/dumpinp/",
|
|
params={"network": network, "inp": temp_name},
|
|
require_auth=True,
|
|
require_network_ctx=True,
|
|
)
|
|
content, duration_download = request_bytes(
|
|
runtime,
|
|
method="GET",
|
|
path="/downloadinp/",
|
|
params={"name": temp_name},
|
|
require_auth=True,
|
|
require_network_ctx=True,
|
|
)
|
|
output.write_bytes(content)
|
|
emit_success(
|
|
summary="导出 INP 成功",
|
|
data={"output": str(output), "bytes": len(content)},
|
|
ctx=runtime,
|
|
duration_ms=duration_dump + duration_download,
|
|
next_commands=["tjwater-cli project info"],
|
|
)
|
|
|
|
|
|
@network_app.command("get-node-properties")
|
|
def network_get_node_properties(
|
|
ctx: typer.Context,
|
|
node: Annotated[str, typer.Option("--node", help="节点 ID")],
|
|
) -> None:
|
|
runtime = runtime_context(ctx)
|
|
emit_api(
|
|
ctx,
|
|
summary="读取节点属性成功",
|
|
method="GET",
|
|
path="/getnodeproperties/",
|
|
params={"network": require_network(runtime), "node": node},
|
|
require_auth=True,
|
|
require_network_ctx=True,
|
|
)
|
|
|
|
|
|
@network_app.command("get-link-properties")
|
|
def network_get_link_properties(
|
|
ctx: typer.Context,
|
|
link: Annotated[str, typer.Option("--link", help="管线 ID")],
|
|
) -> None:
|
|
runtime = runtime_context(ctx)
|
|
emit_api(
|
|
ctx,
|
|
summary="读取管线属性成功",
|
|
method="GET",
|
|
path="/getlinkproperties/",
|
|
params={"network": require_network(runtime), "link": link},
|
|
require_auth=True,
|
|
require_network_ctx=True,
|
|
)
|
|
|
|
|
|
def _component_option_mapping(kind: str, pump: str | None) -> tuple[str, dict[str, Any]]:
|
|
if kind == "time":
|
|
return "/gettimeschema", {}
|
|
if kind == "energy":
|
|
return "/getenergyschema/", {}
|
|
if kind == "pump-energy":
|
|
if not pump:
|
|
raise CLIError(
|
|
"CLI 参数错误",
|
|
code="PUMP_REQUIRED",
|
|
message="--pump is required when --kind pump-energy",
|
|
exit_code=2,
|
|
)
|
|
return "/getpumpenergyschema/", {"pump": pump}
|
|
if kind == "network":
|
|
return "/getoptionschema/", {}
|
|
raise CLIError(
|
|
"CLI 参数错误",
|
|
code="INVALID_KIND",
|
|
message="kind must be one of: time, energy, pump-energy, network",
|
|
exit_code=2,
|
|
)
|
|
|
|
|
|
def _component_option_get_mapping(kind: str, pump: str | None) -> tuple[str, dict[str, Any]]:
|
|
if kind == "time":
|
|
return "/gettimeproperties/", {}
|
|
if kind == "energy":
|
|
return "/getenergyproperties/", {}
|
|
if kind == "pump-energy":
|
|
if not pump:
|
|
raise CLIError(
|
|
"CLI 参数错误",
|
|
code="PUMP_REQUIRED",
|
|
message="--pump is required when --kind pump-energy",
|
|
exit_code=2,
|
|
)
|
|
return "/getpumpenergyproperties/", {"pump": pump}
|
|
if kind == "network":
|
|
return "/getoptionproperties/", {}
|
|
raise CLIError(
|
|
"CLI 参数错误",
|
|
code="INVALID_KIND",
|
|
message="kind must be one of: time, energy, pump-energy, network",
|
|
exit_code=2,
|
|
)
|
|
|
|
|
|
@component_option_app.command("schema")
|
|
def component_option_schema(
|
|
ctx: typer.Context,
|
|
kind: Annotated[str, typer.Option("--kind", help="time|energy|pump-energy|network")],
|
|
pump: Annotated[str | None, typer.Option("--pump", help="泵 ID")] = None,
|
|
) -> None:
|
|
runtime = runtime_context(ctx)
|
|
path, extra = _component_option_mapping(kind, pump)
|
|
params = {"network": require_network(runtime)} | extra
|
|
emit_api(
|
|
ctx,
|
|
summary="读取选项 schema 成功",
|
|
method="GET",
|
|
path=path,
|
|
params=params,
|
|
require_auth=True,
|
|
require_network_ctx=True,
|
|
)
|
|
|
|
|
|
@component_option_app.command("get")
|
|
def component_option_get(
|
|
ctx: typer.Context,
|
|
kind: Annotated[str, typer.Option("--kind", help="time|energy|pump-energy|network")],
|
|
pump: Annotated[str | None, typer.Option("--pump", help="泵 ID")] = None,
|
|
) -> None:
|
|
runtime = runtime_context(ctx)
|
|
path, extra = _component_option_get_mapping(kind, pump)
|
|
params = {"network": require_network(runtime)} | extra
|
|
emit_api(
|
|
ctx,
|
|
summary="读取选项属性成功",
|
|
method="GET",
|
|
path=path,
|
|
params=params,
|
|
require_auth=True,
|
|
require_network_ctx=True,
|
|
)
|