bjkim1
21ebb16fb7
- SPipe_server_2026-10-30.zip 내용을 풀어서 등록 (압축 파일은 올리지 않음) - README에 데모 v2 서버 업데이트 순서 추가, 저장소에 없는 문서 목차 정리 - 패치내역을 docs/로 이동, _worklog.md 추가 - 임시 파일(backend/temp.*)과 API 키가 든 파일은 제외
115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
"""설정파일(config.yaml) 스키마와 로더.
|
|
|
|
사내 관행값·발주처 기준이 필요한 TODO 항목은 §9 표를 참고. 전부 이 모듈의
|
|
기본값으로 노출되어 있으며, config.yaml로 덮어쓴다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
import yaml
|
|
|
|
|
|
@dataclass
|
|
class ConnectionsPerBlock:
|
|
min: int = 500 # TODO: 편람에 수치 규정 없음. 사내 관행값 확정 필요
|
|
max: int = 3000
|
|
|
|
|
|
@dataclass
|
|
class DemandPerBlock:
|
|
max_cmd: Optional[float] = None # m3/day, null이면 미적용
|
|
|
|
|
|
@dataclass
|
|
class PartitionConfig:
|
|
level: str = "small" # large | medium | small
|
|
k_min: int = 4
|
|
k_max: int = 20
|
|
method: str = "spectral" # spectral | louvain
|
|
flow_weight_alpha: float = 0.7
|
|
|
|
|
|
@dataclass
|
|
class ConstraintsConfig:
|
|
connections_per_block: ConnectionsPerBlock = field(default_factory=ConnectionsPerBlock)
|
|
max_pipe_length_per_block: float = 30000.0 # m, TODO
|
|
demand_per_block: DemandPerBlock = field(default_factory=DemandPerBlock)
|
|
elevation_band_m: float = 30.0
|
|
large_consumer_demand_lps: float = 5.0 # TODO
|
|
large_consumer_policy: str = "isolate" # isolate | keep_together
|
|
sources_in_separate_blocks: bool = True
|
|
|
|
|
|
@dataclass
|
|
class HydraulicsConfig:
|
|
duration_hours: int = 24
|
|
timestep_minutes: int = 60
|
|
min_pressure_m: float = 15.0 # TODO: 수도법 시행규칙 / 상수도설계기준 확인 후 확정
|
|
max_pressure_m: float = 71.0 # TODO
|
|
max_water_age_hours: float = 48.0 # TODO
|
|
allow_isolated_nodes: bool = False
|
|
|
|
|
|
@dataclass
|
|
class RefineConfig:
|
|
max_iterations: int = 20
|
|
moves_per_iteration: int = 50
|
|
|
|
|
|
@dataclass
|
|
class ScoringWeights:
|
|
n_valves: float = 0.25
|
|
n_meters: float = 0.15
|
|
pressure_violations: float = 0.30
|
|
max_water_age: float = 0.15
|
|
boundary_pipe_length: float = 0.15
|
|
|
|
|
|
@dataclass
|
|
class ScoringConfig:
|
|
weights: ScoringWeights = field(default_factory=ScoringWeights)
|
|
|
|
|
|
@dataclass
|
|
class BoundaryConfig:
|
|
# 블록당 유입지점(유량계) 목표 개수. 편람에 수치 없음, 통상 1개 예비 포함 2개. TODO
|
|
inlets_per_block: int = 1
|
|
|
|
|
|
@dataclass
|
|
class Config:
|
|
crs: str = "EPSG:5186"
|
|
partition: PartitionConfig = field(default_factory=PartitionConfig)
|
|
constraints: ConstraintsConfig = field(default_factory=ConstraintsConfig)
|
|
hydraulics: HydraulicsConfig = field(default_factory=HydraulicsConfig)
|
|
refine: RefineConfig = field(default_factory=RefineConfig)
|
|
scoring: ScoringConfig = field(default_factory=ScoringConfig)
|
|
boundary: BoundaryConfig = field(default_factory=BoundaryConfig)
|
|
|
|
@classmethod
|
|
def from_yaml(cls, path: str | Path) -> "Config":
|
|
raw = yaml.safe_load(Path(path).read_text(encoding="utf-8")) or {}
|
|
return cls.from_dict(raw)
|
|
|
|
@classmethod
|
|
def from_dict(cls, raw: dict[str, Any]) -> "Config":
|
|
cfg = cls()
|
|
_apply(cfg, raw)
|
|
return cfg
|
|
|
|
|
|
def _apply(instance: Any, data: dict[str, Any]) -> None:
|
|
for key, value in data.items():
|
|
if not hasattr(instance, key):
|
|
continue
|
|
current = getattr(instance, key)
|
|
if dataclasses.is_dataclass(current) and isinstance(value, dict):
|
|
_apply(current, value)
|
|
else:
|
|
setattr(instance, key, value)
|