Wolfgang Hottgenroth
3a4cd499a5
All checks were successful
ci/woodpecker/tag/woodpecker Pipeline was successful
96 lines
2.4 KiB
Python
96 lines
2.4 KiB
Python
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
from pydantic import BaseModel, Field, field_validator
|
|
import yaml
|
|
from loguru import logger
|
|
|
|
|
|
class RegisterConfig(BaseModel):
|
|
"""Modbus Register Configuration"""
|
|
address: int
|
|
attribute: str
|
|
name: str
|
|
unit: str
|
|
register_type: str
|
|
data_type: str
|
|
adaptor: str
|
|
|
|
|
|
class OutputConfig(BaseModel):
|
|
"""Output Configuration for Modbus Devices"""
|
|
name: str
|
|
enabled: bool = Field(default=True)
|
|
scan_rate: Optional[int] = Field(default=60)
|
|
publish_topic: str
|
|
raw_output: Optional[bool] = Field(default=False)
|
|
slave_id: int
|
|
registers: List[RegisterConfig]
|
|
|
|
class InputConfig(BaseModel):
|
|
"""Input Configuration for Modbus Devices (MQTT -> Modbus)"""
|
|
name: str
|
|
enabled: bool = Field(default=True)
|
|
subscribe_topic: str
|
|
slave_id: int
|
|
address: int
|
|
register_type: str
|
|
|
|
|
|
class MqttConfig(BaseModel):
|
|
"""MQTT Configuration"""
|
|
broker: str
|
|
port: int
|
|
|
|
|
|
class ModbusConfig(BaseModel):
|
|
"""Modbus Configuration"""
|
|
gateway: str
|
|
|
|
|
|
class GlobalConfig(BaseModel):
|
|
"""Global settings"""
|
|
scan_interval: int
|
|
log_level: str
|
|
|
|
|
|
class Config(BaseModel):
|
|
"""Main Configuration"""
|
|
global_: GlobalConfig = Field(alias="global")
|
|
mqtt: MqttConfig
|
|
modbus: ModbusConfig
|
|
input: List[InputConfig]
|
|
output: List[OutputConfig]
|
|
|
|
@classmethod
|
|
def load_from_file(cls, config_path: Optional[str] = None) -> 'Config':
|
|
"""
|
|
Load configuration from YAML file with environment variable substitution.
|
|
|
|
Args:
|
|
config_path: Path to config file. If None, uses CFG_FILE environment variable.
|
|
|
|
Returns:
|
|
Config instance
|
|
"""
|
|
if config_path is None:
|
|
config_path = os.getenv('CFG_FILE')
|
|
if config_path is None:
|
|
raise ValueError("Config path not provided and CFG_FILE environment variable not set")
|
|
|
|
config_file = Path(config_path)
|
|
if not config_file.exists():
|
|
raise FileNotFoundError(f"Configuration file not found: {config_path}")
|
|
|
|
# Read YAML file
|
|
with open(config_file, 'r', encoding='utf-8') as f:
|
|
yaml_content = f.read()
|
|
|
|
# Parse YAML
|
|
config_dict = yaml.safe_load(yaml_content)
|
|
|
|
logger.info(f"Configuration loaded from: {config_path}")
|
|
return cls(**config_dict)
|
|
|