9 Commits
Author SHA1 Message Date
jiang fdbcc5c033 docs: clarify production configuration handling 2026-08-20 16:19:44 +08:00
jiang d63d7ef1b6 merge: route project DSNs and remove legacy storage backends
Merge PR #2 after isolated OpenAPI, test, container build, and runtime smoke verification.
2026-08-18 18:34:47 +08:00
jiang 6b09662de6 refactor(storage): route project DSNs and remove legacy backends 2026-08-18 18:29:09 +08:00
jiang b21eaffe40 merge: integrate agent-mvp into master
Merge PR #1 after backend security and contract gates passed.
2026-08-18 17:56:43 +08:00
jiang 8853877fcd fix(security): close backend merge blockers 2026-08-18 17:51:29 +08:00
jiang 2581631b51 feat(simulation): 支持冲洗阀门状态与设置值
Generic Container CI/CD / test-build-publish (push) Successful in 2m39s
Server CI/CD v2 / build-test-publish-and-deploy (push) Successful in 2m40s
2026-08-17 18:28:57 +08:00
jiang c250e97b87 ci(backend): block releases missing frontend API contract 2026-08-11 11:11:39 +08:00
jiang 69a7d53aff ci: replace webhook deployment with v2 workflow
Generic Container CI/CD / test-build-publish (push) Successful in 2m32s
Server CI/CD v2 / build-test-publish-and-deploy (push) Successful in 2m32s
2026-08-11 10:17:01 +08:00
jiang e4975b7be3 fix(container): exclude runtime configuration from image 2026-08-11 10:10:21 +08:00
69 changed files with 1150 additions and 11260 deletions
+5 -2
View File
@@ -11,9 +11,12 @@ dist/
package/
temp/
data/
# db_inp/
db_inp/
inp/
# .env
.env
.env.*
logs/
coverage/
*.pyc
*.dump
app/algorithms/health/model/my_survival_forest_model_quxi.joblib
+1 -1
View File
@@ -1,6 +1,6 @@
# TJWater Server 环境变量配置模板
# 复制此文件为 .env 并填写实际值
# CI/CD: 生产 .env 的完整内容保存为 Gitea 仓库密钥 TJWATER_SERVER_ENV
# CI/CD: 生产环境变量由 Dev 主机的受控 backend.env 注入,不要将完整 .env 保存为 Gitea 仓库密钥
ENVIRONMENT="production"
NETWORK_NAME="tjwater"
# ============================================
+16 -252
View File
@@ -1,263 +1,27 @@
name: Server CI/CD
name: Server CI/CD v2
on:
push:
tags:
- "v*"
- "latest"
workflow_dispatch: {}
jobs:
docker-image:
runs-on: ubuntu-22.04
if: startsWith(github.ref, 'refs/tags/')
permissions:
contents: read
defaults:
run:
shell: bash
steps:
- name: Checkout repository
uses: https://gitea.waternetwork.cn/actions/checkout@v4
build-test-publish-and-deploy:
uses: OrgTJWater/ci-templates/.gitea/workflows/container-cd.yml@main
with:
fetch-depth: 1
- name: Normalize image metadata
env:
RAW_REGISTRY_HOST: ${{ vars.REGISTRY_HOST }}
RAW_REPOSITORY: ${{ github.repository }}
RAW_REF_NAME: ${{ github.ref_name }}
run: |
RAW_REGISTRY_HOST="$(printf '%s' "${RAW_REGISTRY_HOST}" | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')"
if [ -z "${RAW_REGISTRY_HOST}" ]; then
echo "Missing required repository variable: REGISTRY_HOST"
exit 1
fi
REGISTRY_HOST="${RAW_REGISTRY_HOST#http://}"
REGISTRY_HOST="${REGISTRY_HOST#https://}"
REGISTRY_HOST="${REGISTRY_HOST%/}"
if [ -z "${REGISTRY_HOST}" ]; then
echo "Repository variable REGISTRY_HOST resolves to an empty host"
exit 1
fi
REPOSITORY_PATH="${RAW_REPOSITORY#/}"
IMAGE_OWNER="${REPOSITORY_PATH%%/*}"
IMAGE_REPOSITORY_PATH="$(printf '%s' "${IMAGE_OWNER}/tjwater-backend" | tr '[:upper:]' '[:lower:]')"
IMAGE_NAME="${REGISTRY_HOST}/${IMAGE_REPOSITORY_PATH}"
IMAGE_TAG="${RAW_REF_NAME}"
{
echo "REGISTRY_HOST=${REGISTRY_HOST}"
echo "REPOSITORY_PATH=${REPOSITORY_PATH}"
echo "IMAGE_REPOSITORY_PATH=${IMAGE_REPOSITORY_PATH}"
echo "IMAGE_NAME=${IMAGE_NAME}"
echo "IMAGE_TAG=${IMAGE_TAG}"
echo "IMAGE_REF=${IMAGE_NAME}:${IMAGE_TAG}"
} >> "$GITHUB_ENV"
- name: Login to Gitea Container Registry
env:
image_name: gitea.waternetwork.cn/orgtjwater/tjwater-backend
dockerfile: Dockerfile
build_context: .
test_command: |
test -f app/api/v1/endpoints/access.py
grep -Fq 'api_router.include_router(access.router' app/api/v1/router.py
grep -Fq '@router.get("/projects"' app/api/v1/endpoints/meta.py
grep -Fq '@router.get("/projects/current"' app/api/v1/endpoints/project.py
grep -Fq '@router.post("/audit-events"' app/api/v1/endpoints/audit.py
deploy_service: backend
deploy_host: 192.168.1.114
secrets:
REGISTRY_USERNAME: ${{ secrets.REGISTRY_USERNAME }}
REGISTRY_PASSWORD: ${{ secrets.REGISTRY_PASSWORD }}
GITEA_SERVER_URL: ${{ github.server_url }}
run: |
if [ -z "${REGISTRY_HOST:-}" ]; then
echo "Missing resolved environment value: REGISTRY_HOST"
exit 1
fi
if [ -z "${REGISTRY_USERNAME}" ]; then
echo "Missing required repository secret: REGISTRY_USERNAME"
exit 1
fi
if [ -z "${REGISTRY_PASSWORD}" ]; then
echo "Missing required repository secret: REGISTRY_PASSWORD"
exit 1
fi
echo "Registry username: ${REGISTRY_USERNAME}"
echo "Image target: ${IMAGE_REF}"
API_SERVER_URL="${GITEA_SERVER_URL%/}"
api_user="$(curl -fsS \
-H "Authorization: token ${REGISTRY_PASSWORD}" \
"${API_SERVER_URL}/api/v1/user" \
| sed -n 's/.*"login"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' \
| head -n 1 || true)"
if [ -n "${api_user}" ]; then
echo "Registry token resolves to Gitea user: ${api_user}"
else
echo "Could not resolve Gitea user from REGISTRY_PASSWORD token; docker login may still use a password or a token without API access."
fi
echo "Logging into registry host: ${REGISTRY_HOST}"
echo "${REGISTRY_PASSWORD}" | docker login "$REGISTRY_HOST" \
--username "${REGISTRY_USERNAME}" \
--password-stdin
- name: Materialize runtime env file
env:
TJWATER_SERVER_ENV: ${{ secrets.TJWATER_SERVER_ENV }}
run: |
if [ -z "${TJWATER_SERVER_ENV}" ]; then
echo "Missing required repository secret: TJWATER_SERVER_ENV"
echo "Store the backend .env file content as a multiline Gitea repository secret named TJWATER_SERVER_ENV."
exit 1
fi
printf '%s\n' "${TJWATER_SERVER_ENV}" > .env
chmod 600 .env
required_env_keys=(
ENVIRONMENT
NETWORK_NAME
DB_NAME
DB_HOST
DB_PORT
DB_USER
DB_PASSWORD
TIMESCALEDB_DB_NAME
TIMESCALEDB_DB_HOST
TIMESCALEDB_DB_PORT
TIMESCALEDB_DB_USER
TIMESCALEDB_DB_PASSWORD
METADATA_DB_NAME
METADATA_DB_HOST
METADATA_DB_PORT
METADATA_DB_USER
METADATA_DB_PASSWORD
DATABASE_ENCRYPTION_KEY
)
missing_keys=()
for key in "${required_env_keys[@]}"; do
if ! grep -Eq "^[[:space:]]*(export[[:space:]]+)?${key}[[:space:]]*=" .env; then
missing_keys+=("$key")
fi
done
if [ "${#missing_keys[@]}" -gt 0 ]; then
echo "TJWATER_SERVER_ENV is missing required keys: ${missing_keys[*]}"
exit 1
fi
- name: Validate workspace
run: |
if [ ! -f ./Dockerfile ]; then
echo "Dockerfile not found in workspace. Repository checkout may have failed or produced an unexpected workspace."
exit 1
fi
- name: Build and Push Image
run: |
if [ -z "${IMAGE_NAME:-}" ] || [ -z "${IMAGE_TAG:-}" ]; then
echo "Missing resolved image metadata: IMAGE_NAME or IMAGE_TAG"
exit 1
fi
push_with_retry() {
image_ref="$1"
attempt=1
max_attempts=3
while [ "$attempt" -le "$max_attempts" ]; do
if docker push "$image_ref"; then
return 0
fi
if [ "$attempt" -eq "$max_attempts" ]; then
return 1
fi
echo "Push failed for $image_ref (attempt $attempt/$max_attempts); retrying in 10s..."
attempt=$((attempt + 1))
sleep 10
done
}
if [ "${IMAGE_TAG}" = "latest" ]; then
docker build \
-f ./Dockerfile \
-t "${IMAGE_NAME}:latest" \
.
push_with_retry "${IMAGE_NAME}:latest"
else
docker build \
-f ./Dockerfile \
-t "${IMAGE_NAME}:${IMAGE_TAG}" \
-t "${IMAGE_NAME}:latest" \
.
push_with_retry "${IMAGE_NAME}:${IMAGE_TAG}"
push_with_retry "${IMAGE_NAME}:latest"
fi
- name: Notify Deploy Server
run: |
post_deploy_webhook() {
label="$1"
payload="$2"
webhook_url="${{ vars.DEPLOY_WEBHOOK_URL }}"
token="${{ secrets.DEPLOY_WEBHOOK_TOKEN }}"
webhook_url=$(echo "$webhook_url" | xargs)
if [ -z "$webhook_url" ]; then
echo "Missing required repository variable: DEPLOY_WEBHOOK_URL"
return 1
fi
if [ -z "$token" ]; then
echo "Missing required repository secret: DEPLOY_WEBHOOK_TOKEN"
return 1
fi
echo "[$label] Calling webhook: $webhook_url"
http_code=$(curl -sS -D /tmp/deploy_headers.txt -o /tmp/deploy_response.txt -w "%{http_code}" -X POST "$webhook_url" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $token" \
-d "$payload")
echo "[$label] webhook HTTP status: ${http_code}"
if [ "$http_code" -ge 200 ] && [ "$http_code" -lt 300 ]; then
return 0
fi
echo "[$label] response headers:"
cat /tmp/deploy_headers.txt
echo "[$label] response body:"
cat /tmp/deploy_response.txt
return 1
}
PRIMARY_PAYLOAD="{\"image\":\"${IMAGE_REF}\",\"tag\":\"${IMAGE_TAG}\",\"repo\":\"${REPOSITORY_PATH}\"}"
FALLBACK_PAYLOAD="{\"image\":\"${IMAGE_REF}\",\"tag\":\"${IMAGE_TAG}\",\"repo\":\"${IMAGE_REPOSITORY_PATH}\"}"
echo "Deploy webhook target: ${{ vars.DEPLOY_WEBHOOK_URL }}"
echo "Deploy payload(primary): image=${IMAGE_REF}, tag=${IMAGE_TAG}, repo=${REPOSITORY_PATH}"
if post_deploy_webhook "primary" "$PRIMARY_PAYLOAD"; then
exit 0
fi
echo "Primary webhook request failed, retrying with lowercase repo path..."
echo "Deploy payload(fallback): image=${IMAGE_REF}, tag=${IMAGE_TAG}, repo=${IMAGE_REPOSITORY_PATH}"
if post_deploy_webhook "fallback" "$FALLBACK_PAYLOAD"; then
exit 0
fi
echo "Deploy webhook failed after primary and fallback attempts."
exit 1
deploy-fallback-log:
runs-on: ubuntu-22.04
needs: docker-image
if: failure()
steps:
- name: Deployment not triggered
run: echo "Image build/push failed, deployment webhook was not called."
DEV_DEPLOY_SSH_KEY: ${{ secrets.DEV_DEPLOY_SSH_KEY }}
+1 -1
View File
@@ -35,4 +35,4 @@ Pull requests should describe the behavior change, list verification commands, m
## Security & Configuration Tips
Do not commit `.env`, database dumps, generated caches, or local project data. Use `.env.example` as the configuration template. Secrets for CI/CD belong in Gitea repository secrets such as `REGISTRY_USERNAME`, `REGISTRY_PASSWORD`, and deploy webhook credentials.
Do not commit `.env`, database dumps, generated caches, or local project data. Use `.env.example` as the configuration template. CI/CD only uses Gitea repository secrets `REGISTRY_USERNAME`, `REGISTRY_PASSWORD`, and `DEV_DEPLOY_SSH_KEY`; production application settings are injected on the Dev host.
+1 -1
View File
@@ -49,7 +49,7 @@ These route groups expose many command-style concatenated paths. They should not
- Network object CRUD: `addjunction`, `getjunctionelevation`, `setpipediameter`, `getvalvesetting`, and similar junction/pipe/pump/tank/reservoir/valve routes
- Region/DMA/VD commands: `calculatedistrictmeteringareaforregion`, `getdistrictmeteringarea`, `generatevirtualdistrict`, and related routes
- SCADA native CRUD: `getscadadevice`, `setscadadevicedata`, `cleanscadaelement`, and related routes
- Snapshot/cache utilities: `takesnapshotforoperation`, `syncwithserver`, `clearrediskey`, `queryredis`
- Snapshot/synchronization utilities: `takesnapshotforoperation`, `syncwithserver`
- Advanced simulation endpoints with underscore paths: `pressure_regulation`, `daily_scheduling_analysis`, `network_update`, `pressure_sensor_placement_kmeans`
### Direct Cleanup Candidates
+2 -4
View File
@@ -14,13 +14,11 @@ COPY requirements.txt .
RUN pip install --no-cache-dir uv
RUN uv pip install --system --no-cache-dir -r requirements.txt
# 将代码放入子目录 'app',将数据放入子目录 'db_inp'
# 这样临时文件默认会生成在 /app 下,而代码在 /app/app 下,实现了分离
# 本地数据目录和环境变量在运行时通过 Compose 挂载或注入,
# 不应进入镜像构建上下文。
COPY app ./app
RUN python -c "from pathlib import Path; from zipfile import ZipFile; model_dir = Path('app/algorithms/health/model'); zip_path = model_dir / 'my_survival_forest_model_quxi.zip'; joblib_name = 'my_survival_forest_model_quxi.joblib'; joblib_path = model_dir / joblib_name; assert zip_path.exists(), f'Model archive not found: {zip_path}'; archive = ZipFile(zip_path); archive.extract(joblib_name, model_dir); archive.close(); assert joblib_path.exists(), f'Model file not extracted: {joblib_path}'" && \
rm -f app/algorithms/health/model/my_survival_forest_model_quxi.zip
# COPY db_inp ./db_inp
COPY .env .
RUN mkdir -p db_inp temp data inp
# 设置 PYTHONPATH 以便 uvicorn 找到 app 模块
+12 -2
View File
@@ -7,7 +7,7 @@
- Python 3.12
- FastAPI / Uvicorn
- Pydantic / SQLAlchemy / psycopg
- Redis、PostgreSQL、PostGIS、TimescaleDB
- PostgreSQL、PostGIS、TimescaleDB
- WNTR、EPANET、Cython、科学计算与空间分析依赖
- pytest
@@ -19,7 +19,7 @@ app/api/ HTTP API 路由
app/auth/ 认证和权限上下文
app/core/ 配置、日志和基础设施初始化
app/domain/ 领域模型和 Pydantic schema
app/infra/ 数据库、缓存、EPANET 和外部集成
app/infra/ 数据库、EPANET 和外部集成
app/services/ 业务服务编排
app/algorithms/ 管网算法、模拟、爆管、漏损、清洗和健康分析
app/native/ 本地管网数据读写与转换
@@ -65,6 +65,16 @@ docker compose -f infra/docker/docker-compose.yml config
- 优先复用现有 FastAPI/service/repository 边界。
- 不要把临时数据、数据库 dump、日志或本地运行产物纳入提交。
## 项目数据库路由
项目级 REST 请求通过 `X-Project-Id` 解析元数据中的数据库配置:
- `biz_data` DSN 用于管网业务数据;`{project_code}_template` 和模拟临时库沿用该 DSN 的主机、端口与凭据,仅替换数据库名。
- `iot_data` DSN 用于 TimescaleDB,始终使用元数据配置的完整 DSN,不再从项目代码推导数据库名。
- 元数据、业务库和 TimescaleDB 可以部署在同一主机,也可以分别部署。
使用模板复制或临时方案库的模拟功能时,`biz_data` 账号必须具备现有数据库创建、删除和连接终止操作所需的 PostgreSQL 权限。
## 测试与发布
提交前根据改动范围运行最小有效测试:
+4
View File
@@ -316,6 +316,7 @@ def flushing_analysis(
flushing_flow: float = 0,
scheme_name: str = None,
username: str | None = None,
valve_control: dict[str, dict] = None,
) -> None:
"""
管道冲洗模拟
@@ -323,6 +324,7 @@ def flushing_analysis(
:param modify_pattern_start_time: 模拟开始时间,格式为'2024-11-25T09:00:00+08:00'
:param modify_total_duration: 模拟总历时,秒
:param modify_valve_opening: dict中包含多个阀门开启度,str为阀门的id,float为修改后的阀门开启度
:param valve_control: dict中可分别指定阀门的status、setting和k
:param drainage_node_ID: 冲洗排放口所在节点ID
:param flushing_flow: 冲洗水量,传入参数单位为m3/h
:param scheme_name: 方案名称
@@ -334,6 +336,7 @@ def flushing_analysis(
scheme_detail: dict = {
"duration": modify_total_duration,
"valve_opening": modify_valve_opening,
"valve_control": valve_control,
"drainage_node_ID": drainage_node_ID,
"flushing_flow": flushing_flow,
}
@@ -450,6 +453,7 @@ def flushing_analysis(
modify_pattern_start_time=modify_pattern_start_time,
modify_total_duration=modify_total_duration,
modify_valve_opening=modify_valve_opening,
valve_control=valve_control,
scheme_type="flushing_analysis",
scheme_name=scheme_name,
)
+14
View File
@@ -0,0 +1,14 @@
from __future__ import annotations
from collections.abc import Iterable
from typing import Generic, TypeVar
T = TypeVar("T")
class PaginatedList(list[T], Generic[T]):
"""A page of items carrying the total count from its data source."""
def __init__(self, items: Iterable[T], *, total: int) -> None:
super().__init__(items)
self.total = total
+19 -2
View File
@@ -10,6 +10,7 @@ from app.auth.metadata_dependencies import (
get_current_metadata_admin,
get_current_metadata_user,
)
from app.api.pagination import PaginatedList
from app.core.audit import AuditAction, log_audit_event
from app.domain.schemas.audit import AuditLogResponse
from app.infra.db.metadb.database import get_metadata_session
@@ -46,7 +47,7 @@ async def get_audit_logs(
_current_user=Depends(get_current_metadata_admin),
audit_repo: AuditRepository = Depends(get_audit_repository),
) -> list[AuditLogResponse]:
return await audit_repo.get_logs(
items = await audit_repo.get_logs(
user_id=user_id,
project_id=project_id,
action=action,
@@ -56,6 +57,15 @@ async def get_audit_logs(
skip=skip,
limit=limit,
)
total = await audit_repo.get_log_count(
user_id=user_id,
project_id=project_id,
action=action,
resource_type=resource_type,
start_time=start_time,
end_time=end_time,
)
return PaginatedList(items, total=total)
@router.get(
@@ -119,7 +129,7 @@ async def get_my_audit_logs(
current_user=Depends(get_current_metadata_user),
audit_repo: AuditRepository = Depends(get_audit_repository),
) -> list[AuditLogResponse]:
return await audit_repo.get_logs(
items = await audit_repo.get_logs(
user_id=current_user.id,
action=action,
start_time=start_time,
@@ -127,3 +137,10 @@ async def get_my_audit_logs(
skip=skip,
limit=limit,
)
total = await audit_repo.get_log_count(
user_id=current_user.id,
action=action,
start_time=start_time,
end_time=end_time,
)
return PaginatedList(items, total=total)
-57
View File
@@ -1,57 +0,0 @@
from fastapi import APIRouter, Query
from app.infra.cache.redis_client import redis_client
router = APIRouter()
@router.delete("/redis-keys/detail", summary="清除单个缓存键", description="根据键名清除单个Redis缓存")
async def fastapi_clear_redis_key(key: str = Query(..., description="缓存键名")):
"""
清除单个缓存键
根据指定的键名删除Redis中对应的缓存
"""
redis_client.delete(key)
return True
@router.delete("/redis-keys", summary="清除匹配的缓存键", description="根据模式清除匹配的Redis缓存键")
async def fastapi_clear_redis_keys(keys: str = Query(..., description="缓存键模式(支持通配符)")):
"""
清除匹配的缓存键
根据指定的模式删除Redis中所有匹配的缓存键
"""
# delete keys contains the key
matched_keys = redis_client.keys(f"*{keys}*")
if matched_keys:
redis_client.delete(*matched_keys)
return True
@router.delete("/all-redis", summary="清除所有缓存", description="清空整个Redis数据库的所有缓存")
async def fastapi_clear_all_redis():
"""
清除所有缓存
清空Redis数据库中的所有缓存键值对
"""
redis_client.flushdb()
return True
@router.get("/redis", summary="查询缓存键列表", description="获取Redis中所有的缓存键")
async def fastapi_query_redis():
"""
查询缓存键列表
获取Redis数据库中所有的缓存键列表
"""
# Helper to decode bytes to str for JSON response if needed,
# but original just returned keys (which might be bytes in redis-py unless decode_responses=True)
# create_redis_client usually sets decode_responses=False by default.
# We will assume user handles bytes or we should decode.
# Original just returned redis_client.keys("*")
keys = redis_client.keys("*")
# Clean output for API
return [k.decode('utf-8') if isinstance(k, bytes) else k for k in keys]
+22 -5
View File
@@ -17,8 +17,13 @@ from app.auth.metadata_dependencies import (
get_current_metadata_admin,
get_metadata_repository,
)
from app.auth.project_dependencies import (
ProjectContext,
resolve_project_business_routing,
)
from app.core.audit import AuditAction, log_audit_event
from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository
from app.infra.db.project_routing import activate_project_routing
from app.services.network_import import network_update
from app.services.tjnetwork import run_inp
@@ -118,21 +123,21 @@ async def _run_uploaded_inp(content: bytes) -> str:
return run_inp(model_name)
async def _update_from_inp(content: bytes) -> None:
async def _update_from_inp(content: bytes, project_code: str) -> None:
temp_path: Path | None = None
try:
with NamedTemporaryFile(suffix=".inp", delete=False) as temp_file:
temp_file.write(content)
temp_path = Path(temp_file.name)
network_update(str(temp_path))
network_update(str(temp_path), project_code)
finally:
if temp_path is not None:
temp_path.unlink(missing_ok=True)
async def _apply_model_update(content: bytes) -> None:
async def _apply_model_update(content: bytes, project_code: str) -> None:
try:
await _update_from_inp(content)
await _update_from_inp(content, project_code)
except Exception as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
@@ -177,7 +182,19 @@ async def update_project_model(
) -> dict:
project = await _get_active_project(project_id, metadata_repo)
content, filename = await _read_upload(file)
await _apply_model_update(content)
routing = await resolve_project_business_routing(
ProjectContext(
project_id=project.id,
project_code=project.code,
user_id=current_user.id,
project_role="owner",
system_role=current_user.role,
is_superuser=current_user.is_superuser,
),
metadata_repo,
)
with activate_project_routing(routing):
await _apply_model_update(content, project.code)
await _audit_model_change(
request=request,
current_user=current_user,
@@ -333,7 +333,6 @@ async def fastapi_get_all_junction_properties(
list: 包含所有节点属性的列表
"""
# 缓存查询结果提高性能
# global redis_client # Redis logic removed for clean split, can be re-added if needed or imported
results = get_all_junctions(network)
return results
-1
View File
@@ -385,7 +385,6 @@ async def fastapi_get_all_pipe_properties(
包含所有管道属性的字典列表
"""
# 缓存查询结果提高性能
# global redis_client
results = get_all_pipes(network)
return results
-1
View File
@@ -177,7 +177,6 @@ async def fastapi_get_all_pump_properties(
包含所有水泵属性的字典列表
"""
# 缓存查询结果提高性能
# global redis_client
results = get_all_pumps(network)
return results
-1
View File
@@ -540,7 +540,6 @@ async def fastapi_get_all_tank_properties(
包含所有水箱属性的字典列表
"""
# 缓存查询结果提高性能
# global redis_client
results = get_all_tanks(network)
return results
-1
View File
@@ -307,7 +307,6 @@ async def fastapi_get_all_valve_properties(
返回指定水网中所有阀门的完整属性列表
"""
# 缓存查询结果提高性能
# global redis_client
results = get_all_valves(network)
return results
+78 -7
View File
@@ -1,4 +1,4 @@
from typing import Any, List, Optional
from typing import Any, List, Literal, Optional
from datetime import datetime, timedelta
import json
import threading
@@ -302,12 +302,20 @@ async def valve_isolation_endpoint(
return result
@router.post("/flushing-analyses", response_class=PlainTextResponse, summary="冲洗分析(高级)", description="高级版本的冲洗分析,支持同时开启多个阀门进行冲洗,指定排污节点,并设置固定的冲洗流量。返回纯文本格式的分析结果。")
@router.post("/flushing-analyses", response_class=PlainTextResponse, summary="冲洗分析(高级)", description="高级版本的冲洗分析,支持按状态和设置值控制多个可选阀门,指定排污节点,并设置固定的冲洗流量。返回纯文本格式的分析结果。")
async def fastapi_flushing_analysis(
network: str = Query(..., description="管网名称(或数据库名称)"),
start_time: str = Query(..., description="冲洗开始时间(ISO 8601格式)"),
valves: List[str] = Query(..., description="要开启的阀门ID列表"),
valves_k: List[float] = Query(..., description="对应各阀门的开度列表(0-1"),
valves: List[str] | None = Query(None, description="参与控制的阀门ID列表(可选)"),
valves_k: List[float] | None = Query(
None, description="对应各阀门的开度列表(0-1,可选,与valves同时提供)"
),
valve_statuses: List[Literal["OPEN", "CLOSED", "ACTIVE"]] | None = Query(
None, description="对应各阀门的开关状态列表(OPEN、CLOSED或ACTIVE,可选)"
),
valve_settings: List[str] | None = Query(
None, description="对应各阀门的设置值列表(ACTIVE状态下必填)"
),
drainage_node_ID: str = Query(..., description="排污节点ID"),
flush_flow: float = Query(0, description="冲洗流量(L/s),0表示自动计算"),
duration: int | None = Query(None, description="模拟持续时间(秒),默认900秒"),
@@ -319,8 +327,10 @@ async def fastapi_flushing_analysis(
- **network**: 管网名称或数据库名称
- **start_time**: 冲洗开始时间
- **valves**: 要开启的阀门ID列表
- **valves_k**: 各阀门的开度列表0-1与valves对应
- **valves**: 参与控制的阀门ID列表可选
- **valves_k**: 各阀门的开度列表0-1可选与valves同时提供
- **valve_statuses**: 各阀门的开关状态列表OPENCLOSED或ACTIVE可选
- **valve_settings**: 各阀门的设置值列表ACTIVE状态下必填
- **drainage_node_ID**: 排污节点ID
- **flush_flow**: 冲洗流量L/s
- **duration**: 模拟持续时间可选默认900
@@ -328,14 +338,75 @@ async def fastapi_flushing_analysis(
支持多阀联合冲洗操作
"""
valve_opening = None
valve_control = None
if valve_statuses is not None and valves_k is not None:
raise HTTPException(
status_code=422,
detail="valve_statuses 和 valves_k 不能同时提供",
)
if valve_settings is not None and valve_statuses is None:
raise HTTPException(
status_code=422,
detail="valve_settings 必须与 valve_statuses 同时提供",
)
if valves is None:
if (
valves_k is not None
or valve_statuses is not None
or valve_settings is not None
):
raise HTTPException(
status_code=422,
detail="阀门控制参数必须与 valves 同时提供",
)
elif valve_statuses is not None:
if len(valves) != len(valve_statuses):
raise HTTPException(
status_code=422, detail="valves 和 valve_statuses 的数量必须一致"
)
if valve_settings is not None and len(valves) != len(valve_settings):
raise HTTPException(
status_code=422, detail="valves 和 valve_settings 的数量必须一致"
)
settings = valve_settings or [""] * len(valves)
valve_control = {}
for valve_id, raw_status, raw_setting in zip(
valves, valve_statuses, settings
):
status = raw_status
setting = raw_setting.strip()
if status == "ACTIVE" and not setting:
raise HTTPException(
status_code=422,
detail=f"ACTIVE 状态的阀门 {valve_id} 必须提供设置值",
)
control: dict[str, str] = {"status": status}
if status == "ACTIVE":
control["setting"] = setting
valve_control[valve_id] = control
elif valves_k is not None:
if len(valves) != len(valves_k):
raise HTTPException(
status_code=422, detail="valves 和 valves_k 的数量必须一致"
)
valve_opening = {
valve_id: float(valves_k[idx]) for idx, valve_id in enumerate(valves)
valve_id: float(valve_k)
for valve_id, valve_k in zip(valves, valves_k)
}
else:
raise HTTPException(
status_code=422,
detail="提供 valves 时必须同时提供 valve_statuses 或 valves_k",
)
result = flushing_analysis(
name=network,
modify_pattern_start_time=start_time,
modify_total_duration=duration or 900,
modify_valve_opening=valve_opening,
valve_control=valve_control,
drainage_node_ID=drainage_node_ID,
flushing_flow=flush_flow,
scheme_name=scheme_name,
+47 -4
View File
@@ -14,9 +14,19 @@ from pydantic import BaseModel, JsonValue, create_model
from starlette.responses import Response
from app.api.problem_details import ProblemDetails
from app.api.pagination import PaginatedList
from app.api.v1.router import api_router as handler_api_router
from app.auth.metadata_dependencies import get_current_metadata_user
from app.auth.project_dependencies import ProjectContext, get_project_context
from app.auth.project_dependencies import (
ProjectContext,
get_project_business_routing,
get_project_context,
get_project_simulation_routing,
)
from app.infra.db.project_routing import (
ActiveProjectRouting,
activate_project_routing,
)
T = TypeVar("T")
@@ -41,8 +51,15 @@ _PUBLIC_PARAMETER_RENAMES = {
"burst_ID": "burst_id",
"drainage_node_ID": "drainage_node_id",
}
_MODEL_NAME_IS_NETWORK = {"RunSimulationManuallyByDate"}
_MODEL_USERNAME_FROM_AUTH: set[str] = set()
_MODEL_NAME_IS_NETWORK = {"RunSimulationManuallyByDate", "PressureSensorPlacement"}
_MODEL_USERNAME_FROM_AUTH = {"PressureSensorPlacement"}
_TIMESCALE_ROUTED_ENDPOINT_MODULES = {
"app.api.v1.endpoints.burst_detection",
"app.api.v1.endpoints.burst_location",
"app.api.v1.endpoints.leakage",
"app.api.v1.endpoints.simulation",
}
_TIMESCALE_ROUTED_ENDPOINT_NAMES = {"open_project_endpoint"}
def _clean_name(name: str) -> str:
@@ -127,10 +144,14 @@ def _with_header_project_context(endpoint, route_name: str):
None,
)
injected_context_name = existing_context_parameter or "_rest_project_context"
injected_routing_name = "_rest_project_routing"
injected_user_name = "_rest_current_user"
@wraps(endpoint)
async def wrapper(*args, **kwargs):
project_routing = kwargs.pop(injected_routing_name, None)
if not isinstance(project_routing, ActiveProjectRouting):
raise RuntimeError("REST project database routing was not resolved")
project_context = kwargs.get(injected_context_name)
if not isinstance(project_context, ProjectContext):
raise RuntimeError("REST project context was not resolved")
@@ -161,6 +182,7 @@ def _with_header_project_context(endpoint, route_name: str):
kwargs[parameter_name] = original_model.model_validate(data)
if model_has_username:
kwargs.pop(injected_user_name, None)
with activate_project_routing(project_routing):
result = endpoint(*args, **kwargs)
if inspect.isawaitable(result):
return await result
@@ -180,6 +202,14 @@ def _with_header_project_context(endpoint, route_name: str):
if name in body_models:
parameter = parameter.replace(annotation=body_models[name][1])
parameters.append(parameter)
routing_dependency = (
get_project_simulation_routing
if (
endpoint.__module__ in _TIMESCALE_ROUTED_ENDPOINT_MODULES
or endpoint.__name__ in _TIMESCALE_ROUTED_ENDPOINT_NAMES
)
else get_project_business_routing
)
if not existing_context_parameter:
parameters.append(
inspect.Parameter(
@@ -189,6 +219,14 @@ def _with_header_project_context(endpoint, route_name: str):
default=Depends(get_project_context),
)
)
parameters.append(
inspect.Parameter(
injected_routing_name,
kind=inspect.Parameter.KEYWORD_ONLY,
annotation=ActiveProjectRouting,
default=Depends(routing_dependency),
)
)
if username_parameter or model_has_username:
parameters.append(
inspect.Parameter(
@@ -230,9 +268,14 @@ def _with_pagination(endpoint):
if not isinstance(result, list):
return result
if handler_handles_pagination:
if not isinstance(result, PaginatedList):
raise RuntimeError(
f"Paginated handler {endpoint.__name__!r} must return "
"PaginatedList with the real total"
)
return Page(
items=result,
total=offset + len(result),
total=result.total,
limit=limit or len(result),
offset=offset,
)
-6
View File
@@ -7,7 +7,6 @@ from app.api.v1.endpoints import (
audit,
burst_detection,
burst_location,
cache,
extension,
geocoding,
leakage,
@@ -166,11 +165,6 @@ api_router.include_router(
tags=["Risk"],
dependencies=[risk_run_access],
)
api_router.include_router(
cache.router,
tags=["Cache"],
dependencies=[simulation_run_access],
)
api_router.include_router(
web_search.router,
tags=["Web Search"],
+57
View File
@@ -16,6 +16,7 @@ from app.infra.db.metadb.repositories.metadata_repository import (
MetadataRepository,
ProjectDbRouting,
)
from app.infra.db.project_routing import ActiveProjectRouting
DB_ROLE_BIZ_DATA = "biz_data"
DB_ROLE_IOT_DATA = "iot_data"
@@ -99,6 +100,62 @@ async def get_project_context(
return await resolve_project_context(x_project_id, current_user, metadata_repo)
async def get_project_business_routing(
ctx: ProjectContext = Depends(get_project_context),
metadata_repo: MetadataRepository = Depends(get_metadata_repository),
) -> ActiveProjectRouting:
return await resolve_project_business_routing(ctx, metadata_repo)
async def resolve_project_business_routing(
ctx: ProjectContext,
metadata_repo: MetadataRepository,
) -> ActiveProjectRouting:
business = await _get_project_routing(
metadata_repo,
ctx.project_id,
DB_ROLE_BIZ_DATA,
DB_TYPE_POSTGRES,
"PostgreSQL",
)
return ActiveProjectRouting(
project_code=ctx.project_code,
business_dsn=business.dsn,
)
async def get_project_simulation_routing(
ctx: ProjectContext = Depends(get_project_context),
metadata_repo: MetadataRepository = Depends(get_metadata_repository),
) -> ActiveProjectRouting:
return await resolve_project_simulation_routing(ctx, metadata_repo)
async def resolve_project_simulation_routing(
ctx: ProjectContext,
metadata_repo: MetadataRepository,
) -> ActiveProjectRouting:
business = await _get_project_routing(
metadata_repo,
ctx.project_id,
DB_ROLE_BIZ_DATA,
DB_TYPE_POSTGRES,
"PostgreSQL",
)
timescale = await _get_project_routing(
metadata_repo,
ctx.project_id,
DB_ROLE_IOT_DATA,
DB_TYPE_TIMESCALE,
"TimescaleDB",
)
return ActiveProjectRouting(
project_code=ctx.project_code,
business_dsn=business.dsn,
timescale_dsn=timescale.dsn,
)
async def _get_project_routing(
metadata_repo: MetadataRepository,
project_id: UUID,
+1
View File
@@ -130,6 +130,7 @@ def sanitize_sensitive_data(data: dict) -> dict:
"token",
"api_key",
"apikey",
"dsn",
"credit_card",
"ssn",
"social_security",
-6
View File
@@ -26,12 +26,6 @@ class Settings(BaseSettings):
TIMESCALEDB_DB_PORT: str = "5433"
TIMESCALEDB_DB_USER: str = "postgres"
TIMESCALEDB_DB_PASSWORD: str = "password"
# InfluxDB
INFLUXDB_URL: str = "http://localhost:8086"
INFLUXDB_TOKEN: str = "token"
INFLUXDB_ORG: str = "org"
INFLUXDB_BUCKET: str = "bucket"
# Metadata Database Config (PostgreSQL)
METADATA_DB_NAME: str = "system_hub"
METADATA_DB_HOST: str = "localhost"
View File
-19
View File
@@ -1,19 +0,0 @@
import redis
import msgpack
from datetime import datetime
from typing import Any
# Initialize Redis connection
redis_client = redis.Redis(host="127.0.0.1", port=6379, db=0)
def encode_datetime(obj: Any) -> Any:
"""Serialize datetime objects to dictionary format."""
if isinstance(obj, datetime):
return {"__datetime__": True, "as_str": obj.strftime("%Y%m%dT%H:%M:%S.%f")}
return obj
def decode_datetime(obj: Any) -> Any:
"""Deserialize dictionary format to datetime objects."""
if "__datetime__" in obj:
return datetime.strptime(obj["as_str"], "%Y%m%dT%H:%M:%S.%f")
return obj
View File
File diff suppressed because it is too large Load Diff
-6
View File
@@ -1,6 +0,0 @@
# influxdb数据库连接信息
url = "http://127.0.0.1:8086" # 替换为你的InfluxDB实例地址
token = "kMPX2V5HsbzPpUT2B9HPBu1sTG1Emf-lPlT2UjxYnGAuocpXq_f_0lK4HHs-TbbKyjsZpICkMsyXG_V2D7P7yQ==" # 替换为你的InfluxDB Token
# _ENCODED_TOKEN = "eEdETTVSWnFSSkF1ekFHUy1vdFhVZEMyTkZkWTc1cUpBalJMcUFCNHA1V2NJSUFsSVVwT3BUOF95QTE2QU9IbUpXZXJ3UV8wOGd3Yjg0c3k0MmpuWlE9PQ=="
# token = base64.b64decode(_ENCODED_TOKEN).decode("utf-8")
org = "TJWATERORG" # 替换为你的Organization名称
-33
View File
@@ -1,33 +0,0 @@
from influxdb_client import InfluxDBClient, Point, WriteOptions
from influxdb_client.client.query_api import QueryApi
import influxdb_info
# 配置 InfluxDB 连接
url = influxdb_info.url
token = influxdb_info.token
org = influxdb_info.org
bucket = "SCADA_data"
# 创建 InfluxDB 客户端
client = InfluxDBClient(url=url, token=token, org=org)
# 创建查询 API 对象
query_api = client.query_api()
# 构建查询语句
query = f'''
from(bucket: "{bucket}")
|> range(start: -1h)
'''
# 执行查询
result = query_api.query(query)
print(result)
# 处理查询结果
for table in result:
for record in table.records:
print(f"Time: {record.get_time()}, Value: {record.get_value()}, Measurement: {record.get_measurement()}, Field: {record.get_field()}")
# 关闭客户端连接
client.close()
@@ -20,14 +20,17 @@ def _normalize_postgres_dsn(dsn: str) -> str:
scheme, rest = dsn.split("://", 1)
if scheme not in ("postgresql", "postgres", "postgresql+psycopg"):
return dsn
if scheme == "postgresql+psycopg":
scheme = "postgresql"
normalized_dsn = f"{scheme}://{rest}"
if "@" not in rest:
return dsn
return normalized_dsn
userinfo, hostinfo = rest.rsplit("@", 1)
if ":" not in userinfo:
return dsn
return normalized_dsn
username, password = userinfo.split(":", 1)
if "@" not in password:
return dsn
return normalized_dsn
password = password.replace("@", "%40")
return f"{scheme}://{username}:{password}@{hostinfo}"
+11 -3
View File
@@ -3,7 +3,7 @@ from contextlib import asynccontextmanager
from typing import AsyncGenerator, Dict, Optional
import psycopg_pool
from psycopg.rows import dict_row
import app.core.config as postgresql_info
from app.infra.db.project_routing import get_project_pgconn_string
# Configure logging
logger = logging.getLogger(__name__)
@@ -13,6 +13,7 @@ class Database:
def __init__(self, db_name=None):
self.pool = None
self.db_name = db_name
self.conninfo = None
def init_pool(self, db_name=None):
"""Initialize the connection pool."""
@@ -21,9 +22,10 @@ class Database:
# Get connection string, handling default case where target_db_name might be None
if target_db_name:
conn_string = postgresql_info.get_pgconn_string(db_name=target_db_name)
conn_string = get_project_pgconn_string(db_name=target_db_name)
else:
conn_string = postgresql_info.get_pgconn_string()
conn_string = get_project_pgconn_string()
self.conninfo = conn_string
try:
self.pool = psycopg_pool.AsyncConnectionPool(
@@ -75,6 +77,12 @@ async def get_database_instance(db_name: Optional[str] = None) -> Database:
if not db_name:
return db # 返回默认数据库实例
expected_conninfo = get_project_pgconn_string(db_name=db_name)
existing = _database_instances.get(db_name)
if existing is not None and existing.conninfo != expected_conninfo:
await existing.close()
del _database_instances[db_name]
if db_name not in _database_instances:
# 创建新的数据库实例
instance = create_database_instance(db_name)
+65
View File
@@ -0,0 +1,65 @@
from __future__ import annotations
from contextlib import contextmanager
from contextvars import ContextVar, Token
from dataclasses import dataclass
from typing import Iterator
from psycopg.conninfo import make_conninfo
from app.core.config import get_pgconn_string, get_timescaledb_pgconn_string
@dataclass(frozen=True)
class ActiveProjectRouting:
project_code: str
business_dsn: str
timescale_dsn: str | None = None
_active_project_routing: ContextVar[ActiveProjectRouting | None] = ContextVar(
"active_project_routing",
default=None,
)
def get_active_project_routing() -> ActiveProjectRouting | None:
return _active_project_routing.get()
@contextmanager
def activate_project_routing(
routing: ActiveProjectRouting,
) -> Iterator[ActiveProjectRouting]:
token: Token[ActiveProjectRouting | None] = _active_project_routing.set(routing)
try:
yield routing
finally:
_active_project_routing.reset(token)
def _dsn_for_database(dsn: str, database_name: str) -> str:
return make_conninfo(dsn, dbname=database_name)
def get_project_pgconn_string(db_name: str | None = None) -> str:
routing = get_active_project_routing()
if routing is None:
return get_pgconn_string(db_name=db_name)
if db_name is None or db_name == routing.project_code:
return routing.business_dsn
return _dsn_for_database(routing.business_dsn, db_name)
def get_project_timescale_pgconn_string(db_name: str | None = None) -> str:
routing = get_active_project_routing()
if routing is None:
return get_timescaledb_pgconn_string(db_name=db_name)
if routing.timescale_dsn is None:
raise RuntimeError(
f"TimescaleDB routing is not configured for project {routing.project_code}"
)
# Legacy simulation code used to derive the Timescale database name from
# the project code. Project-scoped requests must instead use the complete
# iot_data DSN selected by metadata routing.
return routing.timescale_dsn
+13 -5
View File
@@ -3,7 +3,7 @@ from contextlib import asynccontextmanager
from typing import AsyncGenerator, Dict, Optional
import psycopg_pool
from psycopg.rows import dict_row
from app.core.config import get_timescaledb_pgconn_string
from app.infra.db.project_routing import get_project_timescale_pgconn_string
# Configure logging
logger = logging.getLogger(__name__)
@@ -13,6 +13,7 @@ class Database:
def __init__(self, db_name=None):
self.pool = None
self.db_name = db_name
self.conninfo = None
def init_pool(self, db_name=None):
"""Initialize the connection pool."""
@@ -21,9 +22,10 @@ class Database:
# Get connection string, handling default case where target_db_name might be None
if target_db_name:
conn_string = get_timescaledb_pgconn_string(db_name=target_db_name)
conn_string = get_project_timescale_pgconn_string(db_name=target_db_name)
else:
conn_string = get_timescaledb_pgconn_string()
conn_string = get_project_timescale_pgconn_string()
self.conninfo = conn_string
try:
self.pool = psycopg_pool.AsyncConnectionPool(
@@ -54,8 +56,8 @@ class Database:
"""Get the TimescaleDB connection string."""
target_db_name = db_name or self.db_name
if target_db_name:
return get_timescaledb_pgconn_string(db_name=target_db_name)
return get_timescaledb_pgconn_string()
return get_project_timescale_pgconn_string(db_name=target_db_name)
return get_project_timescale_pgconn_string()
@asynccontextmanager
async def get_connection(self) -> AsyncGenerator:
@@ -84,6 +86,12 @@ async def get_database_instance(db_name: Optional[str] = None) -> Database:
if not db_name:
return db # 返回默认数据库实例
expected_conninfo = get_project_timescale_pgconn_string(db_name=db_name)
existing = _database_instances.get(db_name)
if existing is not None and existing.conninfo != expected_conninfo:
await existing.close()
del _database_instances[db_name]
if db_name not in _database_instances:
# 创建新的数据库实例
instance = create_database_instance(db_name)
+13 -13
View File
@@ -6,7 +6,7 @@ import psycopg
from psycopg import sql
from psycopg.rows import dict_row
import time
from app.core.config import get_timescaledb_pgconn_string
from app.infra.db.project_routing import get_project_timescale_pgconn_string
from app.infra.db.timescaledb.repositories.scheme import SchemeRepository
from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository
from app.infra.db.timescaledb.repositories.scada import ScadaRepository
@@ -26,9 +26,9 @@ class InternalStorage:
for attempt in range(max_retries):
try:
conn_string = (
get_timescaledb_pgconn_string(db_name=db_name)
get_project_timescale_pgconn_string(db_name=db_name)
if db_name
else get_timescaledb_pgconn_string()
else get_project_timescale_pgconn_string()
)
with psycopg.Connection.connect(conn_string) as conn:
RealtimeRepository.store_realtime_simulation_result_sync(
@@ -58,9 +58,9 @@ class InternalStorage:
for attempt in range(max_retries):
try:
conn_string = (
get_timescaledb_pgconn_string(db_name=db_name)
get_project_timescale_pgconn_string(db_name=db_name)
if db_name
else get_timescaledb_pgconn_string()
else get_project_timescale_pgconn_string()
)
with psycopg.Connection.connect(conn_string) as conn:
SchemeRepository.store_scheme_simulation_result_sync(
@@ -99,9 +99,9 @@ class InternalQueries:
for attempt in range(max_retries):
try:
conn_string = (
get_timescaledb_pgconn_string(db_name=db_name)
get_project_timescale_pgconn_string(db_name=db_name)
if db_name
else get_timescaledb_pgconn_string()
else get_project_timescale_pgconn_string()
)
with psycopg.Connection.connect(conn_string) as conn:
rows = ScadaRepository.get_scada_by_ids_time_range_sync(
@@ -140,9 +140,9 @@ class InternalQueries:
for attempt in range(max_retries):
try:
conn_string = (
get_timescaledb_pgconn_string(db_name=db_name)
get_project_timescale_pgconn_string(db_name=db_name)
if db_name
else get_timescaledb_pgconn_string()
else get_project_timescale_pgconn_string()
)
with psycopg.Connection.connect(conn_string) as conn:
rows = ScadaRepository.get_scada_by_ids_time_range_sync(
@@ -185,9 +185,9 @@ class InternalQueries:
for attempt in range(max_retries):
try:
conn_string = (
get_timescaledb_pgconn_string(db_name=db_name)
get_project_timescale_pgconn_string(db_name=db_name)
if db_name
else get_timescaledb_pgconn_string()
else get_project_timescale_pgconn_string()
)
with psycopg.Connection.connect(conn_string) as conn:
return ScadaRepository.get_latest_scada_time_sync(
@@ -286,9 +286,9 @@ class InternalQueries:
for attempt in range(max_retries):
try:
conn_string = (
get_timescaledb_pgconn_string(db_name=db_name)
get_project_timescale_pgconn_string(db_name=db_name)
if db_name
else get_timescaledb_pgconn_string()
else get_project_timescale_pgconn_string()
)
with psycopg.Connection.connect(conn_string) as conn:
with conn.cursor(row_factory=dict_row) as cur:
+17 -5
View File
@@ -4,9 +4,10 @@ from threading import RLock
import psycopg as pg
from app.core.config import get_pgconn_string
from app.infra.db.project_routing import get_project_pgconn_string
g_conn_dict: dict[str, pg.Connection] = {}
g_conninfo_dict: dict[str, str] = {}
_registry_lock = RLock()
_project_locks: dict[str, RLock] = {}
@@ -42,14 +43,18 @@ def _get_project_lock(name: str) -> RLock:
def open_connection(name: str) -> pg.Connection:
with _get_project_lock(name):
conninfo = get_project_pgconn_string(db_name=name)
connection = g_conn_dict.get(name)
if connection is None or not _is_healthy(connection):
if (
connection is None
or g_conninfo_dict.get(name) != conninfo
or not _is_healthy(connection)
):
if connection is not None:
_close_connection(connection)
connection = pg.connect(
conninfo=get_pgconn_string(db_name=name), autocommit=True
)
connection = pg.connect(conninfo=conninfo, autocommit=True)
g_conn_dict[name] = connection
g_conninfo_dict[name] = conninfo
return connection
@@ -60,6 +65,12 @@ def is_connection_open(name: str) -> bool:
return False
if not _is_healthy(connection):
del g_conn_dict[name]
g_conninfo_dict.pop(name, None)
_close_connection(connection)
return False
if g_conninfo_dict.get(name) != get_project_pgconn_string(db_name=name):
del g_conn_dict[name]
g_conninfo_dict.pop(name, None)
_close_connection(connection)
return False
return True
@@ -68,6 +79,7 @@ def is_connection_open(name: str) -> bool:
def close_connection(name: str) -> None:
with _get_project_lock(name):
connection = g_conn_dict.pop(name, None)
g_conninfo_dict.pop(name, None)
if connection is not None:
_close_connection(connection)
+18 -6
View File
@@ -1,3 +1,4 @@
from collections.abc import Mapping, Sequence
from typing import Any
from psycopg.rows import dict_row, Row
from .connection import project_connection
@@ -82,27 +83,38 @@ class DbChangeSet:
return DbChangeSet(redo_sql, undo_sql, redo_cs_s, undo_cs_s)
def read(name: str, sql: str) -> Row:
QueryParams = Sequence[Any] | Mapping[str, Any]
def _execute(cur, sql: str, params: QueryParams | None = None):
return cur.execute(sql, params) if params is not None else cur.execute(sql)
def read(name: str, sql: str, params: QueryParams | None = None) -> Row:
with project_connection(name) as conn:
with conn.cursor(row_factory=dict_row) as cur:
cur.execute(sql)
_execute(cur, sql, params)
row = cur.fetchone()
if row == None:
raise Exception(sql)
return row
def read_all(name: str, sql: str) -> list[Row]:
def read_all(
name: str, sql: str, params: QueryParams | None = None
) -> list[Row]:
with project_connection(name) as conn:
with conn.cursor(row_factory=dict_row) as cur:
cur.execute(sql)
_execute(cur, sql, params)
return cur.fetchall()
def try_read(name: str, sql: str) -> Row | None:
def try_read(
name: str, sql: str, params: QueryParams | None = None
) -> Row | None:
with project_connection(name) as conn:
with conn.cursor(row_factory=dict_row) as cur:
cur.execute(sql)
_execute(cur, sql, params)
return cur.fetchone()
+11 -8
View File
@@ -7,7 +7,8 @@ from .connection import (
is_connection_open,
open_connection,
)
from app.core.config import get_pgconn_string, get_pg_config, get_pg_password
from app.core.config import get_pg_config, get_pg_password
from app.infra.db.project_routing import get_project_pgconn_string
# no undo/redo
@@ -16,7 +17,7 @@ _server_databases = ["template0", "template1", "postgres", "project"]
def list_project() -> list[str]:
ps = []
with pg.connect(conninfo=get_pgconn_string(), autocommit=True) as conn:
with pg.connect(conninfo=get_project_pgconn_string(), autocommit=True) as conn:
with conn.cursor(row_factory=dict_row) as cur:
for p in cur.execute(
f"select datname from pg_database where datname <> 'postgres' and datname <> 'template0' and datname <> 'template1' and datname <> 'project'"
@@ -27,7 +28,7 @@ def list_project() -> list[str]:
def have_project(name: str) -> bool:
with pg.connect(
conninfo=get_pgconn_string(db_name="postgres"), autocommit=True
conninfo=get_project_pgconn_string(db_name="postgres"), autocommit=True
) as conn:
with conn.cursor() as cur:
cur.execute("select 1 from pg_database where datname = %s", (name,))
@@ -38,7 +39,7 @@ def copy_project(source: str, new: str) -> None:
close_connection(source)
with pg.connect(
conninfo=get_pgconn_string(db_name="postgres"), autocommit=True
conninfo=get_project_pgconn_string(db_name="postgres"), autocommit=True
) as admin_conn:
with admin_conn.cursor() as cur:
cur.execute(
@@ -131,7 +132,9 @@ class CopyProjectEx:
connection.commit()
def __call__(self, source: str, new_db: str, excluded_tables: [str] = None) -> None:
source_connection = pg.connect(conninfo=get_pgconn_string(), autocommit=True)
source_connection = pg.connect(
conninfo=get_project_pgconn_string(), autocommit=True
)
self.create_database(source_connection, new_db)
@@ -140,7 +143,7 @@ class CopyProjectEx:
source_connection.close()
new_db_connection = pg.connect(
conninfo=get_pgconn_string(db_name=new_db), autocommit=True
conninfo=get_project_pgconn_string(db_name=new_db), autocommit=True
)
self.init_operation_table(new_db_connection, excluded_tables)
new_db_connection.close()
@@ -151,7 +154,7 @@ def create_project(name: str) -> None:
def delete_project(name: str) -> None:
with pg.connect(conninfo=get_pgconn_string(), autocommit=True) as conn:
with pg.connect(conninfo=get_project_pgconn_string(), autocommit=True) as conn:
with conn.cursor() as cur:
cur.execute(
f"select pg_terminate_backend(pid) from pg_stat_activity where datname = '{name}'"
@@ -161,7 +164,7 @@ def delete_project(name: str) -> None:
def clean_project(excluded: list[str] = []) -> None:
projects = list_project()
with pg.connect(conninfo=get_pgconn_string(), autocommit=True) as conn:
with pg.connect(conninfo=get_project_pgconn_string(), autocommit=True) as conn:
with conn.cursor(row_factory=dict_row) as cur:
row = cur.execute(f"select current_database()").fetchone()
if row != None:
+20 -9
View File
@@ -1,3 +1,4 @@
from psycopg import sql
from psycopg.rows import dict_row, Row
from .connection import project_connection
from .database import read
@@ -49,7 +50,12 @@ ELEMENT_TYPES : dict[str, int] = {
def _get_from(name: str, id: str, base_type: str) -> Row | None:
with project_connection(name) as conn:
with conn.cursor(row_factory=dict_row) as cur:
cur.execute(f"select * from {base_type} where id = '{id}'")
cur.execute(
sql.SQL("select * from {} where id = %s").format(
sql.Identifier(base_type)
),
(id,),
)
return cur.fetchone()
@@ -243,11 +249,17 @@ def get_node_links(name: str, id: str) -> list[str]:
with project_connection(name) as conn:
with conn.cursor(row_factory=dict_row) as cur:
links: list[str] = []
for p in cur.execute(f"select id from pipes where node1 = '{id}' or node2 = '{id}'").fetchall():
for p in cur.execute(
"select id from pipes where node1 = %s or node2 = %s", (id, id)
).fetchall():
links.append(p['id'])
for p in cur.execute(f"select id from pumps where node1 = '{id}' or node2 = '{id}'").fetchall():
for p in cur.execute(
"select id from pumps where node1 = %s or node2 = %s", (id, id)
).fetchall():
links.append(p['id'])
for p in cur.execute(f"select id from valves where node1 = '{id}' or node2 = '{id}'").fetchall():
for p in cur.execute(
"select id from valves where node1 = %s or node2 = %s", (id, id)
).fetchall():
links.append(p['id'])
return links
@@ -255,16 +267,15 @@ def get_node_links(name: str, id: str) -> list[str]:
def get_link_nodes(name: str, id: str) -> list[str]:
row = {}
if is_pipe(name, id):
row = read(name, f"select node1, node2 from pipes where id = '{id}'")
row = read(name, "select node1, node2 from pipes where id = %s", (id,))
elif is_pump(name, id):
row = read(name, f"select node1, node2 from pumps where id = '{id}'")
row = read(name, "select node1, node2 from pumps where id = %s", (id,))
elif is_valve(name, id):
row = read(name, f"select node1, node2 from valves where id = '{id}'")
row = read(name, "select node1, node2 from valves where id = %s", (id,))
return [str(row['node1']), str(row['node2'])]
def get_region_type(name: str, id: str)->str:
if(is_region(name,id)):
type = read(name, f"select type from _region where id = '{id}'")
type = read(name, "select type from _region where id = %s", (id,))
return type
+8 -2
View File
@@ -23,7 +23,11 @@ def from_postgis_point(coord: str) -> dict[str, float]:
def get_node_coord(name: str, node: str) -> dict[str, float]:
row = try_read(name, f"select st_astext(coord) as coord_geom from coordinates where node = '{node}'")
row = try_read(
name,
"select st_astext(coord) as coord_geom from coordinates where node = %s",
(node,),
)
if row == None:
write(name, sql_insert_coord(node, 0.0, 0.0))
return {'x': 0.0, 'y': 0.0}
@@ -66,7 +70,9 @@ def get_links_in_extent(name: str, x1: float, y1: float, x2: float, y2: float) -
def node_has_coord(name: str, node: str) -> bool:
return try_read(name, f"select node from coordinates where node = '{node}'") != None
return try_read(
name, "select node from coordinates where node = %s", (node,)
) != None
#--------------------------------------------------------------
+1 -1
View File
@@ -12,7 +12,7 @@ def get_junction_schema(name: str) -> dict[str, dict[str, Any]]:
def get_junction(name: str, id: str) -> dict[str, Any]:
j = try_read(name, f"select * from junctions where id = '{id}'")
j = try_read(name, "select * from junctions where id = %s", (id,))
if j == None:
return {}
xy = get_node_coord(name, id)
-1
View File
@@ -23,7 +23,6 @@ non_realtime_region_patterns = {} # 基于source_outflow_region进行区分
realtime_region_pipe_flow_and_demand_id = {} # 基于source_outflow_region搜索该分区中的实时pipe_flow和demand的api_query_id,后续用region的流量 - 实时流量计的流量
realtime_region_pipe_flow_and_demand_patterns = {} # 基于source_outflow_region搜索该分区中的实时pipe_flow和demand的associated_pattern,后续用region的流量 - 实时流量计的流量
# ---------------------------------------------------------
# influxdb_api.py中的全局变量
# 全局变量,用于存储不同类型的realtime api_query_id
reservoir_liquid_level_realtime_ids = []
tank_liquid_level_realtime_ids = []
+6 -7
View File
@@ -5,8 +5,7 @@ import chardet
import psycopg
from psycopg import sql
import app.services.project_info as project_info
from app.core.config import get_pgconn_string
from app.infra.db.project_routing import get_project_pgconn_string
from app.services.tjnetwork import read_inp
@@ -15,13 +14,14 @@ from app.services.tjnetwork import read_inp
############################################################
def network_update(file_path: str) -> None:
def network_update(file_path: str, project_code: str) -> None:
"""
更新pg数据库中的inp文件
:param file_path: inp文件
:param project_code: 元数据项目代码
:return:
"""
read_inp("szh", file_path)
read_inp(project_code, file_path)
csv_path = "./history_pattern_flow.csv"
@@ -51,8 +51,7 @@ def network_update(file_path: str) -> None:
if os.path.exists(csv_path):
print(f"history_patterns_flows文件存在,开始处理...")
# 连接到 PostgreSQL 数据库(这里是数据库 "bb"
with psycopg.connect(f"dbname={project_info.name} host=127.0.0.1") as conn:
with psycopg.connect(get_project_pgconn_string(project_code)) as conn:
with conn.cursor() as cur:
with open(csv_path, newline="", encoding="utf-8-sig") as csvfile:
reader = csv.DictReader(csvfile)
@@ -92,7 +91,7 @@ def submit_scada_info(name: str, coord_id: str) -> None:
print(f"检测到的文件编码:{file_encoding}")
try:
# 动态替换数据库名称
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
# 连接到 PostgreSQL 数据库(这里是数据库 "bb"
with psycopg.connect(conn_string) as conn:
+15 -15
View File
@@ -7,7 +7,7 @@ import pandas as pd
import psycopg
from sqlalchemy import create_engine
from app.core.config import get_pgconn_string
from app.infra.db.project_routing import get_project_pgconn_string
from app.services.time_api import parse_utc_time
@@ -20,7 +20,7 @@ def scheme_name_exists(name: str, scheme_name: str) -> bool:
:return: 如果存在返回 True否则返回 False
"""
try:
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
cur.execute(
@@ -57,7 +57,7 @@ def store_scheme_info(
:return:
"""
try:
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
sql = """
@@ -93,7 +93,7 @@ def delete_scheme_info(name: str, scheme_name: str) -> None:
:param scheme_name: 要删除的方案名称
"""
try:
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
# 使用参数化查询删除方案记录
@@ -121,7 +121,7 @@ def query_scheme_list(
"""
try:
# 动态替换数据库名称
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
# 连接到 PostgreSQL 数据库(这里是数据库 "bb"
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
@@ -203,7 +203,7 @@ def query_scheme_detail(
scheme_type,
)
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
if scheme_type:
@@ -255,7 +255,7 @@ def store_leakage_identify_result(
run_status: str = "completed",
error_message: str | None = None,
) -> None:
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
cur.execute(
@@ -299,7 +299,7 @@ def query_leakage_identify_schemes(
scheme_type: str = "dma_leak_identification",
query_date: date | None = None,
) -> list[dict]:
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
if query_date is None:
@@ -343,7 +343,7 @@ def query_leakage_identify_schemes(
def query_leakage_identify_scheme_detail(name: str, scheme_name: str) -> dict:
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
cur.execute(
@@ -400,7 +400,7 @@ def query_burst_location_schemes(
scheme_type: str = "burst_location",
query_date: date | None = None,
) -> list[dict]:
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
if query_date is None:
@@ -444,7 +444,7 @@ def query_burst_location_schemes(
def query_burst_location_scheme_detail(name: str, scheme_name: str) -> dict:
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
cur.execute(
@@ -479,7 +479,7 @@ def query_burst_detection_schemes(
scheme_type: str = "burst_detection",
query_date: date | None = None,
) -> list[dict]:
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
if query_date is None:
@@ -523,7 +523,7 @@ def query_burst_detection_schemes(
def query_burst_detection_scheme_detail(name: str, scheme_name: str) -> dict:
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
cur.execute(
@@ -564,7 +564,7 @@ def upload_shp_to_pg(name: str, table_name: str, role: str, shp_file_path: str):
"""
try:
# 动态连接到指定的数据库
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
with psycopg.connect(conn_string) as conn:
# 读取 Shapefile 文件
gdf = gpd.read_file(shp_file_path)
@@ -604,7 +604,7 @@ def submit_risk_probability_result(name: str, result_file_path: str) -> None:
try:
# 动态替换数据库名称
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
# 连接到 PostgreSQL 数据库
with psycopg.connect(conn_string) as conn:
+39 -13
View File
@@ -28,14 +28,13 @@ import pytz
import requests
import time
from typing import Optional, Tuple
import app.infra.db.influxdb.api as influxdb_api
import typing
import psycopg
import logging
import app.services.globals as globals
import app.services.project_info as project_info
from app.services.time_api import parse_beijing_time, parse_clock_duration_seconds
from app.core.config import get_pgconn_string
from app.infra.db.project_routing import get_project_pgconn_string
from app.infra.db.timescaledb.internal_queries import (
InternalQueries as TimescaleInternalQueries,
)
@@ -55,7 +54,7 @@ def query_corresponding_element_id_and_query_id(name: str) -> None:
:return:
"""
# 连接数据库
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
try:
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
@@ -100,7 +99,7 @@ def query_corresponding_pattern_id_and_query_id(name: str) -> None:
:return:
"""
# 连接数据库
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
try:
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
@@ -140,7 +139,7 @@ def query_non_realtime_region(name: str) -> dict:
"""
source_outflow_regions = [] # 用于存储所有 region(包含重复的)
# 构建连接字符串
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
try:
# 连接到数据库
with psycopg.connect(conn_string) as conn:
@@ -217,7 +216,7 @@ def query_non_realtime_region_patterns(
region_tuple_to_key = {
frozenset(ids): region for region, ids in globals.source_outflow_region.items()
}
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
try:
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
@@ -305,7 +304,7 @@ def query_realtime_region_pipe_flow_and_demand_id(
region_tuple_to_key = {
frozenset(ids): region for region, ids in globals.source_outflow_region.items()
}
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
try:
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
@@ -377,7 +376,7 @@ def query_pipe_flow_region_patterns(
:param column_prefix: 需要提取的列的前缀
:return: pipe_flow_region_patterns 字典
"""
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
try:
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
@@ -440,7 +439,7 @@ def query_SCADA_ID_corresponding_info(name: str, SCADA_ID: str) -> dict:
:param SCADA_ID: SCADA设备的ID
:return: 包含associated_element_id和api_query_id的字典
"""
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
try:
# 使用 psycopg.connect 创建连接
with psycopg.connect(conn_string) as conn:
@@ -496,7 +495,7 @@ def get_source_outflow_region_id(
"No associated_source_outflow_id found in source_outflow_region."
)
return globals.source_outflow_region_id
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
try:
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
@@ -553,7 +552,7 @@ def get_realtime_region_patterns(
globals.realtime_region_pipe_flow_and_demand_patterns = {
region: [] for region in globals.realtime_region_pipe_flow_and_demand_id.keys()
}
conn_string = get_pgconn_string(db_name=name)
conn_string = get_project_pgconn_string(db_name=name)
try:
with psycopg.connect(conn_string) as conn:
with conn.cursor() as cur:
@@ -686,6 +685,28 @@ def get_history_pattern_info(project_name, pattern_name):
return flow_list, factor_list
def _apply_valve_control(
project_name: str, valve_control: dict[str, dict]
) -> None:
"""Apply explicit valve status, setting, and opening controls."""
for valve_name, control in valve_control.items():
valve_status = get_status(project_name, valve_name)
if "status" in control:
valve_status["status"] = control["status"]
if "setting" in control:
valve_status["setting"] = control["setting"]
if "k" in control:
valve_k = control["k"]
if valve_k == 0:
valve_status["status"] = "CLOSED"
else:
valve_status["setting"] = 0.1036 * pow(valve_k, -3.105)
cs = ChangeSet()
cs.append(valve_status)
set_status(project_name, cs)
# 2025/01/11
def run_simulation(
name: str,
@@ -701,6 +722,7 @@ def run_simulation(
modify_valve_opening: dict[str, float] = None,
scheme_type: str = None,
scheme_name: str = None,
valve_control: dict[str, dict] = None,
) -> None:
"""
传入需要修改的参数改变数据库中对应位置的值然后计算返回结果
@@ -715,6 +737,7 @@ def run_simulation(
:param modify_fixed_pump_pattern: dict中包含多个水泵模式str为工频水泵的idlist为修改后的pattern
:param modify_variable_pump_pattern: dict中包含多个水泵模式str为变频水泵的idlist为修改后的pattern
:param modify_valve_opening: dict中包含多个阀门开启度str为阀门的idfloat为修改后的阀门开启度
:param valve_control: dict中可分别指定阀门的statussetting和k存在时优先于modify_valve_opening
:param scheme_type: 模拟方案类型
:param scheme_name模拟方案名称
:return:
@@ -1200,8 +1223,11 @@ def run_simulation(
cs = ChangeSet()
cs.append(pump_pattern)
set_pattern(name_c, cs)
# 修改阀门(valve)的状态setting和status
if modify_valve_opening:
# 显式阀门控制沿用 run_simulation_ex 的处理顺序和覆盖规则。
if valve_control is not None:
_apply_valve_control(name_c, valve_control)
# 保留原开度参数逻辑,兼容现有方案调用。
elif modify_valve_opening:
for valve_name in modify_valve_opening.keys():
if not np.isnan(modify_valve_opening[valve_name]):
valve_status = get_status(name_c, valve_name)
+1 -1
View File
@@ -3,7 +3,7 @@
"contracts": {
"server": {
"file": "server-v1.openapi.json",
"sha256": "b003a57c9b8c1a041b644363ef58afb8bfcede1fe076ce48a190d2a1ac915e3f"
"sha256": "ac9b6fac185dfd999f1791cba51eb482df17a427b361963250aafa5fb1a276b4"
}
}
}
+75 -419
View File
@@ -1859,7 +1859,7 @@
"title": "PressureRegulationRest",
"type": "object"
},
"PressureSensorPlacement": {
"PressureSensorPlacementRest": {
"properties": {
"min_diameter": {
"default": 0,
@@ -1867,11 +1867,6 @@
"title": "Min Diameter",
"type": "integer"
},
"name": {
"description": "管网名称(或数据库名称)",
"title": "Name",
"type": "string"
},
"scheme_name": {
"description": "方案名称",
"title": "Scheme Name",
@@ -1881,20 +1876,13 @@
"description": "传感器数量",
"title": "Sensor Number",
"type": "integer"
},
"username": {
"description": "用户名",
"title": "Username",
"type": "string"
}
},
"required": [
"name",
"scheme_name",
"sensor_number",
"username"
"sensor_number"
],
"title": "PressureSensorPlacement",
"title": "PressureSensorPlacementRest",
"type": "object"
},
"ProblemDetails": {
@@ -5239,97 +5227,6 @@
]
}
},
"/api/v1/all-redis": {
"delete": {
"description": "清空整个Redis数据库的所有缓存",
"operationId": "delete_all_redis",
"parameters": [
{
"in": "header",
"name": "X-Project-Id",
"required": true,
"schema": {
"title": "X-Project-Id",
"type": "string"
}
}
],
"responses": {
"204": {
"description": "Successful Response"
},
"401": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Authentication required"
},
"403": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Insufficient permission"
},
"404": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Resource not found"
},
"409": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Resource conflict"
},
"422": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Validation error"
},
"503": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Dependency unavailable"
}
},
"security": [
{
"OAuth2PasswordBearer": []
}
],
"summary": "清除所有缓存",
"tags": [
"Cache"
]
}
},
"/api/v1/all-scada-properties": {
"get": {
"description": "获取指定水网中所有SCADA点的属性信息",
@@ -11038,7 +10935,7 @@
},
"/api/v1/flushing-analyses": {
"post": {
"description": "高级版本的冲洗分析,支持同时开启多个阀门进行冲洗,指定排污节点,并设置固定的冲洗流量。返回纯文本格式的分析结果。",
"description": "高级版本的冲洗分析,支持按状态和设置值控制多个可选阀门,指定排污节点,并设置固定的冲洗流量。返回纯文本格式的分析结果。",
"operationId": "post_flushing_analyses",
"parameters": [
{
@@ -11053,31 +10950,92 @@
}
},
{
"description": "要开启的阀门ID列表",
"description": "参与控制的阀门ID列表(可选)",
"in": "query",
"name": "valves",
"required": true,
"required": false,
"schema": {
"description": "要开启的阀门ID列表",
"anyOf": [
{
"items": {
"type": "string"
},
"title": "Valves",
"type": "array"
},
{
"type": "null"
}
],
"description": "参与控制的阀门ID列表(可选)",
"title": "Valves"
}
},
{
"description": "对应各阀门的开度列表(0-1",
"description": "对应各阀门的开度列表(0-1,可选,与valves同时提供",
"in": "query",
"name": "valves_k",
"required": true,
"required": false,
"schema": {
"description": "对应各阀门的开度列表(0-1",
"anyOf": [
{
"items": {
"type": "number"
},
"title": "Valves K",
"type": "array"
},
{
"type": "null"
}
],
"description": "对应各阀门的开度列表(0-1,可选,与valves同时提供)",
"title": "Valves K"
}
},
{
"description": "对应各阀门的开关状态列表(OPEN、CLOSED或ACTIVE,可选)",
"in": "query",
"name": "valve_statuses",
"required": false,
"schema": {
"anyOf": [
{
"items": {
"enum": [
"OPEN",
"CLOSED",
"ACTIVE"
],
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"description": "对应各阀门的开关状态列表(OPEN、CLOSED或ACTIVE,可选)",
"title": "Valve Statuses"
}
},
{
"description": "对应各阀门的设置值列表(ACTIVE状态下必填)",
"in": "query",
"name": "valve_settings",
"required": false,
"schema": {
"anyOf": [
{
"items": {
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"description": "对应各阀门的设置值列表(ACTIVE状态下必填)",
"title": "Valve Settings"
}
},
{
@@ -24833,7 +24791,7 @@
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/PressureSensorPlacement",
"$ref": "#/components/schemas/PressureSensorPlacementRest",
"description": "传感器放置分析参数"
}
}
@@ -25073,7 +25031,7 @@
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/PressureSensorPlacement",
"$ref": "#/components/schemas/PressureSensorPlacementRest",
"description": "传感器放置分析参数"
}
}
@@ -29403,308 +29361,6 @@
]
}
},
"/api/v1/redis": {
"get": {
"description": "获取Redis中所有的缓存键",
"operationId": "get_redis",
"parameters": [
{
"in": "header",
"name": "X-Project-Id",
"required": true,
"schema": {
"title": "X-Project-Id",
"type": "string"
}
}
],
"responses": {
"200": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/JsonValue"
}
}
},
"description": "Successful Response"
},
"401": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Authentication required"
},
"403": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Insufficient permission"
},
"404": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Resource not found"
},
"409": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Resource conflict"
},
"422": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Validation error"
},
"503": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Dependency unavailable"
}
},
"security": [
{
"OAuth2PasswordBearer": []
}
],
"summary": "查询缓存键列表",
"tags": [
"Cache"
]
}
},
"/api/v1/redis-keys": {
"delete": {
"description": "根据模式清除匹配的Redis缓存键",
"operationId": "delete_redis_keys",
"parameters": [
{
"description": "缓存键模式(支持通配符)",
"in": "query",
"name": "keys",
"required": true,
"schema": {
"description": "缓存键模式(支持通配符)",
"title": "Keys",
"type": "string"
}
},
{
"in": "header",
"name": "X-Project-Id",
"required": true,
"schema": {
"title": "X-Project-Id",
"type": "string"
}
}
],
"responses": {
"204": {
"description": "Successful Response"
},
"401": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Authentication required"
},
"403": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Insufficient permission"
},
"404": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Resource not found"
},
"409": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Resource conflict"
},
"422": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Validation error"
},
"503": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Dependency unavailable"
}
},
"security": [
{
"OAuth2PasswordBearer": []
}
],
"summary": "清除匹配的缓存键",
"tags": [
"Cache"
]
}
},
"/api/v1/redis-keys/detail": {
"delete": {
"description": "根据键名清除单个Redis缓存",
"operationId": "delete_redis_keys_detail",
"parameters": [
{
"description": "缓存键名",
"in": "query",
"name": "key",
"required": true,
"schema": {
"description": "缓存键名",
"title": "Key",
"type": "string"
}
},
{
"in": "header",
"name": "X-Project-Id",
"required": true,
"schema": {
"title": "X-Project-Id",
"type": "string"
}
}
],
"responses": {
"204": {
"description": "Successful Response"
},
"401": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Authentication required"
},
"403": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Insufficient permission"
},
"404": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Resource not found"
},
"409": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Resource conflict"
},
"422": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Validation error"
},
"503": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Dependency unavailable"
}
},
"security": [
{
"OAuth2PasswordBearer": []
}
],
"summary": "清除单个缓存键",
"tags": [
"Cache"
]
}
},
"/api/v1/redos": {
"post": {
"description": "重做网络上被撤销的操作",
-15
View File
@@ -16,12 +16,8 @@ services:
- ../../resources:/app/resources
environment:
- PYTHONPATH=/app
- REDIS_HOST=redis
- REDIS_PORT=${REDIS_PORT}
- REDIS_PASSWORD=${REDIS_PASSWORD}
# Add other DB connections here as needed by your app
depends_on:
- redis
- timescaledb
- postgis
@@ -29,17 +25,6 @@ services:
# Infrastructure Services
# ==========================================
# --- Redis ---
redis:
image: redis:latest
container_name: redis
restart: always
command: redis-server --requirepass ${REDIS_PASSWORD}
ports:
- "${REDIS_PORT}:6379"
volumes:
- ./redis/data:/data
# --- Keycloak ---
keycloakDB:
image: postgis/postgis:14-3.5
-4
View File
@@ -29,7 +29,6 @@ email-validator==2.3.0
esda==2.7.0
et_xmlfile==2.0.0
exceptiongroup==1.3.1
fakeredis==2.33.0
fastapi==0.128.0
fastmcp==2.9.2
fonttools==4.58.0
@@ -43,7 +42,6 @@ httpx==0.28.1
httpx-sse==0.4.3
idna==3.10
importlib_metadata==8.7.1
influxdb-client==1.48.0
iniconfig==2.0.0
jaraco.classes==3.4.0
jaraco.context==6.1.0
@@ -107,7 +105,6 @@ pydantic==2.10.6
pydantic-settings==2.12.0
pydantic_core==2.27.2
pydevd-pycharm==243.16718.36
pydocket==0.16.6
Pygments==2.18.0
PyJWT==2.10.1
pykalman==0.10.2
@@ -127,7 +124,6 @@ pytz==2025.2
PyYAML==6.0.3
pyzmq==26.2.1
reactivex==4.0.4
redis==5.2.1
referencing==0.36.2
requests==2.32.3
rich==14.2.0
Binary file not shown.
-25
View File
@@ -1,25 +0,0 @@
import auto_realtime
import auto_store_non_realtime_SCADA_data
import asyncio
import influxdb_api
import influxdb_info
import project_info
# 为了让多个任务并发运行,我们可以用 asyncio.to_thread 分别启动它们
async def main():
task1 = asyncio.to_thread(auto_realtime.realtime_task)
task2 = asyncio.to_thread(auto_store_non_realtime_SCADA_data.store_non_realtime_SCADA_data_task)
await asyncio.gather(task1, task2)
if __name__ == "__main__":
url = influxdb_info.url
token = influxdb_info.token
org_name = influxdb_info.org
influxdb_api.query_pg_scada_info_realtime(project_info.name)
influxdb_api.query_pg_scada_info_non_realtime(project_info.name)
# 用 asyncio 并发启动两个任务
asyncio.run(main())
-115
View File
@@ -1,115 +0,0 @@
import schedule
import time
import datetime
import shutil
import redis
import urllib.request
import influxdb_api
import msgpack
import datetime
# 将 Query的信息 序列号到 redis/json 默认不支持datetime,需要自定义
# 自定义序列化函数
# 序列化处理器
def encode_datetime(obj):
"""将datetime转换为可序列化的字典结构"""
if isinstance(obj, datetime.datetime):
return {
'__datetime__': True,
'as_str': obj.strftime("%Y%m%dT%H:%M:%S.%f")
}
return obj
# 反序列化处理器
def decode_datetime(obj):
"""将字典还原为datetime对象"""
if '__datetime__' in obj:
return datetime.datetime.strptime(
obj['as_str'], "%Y%m%dT%H:%M:%S.%f"
)
return obj
##########################
# 需要用Python 3.12 来运行才能提高performance
##########################
def queryallrecordsbydate(querydate: str, redis_client: redis.Redis):
cache_key = f"queryallrecordsbydate_{querydate}"
exists = redis_client.exists(cache_key)
if not exists:
nodes_links: tuple = influxdb_api.query_all_records_by_date(query_date=querydate)
redis_client.set(cache_key, msgpack.packb(nodes_links, default=encode_datetime))
def queryallrecordsbydate_by_url(querydate: str):
print(f'queryallrecordsbydate: {querydate}')
try:
response = urllib.request.urlopen(
f"http://127.0.0.1/queryallrecordsbydate/?querydate={querydate}"
)
html = response.read().decode("utf-8")
except urllib.error.URLError as e:
print("Error")
def queryallscadarecordsbydate(querydate: str, redis_client: redis.Redis):
cache_key = f"queryallscadarecordsbydate_{querydate}"
exists = redis_client.exists(cache_key)
if not exists:
result_dict = influxdb_api.query_all_SCADA_records_by_date(query_date=querydate)
redis_client.set(cache_key, msgpack.packb(result_dict, default=encode_datetime))
def queryallscadarecordsbydate_by_url(querydate: str):
print(f'queryallscadarecordsbydate: {querydate}')
try:
response = urllib.request.urlopen(
f"http://127.0.0.1/queryallscadarecordsbydate/?querydate={querydate}"
)
html = response.read().decode("utf-8")
except urllib.error.URLError as e:
print("Error")
def auto_cache_data():
# 初始化 Redis 连接
# 用redis 限制并发访u
redis_client = redis.Redis(host="127.0.0.1", port=6379, db=0)
# auto cache data for the last 3 days
today = datetime.date.today()
for i in range(1, 4):
prev_day = today - datetime.timedelta(days=i)
str_prev_day = prev_day.strftime('%Y-%m-%d')
print(str_prev_day)
queryallrecordsbydate(str_prev_day, redis_client)
queryallscadarecordsbydate(str_prev_day, redis_client)
redis_client.close()
def auto_cache_data_by_url():
# auto cache data for the last 3 days
today = datetime.date.today()
for i in range(1, 4):
prev_day = today - datetime.timedelta(days=i)
str_prev_day = prev_day.strftime('%Y-%m-%d')
print(str_prev_day)
queryallrecordsbydate_by_url(str_prev_day)
queryallscadarecordsbydate_by_url(str_prev_day)
if __name__ == "__main__":
auto_cache_data_by_url()
# auto run in the midnight
schedule.every().day.at("03:00").do(auto_cache_data_by_url)
while True:
schedule.run_pending()
time.sleep(1)
-156
View File
@@ -1,156 +0,0 @@
from logging.handlers import TimedRotatingFileHandler
import influxdb_api
import os
import logging
import globals
from datetime import datetime, timedelta, timezone
import schedule
import time
import shutil
from influxdb_client import InfluxDBClient, BucketsApi, WriteApi, OrganizationsApi, Point, QueryApi
import simulation
import influxdb_info
import project_info
def setup_logger():
# 创建日志目录
log_dir = "logs"
os.makedirs(log_dir, exist_ok=True)
# 配置基础日志格式
log_format = "%(asctime)s - %(levelname)s - %(message)s"
formatter = logging.Formatter(log_format)
# 创建主 Logger
logger = logging.getLogger()
logger.setLevel(logging.INFO) # 全局日志级别
# --- 1. 按日期分割的日志文件 Handler ---
log_file = os.path.join(log_dir, "simulation.log")
file_handler = TimedRotatingFileHandler(
filename=log_file,
when="midnight", # 每天午夜轮转
interval=1,
backupCount=7,
encoding="utf-8"
)
file_handler.suffix = "simulation-%Y-%m-%d.log" # 文件名格式
file_handler.setFormatter(formatter)
file_handler.setLevel(logging.INFO) # 文件记录所有级别日志
# --- 2. 控制台实时输出 Handler ---
console_handler = logging.StreamHandler() # 默认输出到 sys.stderr (控制台)
console_handler.setFormatter(formatter)
console_handler.setLevel(logging.INFO) # 控制台仅显示 INFO 及以上级别
# 将 Handler 添加到 Logger
logger.addHandler(file_handler)
#logger.addHandler(console_handler)
return logger
logger = setup_logger()
# 2025/02/01
def get_next_time() -> str:
"""
获取下一个1分钟时间点返回格式为字符串'YYYY-MM-DDTHH:MM:00+08:00'
:return: 返回字符串格式的时间表示下一个1分钟的时间点
"""
# 获取当前时间,并设定为北京时间
now = datetime.now() # now 类型为 datetime,表示当前本地时间
# 获取当前的分钟,并且将秒和微秒置为零
current_time = now.replace(second=0, microsecond=0) # current_time 类型为 datetime,时间的秒和微秒部分被清除
return current_time.strftime('%Y-%m-%dT%H:%M:%S+08:00')
# 2025/02/06
def store_realtime_SCADA_data_job() -> None:
"""
定义的任务1每分钟执行1次每次执行时更新get_real_value_time并调用store_realtime_SCADA_data_to_influxdb函数
:return: None
"""
# 获取当前时间并更新get_real_value_time,转换为字符串格式
get_real_value_time: str = get_next_time() # get_real_value_time 类型为 str,格式为'2025-02-01T18:45:00+08:00'
# 调用函数执行任务
influxdb_api.store_realtime_SCADA_data_to_influxdb(get_real_value_time)
logger.info('{} -- Successfully store realtime SCADA data.'.format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
# 2025/02/06
def get_next_15minute_time() -> str:
"""
获取下一个15分钟的时间点返回格式为字符串'YYYY-MM-DDTHH:MM:00+08:00'
:return: 返回字符串格式的时间表示下一个15分钟执行时间点
"""
now = datetime.now()
# 向上舍入到下一个15分钟
next_15minute = (now.minute // 15 + 1) * 15 - 15
if next_15minute == 60:
next_15minute = 0
now = now + timedelta(hours=1)
next_time = now.replace(minute=next_15minute, second=0, microsecond=0)
return next_time.strftime('%Y-%m-%dT%H:%M:%S+08:00')
# 2025/02/07
def run_simulation_job() -> None:
"""
定义的任务3每15分钟执行一次在store_realtime_SCADA_data_to_influxdb之后执行run_simulation
:return: None
"""
# 获取当前时间,并检查是否是整点15分钟
current_time = datetime.now()
if current_time.minute % 15 == 0:
print(f"{current_time.strftime('%Y-%m-%d %H:%M:%S')} -- Start simulation task.")
# 计算前,获取scada_info中的信息,按照设定的方法修改pg数据库
simulation.query_corresponding_element_id_and_query_id(project_info.name)
simulation.query_corresponding_pattern_id_and_query_id(project_info.name)
region_result = simulation.query_non_realtime_region(project_info.name)
globals.source_outflow_region_id = simulation.get_source_outflow_region_id(project_info.name, region_result)
globals.realtime_region_pipe_flow_and_demand_id = simulation.query_realtime_region_pipe_flow_and_demand_id(project_info.name, region_result)
globals.pipe_flow_region_patterns = simulation.query_pipe_flow_region_patterns(project_info.name)
globals.non_realtime_region_patterns = simulation.query_non_realtime_region_patterns(project_info.name, region_result)
globals.source_outflow_region_patterns, realtime_region_pipe_flow_and_demand_patterns = simulation.get_realtime_region_patterns(project_info.name,
globals.source_outflow_region_id,
globals.realtime_region_pipe_flow_and_demand_id)
modify_pattern_start_time: str = get_next_15minute_time() # 获取下一个15分钟时间点
# print(modify_pattern_start_time)
simulation.run_simulation(name=project_info.name, simulation_type="realtime", modify_pattern_start_time=modify_pattern_start_time)
logger.info('{} -- Successfully run simulation and store realtime simulation result.'.format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
else:
logger.info(f"{current_time.strftime('%Y-%m-%d %H:%M:%S')} -- Skipping the simulation task.")
# 2025/02/06
def realtime_task() -> None:
"""
定时执行任务1和使用schedule库每1分钟执行一次store_realtime_SCADA_data_job函数
该任务会一直运行定期调用store_realtime_SCADA_data_job获取SCADA数据
:return:
"""
# 等待到整分对齐
now = datetime.now()
wait_seconds = 60 - now.second
time.sleep(wait_seconds)
# 使用 .at(":00") 指定在每分钟的第0秒执行
schedule.every(1).minute.at(":00").do(store_realtime_SCADA_data_job)
# 每15分钟执行一次run_simulation_job
schedule.every(1).minute.at(":00").do(run_simulation_job)
# 持续执行任务,检查是否有待执行的任务
while True:
schedule.run_pending() # 执行所有待处理的定时任务
time.sleep(1) # 暂停1秒,避免过于频繁的任务检查
if __name__ == "__main__":
url = influxdb_info.url
token = influxdb_info.token
org_name = influxdb_info.org
client = InfluxDBClient(url=url, token=token)
# step2: 先查询pg数据库中scada_info的信息,然后存储SCADA数据到SCADA_data这个bucket里
influxdb_api.query_pg_scada_info_realtime(project_info.name)
# 自动执行
realtime_task()
@@ -1,139 +0,0 @@
import influxdb_api
import globals
from datetime import datetime, timedelta, timezone
import schedule
import os
import logging
from logging.handlers import TimedRotatingFileHandler
import time
from influxdb_client import InfluxDBClient, BucketsApi, WriteApi, OrganizationsApi, Point, QueryApi
import influxdb_info
import project_info
def setup_logger():
# 创建日志目录
log_dir = "logs"
os.makedirs(log_dir, exist_ok=True)
# 配置基础日志格式
log_format = "%(asctime)s - %(levelname)s - %(message)s"
formatter = logging.Formatter(log_format)
# 创建主 Logger
logger = logging.getLogger()
logger.setLevel(logging.INFO) # 全局日志级别
# --- 1. 按日期分割的日志文件 Handler ---
log_file = os.path.join(log_dir, "scada.log")
file_handler = TimedRotatingFileHandler(
filename=log_file,
when="midnight", # 每天午夜轮转
interval=1,
backupCount=7,
encoding="utf-8"
)
file_handler.suffix = "scada-%Y-%m-%d.log" # 文件名格式
file_handler.setFormatter(formatter)
file_handler.setLevel(logging.INFO) # 文件记录 INFO 及以上级别
# --- 2. 控制台实时输出 Handler ---
console_handler = logging.StreamHandler() # 默认输出到 sys.stderr (控制台)
console_handler.setFormatter(formatter)
console_handler.setLevel(logging.INFO) # 控制台仅显示 INFO 及以上级别
# 将 Handler 添加到 Logger
logger.addHandler(file_handler)
# logger.addHandler(console_handler)
return logger
logger = setup_logger()
# 2025/02/01
def get_next_time() -> str:
"""
获取下一个1分钟时间点返回格式为字符串'YYYY-MM-DDTHH:MM:00+08:00'
:return: 返回字符串格式的时间表示下一个1分钟的时间点
"""
# 获取当前时间,并设定为北京时间
now = datetime.now() # now 类型为 datetime,表示当前本地时间
# 获取当前的分钟,并且将秒和微秒置为零
current_time = now.replace(second=0, microsecond=0) # current_time 类型为 datetime,时间的秒和微秒部分被清除
return current_time.strftime('%Y-%m-%dT%H:%M:%S+08:00')
# 2025/02/06
def get_next_period_time() -> str:
"""
获取下一个6小时时间点返回格式为字符串'YYYY-MM-DDTHH:00:00+08:00'
:return: 返回字符串格式的时间表示下一个6小时执行时间点
"""
# 获取当前时间,并设定为北京时间
now = datetime.now() # now 类型为 datetime,表示当前本地时间
# 获取当前的小时数并计算下一个6小时时间点
next_period_hour = (now.hour // 6 + 1) * 6 - 6 # next_period_hour 类型为 int,表示下一个6小时时间点的小时部分
# 如果计算的小时大于23,表示进入第二天,调整为00:00
if next_period_hour >= 24:
next_period_hour = 0
now = now + timedelta(days=1) # 如果超过24小时,日期增加1天
# 将秒和微秒部分清除,构建出下一个6小时点的datetime对象
next_period_time = now.replace(hour=next_period_hour, minute=0, second=0, microsecond=0)
return next_period_time.strftime('%Y-%m-%dT%H:%M:%S+08:00') # 格式化为指定的字符串格式并返回
# 2025/02/06
def store_non_realtime_SCADA_data_job() -> None:
"""
定义的任务2每6小时执行一次在0点61218点执行执行时更新get_history_data_end_time并调用store_non_realtime_SCADA_data_to_influxdb函数
:return: None
"""
# 获取当前时间
current_time = datetime.now()
# 只在0点、6点、12点、18点执行任务
# if current_time.hour % 6 == 0 and current_time.minute == 0:
if current_time.minute % 10 == 0:
logger.info(f"{current_time.strftime('%Y-%m-%d %H:%M:%S')} -- Start store non realtime SCADA data task.")
# 获取下一个6小时的时间点,并更新get_history_data_end_time
get_history_data_end_time: str = get_next_time() # get_history_data_end_time 类型为 str,格式为'2025-02-06T12:00:00+08:00'
# print(get_next_time)
# 调用函数执行任务
influxdb_api.store_non_realtime_SCADA_data_to_influxdb(get_history_data_end_time)
logger.info('{} -- Successfully store non realtime SCADA data.'.format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
else:
logger.info(f"{current_time.strftime('%Y-%m-%d %H:%M:%S')} -- Skipping store non realtime SCADA data task.")
# 2025/02/06
def store_non_realtime_SCADA_data_task() -> None:
"""
定时执行6小时的任务使用schedule库每分钟执行一次store_non_realtime_SCADA_data_job函数
该任务会一直运行定期调用store_non_realtime_SCADA_data_job获取SCADA数据
:return:
"""
# 等待到整分对齐
now = datetime.now()
wait_seconds = 60 - now.second
time.sleep(wait_seconds)
try:
# 每分钟检查一次,执行store_non_realtime_SCADA_data_job
schedule.every(1).minute.at(":00").do(store_non_realtime_SCADA_data_job)
# 持续执行任务,检查是否有待执行的任务
while True:
schedule.run_pending() # 执行所有待处理的定时任务
time.sleep(1) # 暂停1秒,避免过于频繁的任务检查
pass
except Exception as e:
logger.error(f"Error occurred in store_non_realtime_SCADA_data_task: {e}")
if __name__ == "__main__":
url = influxdb_info.url
token = influxdb_info.token
org_name = influxdb_info.org
client = InfluxDBClient(url=url, token=token)
# step2: 先查询pg数据库中scada_info的信息,然后存储SCADA数据到SCADA_data这个bucket里
influxdb_api.query_pg_scada_info_non_realtime(project_info.name)
# 自动执行
store_non_realtime_SCADA_data_task()
-5
View File
@@ -2,9 +2,6 @@ from distutils.core import setup
from Cython.Build import cythonize
setup(ext_modules=cythonize([
"main.py",
"auto_realtime.py",
"auto_store_non_realtime_SCADA_data.py",
"tjnetwork.py",
"online_Analysis.py",
"sensitivity.py",
@@ -15,8 +12,6 @@ setup(ext_modules=cythonize([
"get_data.py",
"get_current_total_Q.py",
"get_current_status.py",
"influxdb_api.py",
"influxdb_query_SCADA_data.py",
"simulation.py",
"time_api.py",
"api/*.py",
-2
View File
@@ -121,7 +121,6 @@ def get_history_data(
# print(data)
# # 定义 CSV 文件的路径
# csv_file_path = './influxdb_data_4984.csv'
# # 将数据写入 CSV 文件
# # with open(csv_file_path, mode='w', newline='') as file:
# # writer = csv.writer(file)
@@ -143,7 +142,6 @@ def get_history_data(
# #
# # print(f"数据已保存到 {csv_file_path}")
#
# filtered_csv_file_path = './filtered_influxdb_data_4984.csv'
# #
# # # # 读取并筛选数据
# data_list1 = []
-2
View File
@@ -18,14 +18,12 @@ def install():
packages = [
'"psycopg[binary]"',
'pytest',
'influxdb_client',
'numpy',
'fastapi',
"msgpack",
'schedule',
'pandas',
'openpyxl',
'redis',
'pydantic',
'python-dateutil',
'starlette',
-4481
View File
File diff suppressed because it is too large Load Diff
-395
View File
@@ -1,395 +0,0 @@
# API Endpoints (scripts/main.py)
Non-commented FastAPI routes defined in `scripts/main.py`.
- `POST /login/`
- `GET /getallextensiondatakeys/`
- `GET /getallextensiondata/`
- `GET /getextensiondata/`
- `POST /setextensiondata`
- `GET /listprojects/`
- `GET /haveproject/`
- `POST /createproject/`
- `POST /deleteproject/`
- `GET /isprojectopen/`
- `POST /openproject/`
- `POST /closeproject/`
- `POST /copyproject/`
- `POST /importinp/`
- `GET /exportinp/`
- `POST /readinp/`
- `GET /dumpinp/`
- `GET /runproject/`
- `GET /runprojectreturndict/`
- `GET /runinp/`
- `GET /dumpoutput/`
- `GET /isprojectlocked/`
- `GET /isprojectlockedbyme/`
- `POST /lockproject/`
- `POST /unlockproject/`
- `GET /getcurrentoperationid/`
- `POST /undo/`
- `POST /redo/`
- `GET /getsnapshots/`
- `GET /havesnapshot/`
- `GET /havesnapshotforoperation/`
- `GET /havesnapshotforcurrentoperation/`
- `POST /takesnapshotforoperation/`
- `POST takenapshotforcurrentoperation`
- `POST /takesnapshot/`
- `POST /picksnapshot/`
- `POST /pickoperation/`
- `GET /syncwithserver/`
- `POST /batch/`
- `POST /compressedbatch/`
- `GET /getrestoreoperation/`
- `POST /setrestoreoperation/`
- `GET /isnode/`
- `GET /isjunction/`
- `GET /isreservoir/`
- `GET /istank/`
- `GET /islink/`
- `GET /ispipe/`
- `GET /ispump/`
- `GET /isvalve/`
- `GET /getnodetype/`
- `GET /getlinktype/`
- `GET /getelementtype/`
- `GET /getelementtypevalue/`
- `GET /iscurve/`
- `GET /ispattern/`
- `GET /getnodes/`
- `GET /getlinks/`
- `GET /getcurves/`
- `GET /getpatterns/`
- `GET /getnodelinks/`
- `GET /getnodeproperties/`
- `GET /getlinkproperties/`
- `GET /getscadaproperties/`
- `GET /getallscadaproperties/`
- `GET /getelementpropertieswithtype/`
- `GET /getelementproperties/`
- `GET /gettitleschema/`
- `GET /gettitle/`
- `GET /settitle/`
- `GET /getjunctionschema`
- `POST /addjunction/`
- `POST /deletejunction/`
- `GET /getjunctionelevation/`
- `GET /getjunctionx/`
- `GET /getjunctiony/`
- `GET /getjunctioncoord/`
- `GET /getjunctiondemand/`
- `GET /getjunctionpattern/`
- `POST /setjunctionelevation/`
- `POST /setjunctionx/`
- `POST /setjunctiony/`
- `POST /setjunctioncoord/`
- `POST /setjunctiondemand/`
- `POST /setjunctionpattern/`
- `GET /getjunctionproperties/`
- `GET /getalljunctionproperties/`
- `POST /setjunctionproperties/`
- `GET /getreservoirschema`
- `POST /addreservoir/`
- `POST /deletereservoir/`
- `GET /getreservoirhead/`
- `GET /getreservoirpattern/`
- `GET /getreservoirx/`
- `GET /getreservoiry/`
- `GET /getreservoircoord/`
- `POST /setreservoirhead/`
- `POST /setreservoirpattern/`
- `POST /setreservoirx/`
- `POST /setreservoirx/`
- `POST /setreservoircoord/`
- `GET /getreservoirproperties/`
- `GET /getallreservoirproperties/`
- `POST /setreservoirproperties/`
- `GET /gettankschema`
- `POST /addtank/`
- `POST /deletetank/`
- `GET /gettankelevation/`
- `GET /gettankinitlevel/`
- `GET /gettankminlevel/`
- `GET /gettankmaxlevel/`
- `GET /gettankdiameter/`
- `GET /gettankminvol/`
- `GET /gettankvolcurve/`
- `GET /gettankoverflow/`
- `GET /gettankx/`
- `GET /gettanky/`
- `GET /gettankcoord/`
- `POST /settankelevation/`
- `POST /settankinitlevel/`
- `POST /settankminlevel/`
- `POST /settankmaxlevel/`
- `POST settankdiameter//`
- `POST /settankminvol/`
- `POST /settankvolcurve/`
- `POST /settankoverflow/`
- `POST /settankx/`
- `POST /settanky/`
- `POST /settankcoord/`
- `GET /gettankproperties/`
- `GET /getalltankproperties/`
- `POST /settankproperties/`
- `GET /getpipeschema`
- `POST /addpipe/`
- `POST /deletepipe/`
- `GET /getpipenode1/`
- `GET /getpipenode2/`
- `GET /getpipelength/`
- `GET /getpipediameter/`
- `GET /getpiperoughness/`
- `GET /getpipeminorloss/`
- `GET /getpipestatus/`
- `POST /setpipenode1/`
- `POST /setpipenode2/`
- `POST /setpipelength/`
- `POST /setpipediameter/`
- `POST /setpiperoughness/`
- `POST /setpipeminorloss/`
- `POST /setpipestatus/`
- `GET /getpipeproperties/`
- `GET /getallpipeproperties/`
- `POST /setpipeproperties/`
- `GET /getpumpschema`
- `POST /addpump/`
- `POST /deletepump/`
- `GET /getpumpnode1/`
- `GET /getpumpnode2/`
- `POST /setpumpnode1/`
- `POST /setpumpnode2/`
- `GET /getpumpproperties/`
- `GET /getallpumpproperties/`
- `POST /setpumpproperties/`
- `GET /getvalveschema`
- `POST /addvalve/`
- `POST /deletevalve/`
- `GET /getvalvenode1/`
- `GET /getvalvenode2/`
- `GET /getvalvediameter/`
- `GET /getvalvetype/`
- `GET /getvalvesetting/`
- `GET /getvalveminorloss/`
- `POST /setvalvenode1/`
- `POST /setvalvenode2/`
- `POST /setvalvenodediameter/`
- `POST /setvalvetype/`
- `POST /setvalvesetting/`
- `GET /getvalveproperties/`
- `GET /getallvalveproperties/`
- `POST /setvalveproperties/`
- `POST /deletenode/`
- `POST /deletelink/`
- `GET /gettagschema/`
- `GET /gettag/`
- `GET /gettags/`
- `POST /settag/`
- `GET /getdemandschema`
- `GET /getdemandproperties/`
- `POST /setdemandproperties/`
- `GET /getstatusschema`
- `GET /getstatus/`
- `POST /setstatus/`
- `GET /getpatternschema`
- `POST /addpattern/`
- `POST /deletepattern/`
- `GET /getpatternproperties/`
- `POST /setpatternproperties/`
- `GET /getcurveschema`
- `POST /addcurve/`
- `POST /deletecurve/`
- `GET /getcurveproperties/`
- `POST /setcurveproperties/`
- `GET /getcontrolschema/`
- `GET /getcontrolproperties/`
- `POST /setcontrolproperties/`
- `GET /getruleschema/`
- `GET /getruleproperties/`
- `POST /setruleproperties/`
- `GET /getenergyschema/`
- `GET /getenergyproperties/`
- `POST /setenergyproperties/`
- `GET /getpumpenergyschema/`
- `GET /getpumpenergyproperties//`
- `GET /setpumpenergyproperties//`
- `GET /getemitterschema`
- `GET /getemitterproperties/`
- `POST /setemitterproperties/`
- `GET /getqualityschema/`
- `GET /getqualityproperties/`
- `POST /setqualityproperties/`
- `GET /getsourcechema/`
- `GET /getsource/`
- `POST /setsource/`
- `POST /addsource/`
- `POST /deletesource/`
- `GET /getreactionschema/`
- `GET /getreaction/`
- `POST /setreaction/`
- `GET /getpipereactionschema/`
- `GET /getpipereaction/`
- `POST /setpipereaction/`
- `GET /gettankreactionschema/`
- `GET /gettankreaction/`
- `POST /settankreaction/`
- `GET /getmixingschema/`
- `GET /getmixing/`
- `POST /setmixing/`
- `POST /addmixing/`
- `POST /deletemixing/`
- `GET /gettimeschema`
- `GET /gettimeproperties/`
- `POST /settimeproperties/`
- `GET /getoptionschema/`
- `GET /getoptionproperties/`
- `POST /setoptionproperties/`
- `GET /getnodecoord/`
- `GET /getnetworkgeometries/`
- `GET /getmajornodecoords/`
- `GET /getnetworkinextent/`
- `GET /getnetworklinknodes/`
- `GET /getmajorpipenodes/`
- `GET /getvertexschema/`
- `GET /getvertexproperties/`
- `POST /setvertexproperties/`
- `POST /addvertex/`
- `POST /deletevertex/`
- `GET /getallvertexlinks/`
- `GET /getallvertices/`
- `GET /getlabelschema/`
- `GET /getlabelproperties/`
- `POST /setlabelproperties/`
- `POST /addlabel/`
- `POST /deletelabel/`
- `GET /getbackdropschema/`
- `GET /getbackdropproperties/`
- `POST /setbackdropproperties/`
- `GET /getscadadeviceschema/`
- `GET /getscadadevice/`
- `POST /setscadadevice/`
- `POST /addscadadevice/`
- `POST /deletescadadevice/`
- `POST /cleanscadadevice/`
- `GET /getallscadadeviceids/`
- `GET /getallscadadevices/`
- `GET /getscadadevicedataschema/`
- `GET /getscadadevicedata/`
- `POST /setscadadevicedata/`
- `POST /addscadadevicedata/`
- `POST /deletescadadevicedata/`
- `POST /cleanscadadevicedata/`
- `GET /getscadaelementschema/`
- `GET /getscadaelements/`
- `GET /getscadaelement/`
- `POST /setscadaelement/`
- `POST /addscadaelement/`
- `POST /deletescadaelement/`
- `POST /cleanscadaelement/`
- `GET /getregionschema/`
- `GET /getregion/`
- `POST /setregion/`
- `POST /addregion/`
- `POST /deleteregion/`
- `GET /calculatedistrictmeteringareafornodes/`
- `GET /calculatedistrictmeteringareaforregion/`
- `GET /calculatedistrictmeteringareafornetwork/`
- `GET /getdistrictmeteringareaschema/`
- `GET /getdistrictmeteringarea/`
- `POST /setdistrictmeteringarea/`
- `POST /adddistrictmeteringarea/`
- `POST /deletedistrictmeteringarea/`
- `GET /getalldistrictmeteringareaids/`
- `GET /getalldistrictmeteringareas/`
- `POST /generatedistrictmeteringarea/`
- `POST /generatesubdistrictmeteringarea/`
- `GET /calculateservicearea/`
- `GET /getserviceareaschema/`
- `GET /getservicearea/`
- `POST /setservicearea/`
- `POST /addservicearea/`
- `POST /deleteservicearea/`
- `GET /getallserviceareas/`
- `POST /generateservicearea/`
- `GET /calculatevirtualdistrict/`
- `GET /getvirtualdistrictschema/`
- `GET /getvirtualdistrict/`
- `POST /setvirtualdistrict/`
- `POST /addvirtualdistrict/`
- `POST /deletevirtualdistrict/`
- `GET /getallvirtualdistrict/`
- `POST /generatevirtualdistrict/`
- `GET /calculatedemandtonodes/`
- `GET /calculatedemandtoregion/`
- `GET /calculatedemandtonetwork/`
- `GET /getscadainfoschema/`
- `GET /getscadainfo/`
- `GET /getallscadainfo/`
- `GET /getschemeschema/`
- `GET /getscheme/`
- `GET /getallschemes/`
- `GET /getpiperiskprobabilitynow/`
- `GET /getpiperiskprobability/`
- `GET /getpipesriskprobability/`
- `GET /getnetworkpiperiskprobabilitynow/`
- `GET /getpiperiskprobabilitygeometries/`
- `GET /getallsensorplacements/`
- `GET /getallburstlocateresults/`
- `POST /uploadinp/`
- `GET /downloadinp/`
- `GET /convertv3tov2/`
- `GET /getjson/`
- `GET /getrealtimedata/`
- `GET /getsimulationresult/`
- `GET /querynodelatestrecordbyid/`
- `GET /querylinklatestrecordbyid/`
- `GET /queryscadalatestrecordbyid/`
- `GET /queryallrecordsbytime/`
- `GET /queryallrecordsbytimeproperty/`
- `GET /queryallschemerecordsbytimeproperty/`
- `GET /querysimulationrecordsbyidtime/`
- `GET /queryschemesimulationrecordsbyidtime/`
- `GET /queryallrecordsbydate/`
- `GET /queryallrecordsbytimerange/`
- `GET /queryallrecordsbydatewithtype/`
- `GET /queryallrecordsbyidsdatetype/`
- `GET /queryallrecordsbydateproperty/`
- `GET /querynodecurvebyidpropertydaterange/`
- `GET /querylinkcurvebyidpropertydaterange/`
- `GET /queryscadadatabydeviceidandtime/`
- `GET /queryscadadatabydeviceidandtimerange/`
- `GET /queryfillingscadadatabydeviceidandtimerange/`
- `GET /querycleaningscadadatabydeviceidandtimerange/`
- `GET /querysimulationscadadatabydeviceidandtimerange/`
- `GET /querycleanedscadadatabydeviceidandtimerange/`
- `GET /queryscadadatabydeviceidanddate/`
- `GET /queryallscadarecordsbydate/`
- `GET /queryallschemeallrecords/`
- `GET /queryschemeallrecordsproperty/`
- `POST /clearrediskey/`
- `POST /clearrediskeys/`
- `POST /clearallredis/`
- `GET /queryredis/`
- `GET /queryinfluxdbbuckets/`
- `GET /queryinfluxdbbucketmeasurements/`
- `POST /download_history_data_manually/`
- `POST /runsimulationmanuallybydate/`
- `POST /burst_analysis/`
- `GET /valve_close_analysis/`
- `GET /flushing_analysis/`
- `GET /contaminant_simulation/`
- `GET /age_analysis/`
- `POST /scheduling_analysis/`
- `POST /pressure_regulation/`
- `POST /project_management/`
- `POST /network_project/`
- `POST /daily_scheduling_analysis/`
- `POST /network_update/`
- `POST /pump_failure/`
- `POST /pressure_sensor_placement_sensitivity/`
- `POST /pressure_sensor_placement_kmeans/`
- `POST /sensorplacementscheme/create`
- `POST /scadadevicedatacleaning/`
- `POST /test_dict/`
-5
View File
@@ -1,5 +0,0 @@
import redis
redis_client = redis.Redis(host="127.0.0.1", port=6379, db=0)
matched_keys = redis_client.keys(f"**")
redis_client.delete(*matched_keys)
+15
View File
@@ -7,6 +7,21 @@ from fastapi.testclient import TestClient
from app.infra.audit import middleware as audit_middleware
from app.infra.audit.middleware import AuditMiddleware
from app.core.audit import sanitize_sensitive_data
def test_sanitize_sensitive_data_redacts_database_dsn() -> None:
raw_dsn = "postgresql://alice:supersecret@db.internal/project"
sanitized = sanitize_sensitive_data(
{"dsn": raw_dsn, "database": {"readonly_dsn": raw_dsn}}
)
assert sanitized == {
"dsn": "***REDACTED***",
"database": {"readonly_dsn": "***REDACTED***"},
}
assert raw_dsn not in str(sanitized)
def test_post_streaming_response_survives_audit_body_capture(monkeypatch):
+46
View File
@@ -10,6 +10,8 @@ from app.auth.metadata_dependencies import (
get_current_metadata_admin,
get_metadata_repository,
)
from app.infra.db.metadb.repositories.metadata_repository import ProjectDbRouting
from app.infra.db.project_routing import get_project_pgconn_string
from tests.conftest import build_test_app
@@ -98,3 +100,47 @@ def test_model_import_rejects_non_inp_file(monkeypatch):
assert response.status_code == 400
assert response.json()["detail"] == "Only .inp model files are accepted"
model_import.log_audit_event.assert_not_awaited()
def test_model_update_uses_project_business_routing(monkeypatch):
project_id = uuid4()
project = SimpleNamespace(id=project_id, code="demo", status="active")
repo = SimpleNamespace(
session=object(),
get_project_by_id=AsyncMock(return_value=project),
get_project_db_routing=AsyncMock(
return_value=ProjectDbRouting(
project_id=project_id,
db_role="biz_data",
db_type="postgresql",
dsn="postgresql://user:password@biz.example/routed_business",
pool_min_size=1,
pool_max_size=5,
)
),
)
captured: dict[str, str] = {}
async def fake_apply_model_update(content: bytes, project_code: str) -> None:
assert content == VALID_INP
captured["project_code"] = project_code
captured["dsn"] = get_project_pgconn_string(project_code)
monkeypatch.setattr(model_import, "_apply_model_update", fake_apply_model_update)
monkeypatch.setattr(model_import, "log_audit_event", AsyncMock())
client = _client(
admin=SimpleNamespace(id=uuid4(), role="admin", is_superuser=False),
repo=repo,
)
response = client.patch(
f"/api/v1/admin/projects/{project_id}/model-imports",
files={"file": ("desktop-model.inp", VALID_INP)},
)
assert response.status_code == 200
assert captured == {
"project_code": "demo",
"dsn": "postgresql://user:password@biz.example/routed_business",
}
repo.get_project_db_routing.assert_awaited_once_with(project_id, "biz_data")
+121 -5
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import inspect
from datetime import datetime, timezone
from pathlib import Path
from uuid import uuid4
@@ -11,12 +12,44 @@ from fastapi.testclient import TestClient
from app.api.v1.endpoints import schemes as schemes_endpoint
from app.api.v1.endpoints import simulation as simulation_endpoint
from app.api.pagination import PaginatedList
from app.api.v1.rest_router import api_router, build_rest_router
from app.api.v1.router import api_router as source_api_router
from app.auth.project_dependencies import ProjectContext, get_project_context
from app.auth.metadata_dependencies import get_current_metadata_user
from app.auth.project_dependencies import (
ProjectContext,
get_project_business_routing,
get_project_context,
get_project_simulation_routing,
)
from app.infra.db.project_routing import (
ActiveProjectRouting,
get_project_pgconn_string,
get_project_timescale_pgconn_string,
)
from scripts.check_openapi import current_contract_bytes, validate
def _override_project_routing(
app: FastAPI,
project_context: ProjectContext,
) -> None:
app.dependency_overrides[get_project_context] = lambda: project_context
business = ActiveProjectRouting(
project_code=project_context.project_code,
business_dsn=f"postgresql://user:password@biz/{project_context.project_code}",
)
simulation = ActiveProjectRouting(
project_code=project_context.project_code,
business_dsn=business.business_dsn,
timescale_dsn=(
f"postgresql://user:password@timescale/{project_context.project_code}"
),
)
app.dependency_overrides[get_project_business_routing] = lambda: business
app.dependency_overrides[get_project_simulation_routing] = lambda: simulation
def test_rest_router_preserves_every_distinct_source_operation() -> None:
skipped_names = {"fastapi_get_json", "fastapi_test_dict"}
source_names = {
@@ -40,6 +73,16 @@ def test_rest_router_has_unique_method_path_pairs() -> None:
assert len(pairs) == len(set(pairs))
def test_removed_redis_management_routes_are_not_published() -> None:
published_paths = {
route.path for route in api_router.routes if isinstance(route, APIRoute)
}
assert published_paths.isdisjoint(
{"/redis-keys/detail", "/redis-keys", "/all-redis", "/redis"}
)
def test_rest_router_rejects_duplicate_method_path_pairs() -> None:
first = APIRoute(
"/duplicate",
@@ -130,6 +173,12 @@ def test_rest_contract_uses_header_project_context() -> None:
assert "network" not in schema.get("properties", {})
assert "network_name" not in schema.get("properties", {})
placement_schema = document["components"]["schemas"][
"PressureSensorPlacementRest"
]
assert "name" not in placement_schema["properties"]
assert "username" not in placement_schema["properties"]
assert "/api/v1/burst-analysis" not in document["paths"]
assert "/api/v1/getpipeproperties/" not in document["paths"]
@@ -166,6 +215,21 @@ def test_valve_isolation_route_uses_the_isolation_handler() -> None:
assert route.name == "valve_isolation_endpoint"
def test_open_project_route_requires_business_and_timescale_routing() -> None:
route = next(
route
for route in api_router.routes
if isinstance(route, APIRoute)
and route.path == "/projects/current"
and route.methods == {"POST"}
)
routing_parameter = inspect.signature(route.endpoint).parameters[
"_rest_project_routing"
]
assert routing_parameter.default.dependency is get_project_simulation_routing
def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None:
captured: dict[str, object] = {}
@@ -174,6 +238,8 @@ def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None:
network=network,
accident_element=accident_element,
disabled_valves=disabled_valves,
business_dsn=get_project_pgconn_string(network),
timescale_dsn=get_project_timescale_pgconn_string(network),
)
return {"isolatable": True, "must_close_valves": ["V-1"]}
@@ -184,12 +250,13 @@ def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None:
)
app = FastAPI(redirect_slashes=False)
app.include_router(api_router, prefix="/api/v1")
app.dependency_overrides[get_project_context] = lambda: ProjectContext(
project_context = ProjectContext(
project_id=uuid4(),
project_code="fengyang",
user_id=uuid4(),
project_role="member",
)
_override_project_routing(app, project_context)
response = TestClient(app, raise_server_exceptions=False).post(
"/api/v1/valve-isolation-analyses",
@@ -206,6 +273,8 @@ def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None:
"network": "fengyang",
"accident_element": ["P-1", "P-2"],
"disabled_valves": ["V-9"],
"business_dsn": "postgresql://user:password@biz/fengyang",
"timescale_dsn": "postgresql://user:password@timescale/fengyang",
}
@@ -237,6 +306,7 @@ def test_rest_runtime_consumes_injected_project_context(monkeypatch) -> None:
network=network,
scheme_type=scheme_type,
query_date=query_date,
business_dsn=get_project_pgconn_string(network),
)
return [{"scheme_name": "burst_case", "scheme_type": scheme_type}]
@@ -253,7 +323,7 @@ def test_rest_runtime_consumes_injected_project_context(monkeypatch) -> None:
user_id=uuid4(),
project_role="viewer",
)
app.dependency_overrides[get_project_context] = lambda: project_context
_override_project_routing(app, project_context)
response = TestClient(app, raise_server_exceptions=False).get(
"/api/v1/schemes",
@@ -265,6 +335,7 @@ def test_rest_runtime_consumes_injected_project_context(monkeypatch) -> None:
"network": "fengyang",
"scheme_type": "burst_analysis",
"query_date": None,
"business_dsn": "postgresql://user:password@biz/fengyang",
}
assert response.json()["items"] == [
{"scheme_name": "burst_case", "scheme_type": "burst_analysis"}
@@ -280,7 +351,7 @@ def test_rest_runtime_wraps_handler_paginated_list() -> None:
limit: int = Query(2, ge=1, le=10),
) -> list[int]:
records = [10, 20, 30, 40]
return records[skip : skip + limit]
return PaginatedList(records[skip : skip + limit], total=len(records))
app = FastAPI(redirect_slashes=False)
app.include_router(build_rest_router(source_router.routes), prefix="/api/v1")
@@ -293,12 +364,57 @@ def test_rest_runtime_wraps_handler_paginated_list() -> None:
assert response.status_code == 200
assert response.json() == {
"items": [20, 30],
"total": 3,
"total": 4,
"limit": 2,
"offset": 1,
}
def test_sensor_placement_body_uses_authenticated_project_and_user(
monkeypatch,
) -> None:
captured: dict[str, object] = {}
def fake_pressure_sensor_placement_kmeans(**kwargs):
captured.update(kwargs)
monkeypatch.setattr(
simulation_endpoint,
"pressure_sensor_placement_kmeans",
fake_pressure_sensor_placement_kmeans,
)
app = FastAPI(redirect_slashes=False)
app.include_router(api_router, prefix="/api/v1")
project_context = ProjectContext(
project_id=uuid4(),
project_code="project_a",
user_id=uuid4(),
project_role="member",
)
_override_project_routing(app, project_context)
app.dependency_overrides[get_current_metadata_user] = lambda: type(
"User", (), {"username": "alice"}
)()
response = TestClient(app, raise_server_exceptions=False).post(
"/api/v1/pressure-sensor-placement-kmeans",
json={
"scheme_name": "placement_01",
"sensor_number": 5,
"min_diameter": 100,
},
)
assert response.status_code == 200
assert captured == {
"name": "project_a",
"scheme_name": "placement_01",
"sensor_number": 5,
"min_diameter": 100,
"username": "alice",
}
def test_rest_runtime_json_encodes_untyped_datetime_response() -> None:
source_router = APIRouter()
+140
View File
@@ -383,6 +383,7 @@ def test_flushing_endpoint_passes_required_scheme_name(monkeypatch):
"modify_pattern_start_time": "2025-01-02T03:04:05+08:00",
"modify_total_duration": 900,
"modify_valve_opening": {"V1": 0.5},
"valve_control": None,
"drainage_node_ID": "N1",
"flushing_flow": 100.0,
"scheme_name": "flush_case_01",
@@ -390,6 +391,145 @@ def test_flushing_endpoint_passes_required_scheme_name(monkeypatch):
}
def test_flushing_endpoint_allows_omitting_valves(monkeypatch):
module = _load_simulation_module(monkeypatch)
captured = {}
def fake_flushing_analysis(**kwargs):
captured.update(kwargs)
return "ok"
monkeypatch.setattr(module, "flushing_analysis", fake_flushing_analysis)
client = _build_authenticated_client(module)
response = client.post(
"/api/v1/flushing-analyses",
params={
"network": "demo",
"start_time": "2025-01-02T03:04:05+08:00",
"drainage_node_ID": "N1",
"scheme_name": "flush_without_valves",
},
)
assert response.status_code == 200
assert response.text == "ok"
assert captured["modify_valve_opening"] is None
assert captured["valve_control"] is None
assert captured["drainage_node_ID"] == "N1"
def test_flushing_endpoint_passes_explicit_valve_control(monkeypatch):
module = _load_simulation_module(monkeypatch)
captured = {}
def fake_flushing_analysis(**kwargs):
captured.update(kwargs)
return "ok"
monkeypatch.setattr(module, "flushing_analysis", fake_flushing_analysis)
client = _build_authenticated_client(module)
response = client.post(
"/api/v1/flushing-analyses",
params=[
("network", "demo"),
("start_time", "2025-01-02T03:04:05+08:00"),
("valves", "V1"),
("valves", "V2"),
("valve_statuses", "ACTIVE"),
("valve_statuses", "CLOSED"),
("valve_settings", "2.5"),
("valve_settings", ""),
("drainage_node_ID", "N1"),
("scheme_name", "flush_with_valve_control"),
],
)
assert response.status_code == 200
assert captured["modify_valve_opening"] is None
assert captured["valve_control"] == {
"V1": {"status": "ACTIVE", "setting": "2.5"},
"V2": {"status": "CLOSED"},
}
def test_flushing_endpoint_requires_setting_for_active_valve(monkeypatch):
module = _load_simulation_module(monkeypatch)
client = _build_authenticated_client(module)
response = client.post(
"/api/v1/flushing-analyses",
params={
"network": "demo",
"start_time": "2025-01-02T03:04:05+08:00",
"valves": "V1",
"valve_statuses": "ACTIVE",
"drainage_node_ID": "N1",
"scheme_name": "flush_without_active_setting",
},
)
assert response.status_code == 422
def test_flushing_endpoint_rejects_mixed_valve_control_modes(monkeypatch):
module = _load_simulation_module(monkeypatch)
client = _build_authenticated_client(module)
response = client.post(
"/api/v1/flushing-analyses",
params={
"network": "demo",
"start_time": "2025-01-02T03:04:05+08:00",
"valves": "V1",
"valves_k": 0.5,
"valve_statuses": "ACTIVE",
"valve_settings": "2.5",
"drainage_node_ID": "N1",
"scheme_name": "flush_with_mixed_controls",
},
)
assert response.status_code == 422
def test_flushing_endpoint_rejects_settings_without_statuses(monkeypatch):
module = _load_simulation_module(monkeypatch)
client = _build_authenticated_client(module)
response = client.post(
"/api/v1/flushing-analyses",
params={
"network": "demo",
"start_time": "2025-01-02T03:04:05+08:00",
"valves": "V1",
"valves_k": 0.5,
"valve_settings": "2.5",
"drainage_node_ID": "N1",
"scheme_name": "flush_with_orphan_settings",
},
)
assert response.status_code == 422
def test_flushing_endpoint_requires_drainage_node(monkeypatch):
module = _load_simulation_module(monkeypatch)
client = _build_authenticated_client(module)
response = client.post(
"/api/v1/flushing-analyses",
params={
"network": "demo",
"start_time": "2025-01-02T03:04:05+08:00",
"scheme_name": "flush_without_drainage_node",
},
)
assert response.status_code == 422
def test_contaminant_endpoint_passes_current_username(monkeypatch):
module = _load_simulation_module(monkeypatch)
captured = {}
@@ -6,7 +6,10 @@ from uuid import uuid4
import pytest
from cryptography.fernet import InvalidToken
from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository
from app.infra.db.metadb.repositories.metadata_repository import (
MetadataRepository,
_normalize_postgres_dsn,
)
class _DummyResult:
@@ -124,6 +127,12 @@ def test_encrypted_dsn_decrypts_without_migration(monkeypatch):
session.commit.assert_not_awaited()
def test_psycopg_sqlalchemy_dsn_is_normalized_for_direct_psycopg_clients():
assert _normalize_postgres_dsn(
"postgresql+psycopg://user:secret@db.example/project"
) == "postgresql://user:secret@db.example/project"
def test_upsert_project_database_config_encrypts_plaintext_dsn(monkeypatch):
project_id = uuid4()
session = SimpleNamespace(
+85
View File
@@ -0,0 +1,85 @@
import pytest
from psycopg.conninfo import conninfo_to_dict
from app.infra.db.project_routing import (
ActiveProjectRouting,
activate_project_routing,
get_active_project_routing,
get_project_pgconn_string,
get_project_timescale_pgconn_string,
)
def _routing(project_code: str = "project_a") -> ActiveProjectRouting:
return ActiveProjectRouting(
project_code=project_code,
business_dsn=(
"postgresql://biz_user:biz_password@biz.example:5432/biz_database"
"?sslmode=require"
),
timescale_dsn=(
"postgresql://ts_user:ts_password@timescale.example:5433/ts_database"
"?sslmode=require"
),
)
def test_project_database_uses_exact_routing_dsn_for_project_code() -> None:
routing = _routing()
with activate_project_routing(routing):
assert get_project_pgconn_string("project_a") == routing.business_dsn
assert (
get_project_timescale_pgconn_string("project_a")
== routing.timescale_dsn
)
def test_business_template_keeps_server_and_timescale_ignores_legacy_db_name() -> None:
with activate_project_routing(_routing()):
business = conninfo_to_dict(get_project_pgconn_string("project_a_template"))
timescale = conninfo_to_dict(
get_project_timescale_pgconn_string("temporary_scheme")
)
assert business == {
"user": "biz_user",
"password": "biz_password",
"dbname": "project_a_template",
"host": "biz.example",
"port": "5432",
"sslmode": "require",
}
assert timescale == {
"user": "ts_user",
"password": "ts_password",
"dbname": "ts_database",
"host": "timescale.example",
"port": "5433",
"sslmode": "require",
}
def test_project_routing_is_nested_and_request_local() -> None:
first = _routing("project_a")
second = _routing("project_b")
assert get_active_project_routing() is None
with activate_project_routing(first):
assert get_active_project_routing() is first
with activate_project_routing(second):
assert get_active_project_routing() is second
assert get_active_project_routing() is first
assert get_active_project_routing() is None
def test_timescale_access_requires_iot_routing_in_project_request() -> None:
business_only = _routing()
business_only = ActiveProjectRouting(
project_code=business_only.project_code,
business_dsn=business_only.business_dsn,
)
with activate_project_routing(business_only):
with pytest.raises(RuntimeError, match="TimescaleDB routing is not configured"):
get_project_timescale_pgconn_string()
+3 -1
View File
@@ -35,7 +35,9 @@ class _FakeConnection:
def test_query_scheme_list_pushes_scheme_type_into_sql(monkeypatch):
cursor = _FakeCursor()
monkeypatch.setattr(
scheme_management, "get_pgconn_string", lambda db_name=None: "postgres://test"
scheme_management,
"get_project_pgconn_string",
lambda db_name=None: "postgres://test",
)
monkeypatch.setattr(
scheme_management.psycopg, "connect", lambda _conn_string: _FakeConnection(cursor)
@@ -1,3 +1,4 @@
import inspect
import json
from datetime import timedelta
@@ -7,6 +8,70 @@ from app.infra.db.timescaledb.repositories.scheme import SchemeRepository
from app.services.time_api import parse_utc_time
def test_run_simulation_exposes_explicit_valve_control():
from app.services import simulation
parameters = inspect.signature(simulation.run_simulation).parameters
assert "valve_control" in parameters
def test_apply_valve_control_matches_run_simulation_ex_semantics(monkeypatch):
from app.services import simulation
updates: dict[str, dict] = {}
monkeypatch.setattr(
simulation,
"get_status",
lambda project_name, valve_name: {
"link": valve_name,
"status": "OPEN",
"setting": 1.0,
},
)
monkeypatch.setattr(
simulation,
"set_status",
lambda project_name, changeset: updates.update(
{
changeset.operations[0]["link"]: changeset.operations[0].copy()
}
),
)
simulation._apply_valve_control(
"demo",
{
"V-status": {"status": "ACTIVE"},
"V-setting": {"setting": 2.5},
"V-closed": {"status": "ACTIVE", "setting": 9.0, "k": 0},
"V-k": {"status": "ACTIVE", "setting": 9.0, "k": 0.5},
},
)
assert updates["V-status"] == {
"link": "V-status",
"status": "ACTIVE",
"setting": 1.0,
}
assert updates["V-setting"] == {
"link": "V-setting",
"status": "OPEN",
"setting": 2.5,
}
assert updates["V-closed"] == {
"link": "V-closed",
"status": "CLOSED",
"setting": 9.0,
}
assert updates["V-k"] == {
"link": "V-k",
"status": "ACTIVE",
"setting": 0.1036 * pow(0.5, -3.105),
}
def _node_result(periods: int) -> list[dict]:
return [
{
+26 -2
View File
@@ -45,9 +45,11 @@ class _FakeConnection:
@pytest.fixture(autouse=True)
def clear_native_connections():
connection.g_conn_dict.clear()
connection.g_conninfo_dict.clear()
connection._project_locks.clear()
yield
connection.g_conn_dict.clear()
connection.g_conninfo_dict.clear()
connection._project_locks.clear()
@@ -61,6 +63,10 @@ def test_is_project_open_drops_closed_cached_connection():
def test_open_connection_reuses_healthy_cached_connection(monkeypatch):
cached = _FakeConnection()
connection.g_conn_dict["fengyang"] = cached
connection.g_conninfo_dict["fengyang"] = "dbname=fengyang"
monkeypatch.setattr(
connection, "get_project_pgconn_string", lambda db_name: f"dbname={db_name}"
)
def fail_connect(*, conninfo, autocommit):
raise AssertionError("cached connection should be reused")
@@ -84,7 +90,7 @@ def test_read_all_reopens_closed_cached_connection(monkeypatch):
monkeypatch.setattr(connection.pg, "connect", fake_connect)
monkeypatch.setattr(
connection, "get_pgconn_string", lambda db_name: f"dbname={db_name}"
connection, "get_project_pgconn_string", lambda db_name: f"dbname={db_name}"
)
rows = database.read_all("fengyang", "select * from times")
@@ -99,6 +105,7 @@ def test_read_all_reopens_cached_connection_when_health_check_fails(monkeypatch)
stale = _FakeConnection(fail_ping=True)
fresh = _FakeConnection(rows=[{"scheme_name": "base"}])
connection.g_conn_dict["fengyang"] = stale
connection.g_conninfo_dict["fengyang"] = "dbname=fengyang"
opened = []
@@ -108,7 +115,7 @@ def test_read_all_reopens_cached_connection_when_health_check_fails(monkeypatch)
monkeypatch.setattr(connection.pg, "connect", fake_connect)
monkeypatch.setattr(
connection, "get_pgconn_string", lambda db_name: f"dbname={db_name}"
connection, "get_project_pgconn_string", lambda db_name: f"dbname={db_name}"
)
rows = database.read_all("fengyang", "select * from scheme_list")
@@ -119,3 +126,20 @@ def test_read_all_reopens_cached_connection_when_health_check_fails(monkeypatch)
assert opened == [("dbname=fengyang", True)]
assert connection.g_conn_dict["fengyang"] is fresh
assert fresh.executed == ["select * from scheme_list"]
def test_open_connection_replaces_cache_when_project_dsn_changes(monkeypatch):
cached = _FakeConnection()
fresh = _FakeConnection()
connection.g_conn_dict["fengyang"] = cached
connection.g_conninfo_dict["fengyang"] = "host=old dbname=fengyang"
monkeypatch.setattr(
connection,
"get_project_pgconn_string",
lambda db_name: f"host=new dbname={db_name}",
)
monkeypatch.setattr(connection.pg, "connect", lambda **_kwargs: fresh)
assert connection.open_connection("fengyang") is fresh
assert cached.close_calls == 1
assert connection.g_conninfo_dict["fengyang"] == "host=new dbname=fengyang"
+21
View File
@@ -0,0 +1,21 @@
from app.native.wndb import s2_junctions
def test_get_junction_binds_untrusted_identifier(monkeypatch) -> None:
calls: list[tuple[str, str, tuple[str, ...]]] = []
malicious_id = "J-1'; DELETE FROM junctions; --"
def fake_try_read(name, statement, params):
calls.append((name, statement, params))
return None
monkeypatch.setattr(s2_junctions, "try_read", fake_try_read)
assert s2_junctions.get_junction("project_a", malicious_id) == {}
assert calls == [
(
"project_a",
"select * from junctions where id = %s",
(malicious_id,),
)
]