initial for multiple devices, introduce real configuration
This commit is contained in:
@@ -1,33 +1,76 @@
|
||||
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 Config:
|
||||
OPTIONS = {
|
||||
'mqtt': [ 'login',
|
||||
'password',
|
||||
'ca',
|
||||
'cert',
|
||||
'key',
|
||||
'broker',
|
||||
'port',
|
||||
'meterPublishTopic',
|
||||
'meterPublishPeriod',
|
||||
'relaisSubscribeTopic' ],
|
||||
'modbus': [ 'gateway' ]
|
||||
}
|
||||
|
||||
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
|
||||
publish_topic: str
|
||||
publish_period: int
|
||||
slave_id: int
|
||||
registers: List[RegisterConfig]
|
||||
|
||||
|
||||
class MqttConfig(BaseModel):
|
||||
"""MQTT Configuration"""
|
||||
broker: str
|
||||
port: int
|
||||
|
||||
|
||||
class ModbusConfig(BaseModel):
|
||||
"""Modbus Configuration"""
|
||||
gateway: str
|
||||
|
||||
|
||||
class Config(BaseModel):
|
||||
"""Main Configuration"""
|
||||
mqtt: MqttConfig
|
||||
modbus: ModbusConfig
|
||||
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)
|
||||
|
||||
def __init__(self):
|
||||
self.values = {}
|
||||
for section, keys in Config.OPTIONS.items():
|
||||
self.values[section] = {}
|
||||
for key in keys:
|
||||
varname = f"{section}__{key}".upper()
|
||||
try:
|
||||
self.values[section][key] = os.environ[varname]
|
||||
logger.info(f"Config: {section} {key} -> {self.values[section][key]}")
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
def __getitem__(self, section):
|
||||
return self.values[section]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user