vitor-aignosi
1bddde17f4
Refactor Activities and API Integration for PI Web API - Reintroduced the API import in the Activities class for proper integration. - Cleaned up whitespace and formatting in the API class and related tests for improved readability. - Updated test cases to ensure consistent formatting in error messages and configuration structures for PI Web API. - Enhanced connectors_config.py with additional whitespace for better organization.
92 lines
3.6 KiB
Python
92 lines
3.6 KiB
Python
import json
|
|
from os import getenv
|
|
from typing import Any
|
|
|
|
|
|
def build_mlflow_config() -> dict[str, Any]:
|
|
"""
|
|
Build MLFlow server configuration from environment variables.
|
|
|
|
This function constructs an MLFlow configuration dictionary from
|
|
environment variables with sensible defaults for local development.
|
|
It handles server connection and authentication parameters.
|
|
|
|
Environment Variables:
|
|
MLFLOW_HOST: MLFlow server hostname (default: http://localhost)
|
|
MLFLOW_PORT: MLFlow server port (default: 5080)
|
|
MLFLOW_USERNAME: MLFlow username (default: aignosi)
|
|
MLFLOW_PASSWORD: MLFlow password (default: aignosi)
|
|
|
|
Returns:
|
|
dict: MLFlow configuration dictionary with all required parameters
|
|
"""
|
|
return {
|
|
'host': getenv('MLFLOW_HOST', 'http://localhost'),
|
|
'port': int(getenv('MLFLOW_PORT', '5080')),
|
|
'username': getenv('MLFLOW_USERNAME', 'aignosi'),
|
|
'password': getenv('MLFLOW_PASSWORD', 'aignosi'),
|
|
}
|
|
|
|
|
|
def build_opc_config() -> dict[str, Any]:
|
|
"""
|
|
Build OPC server configuration from environment variables.
|
|
|
|
This function constructs an OPC server configuration dictionary from
|
|
environment variables. It supports both single server and multi-server
|
|
configurations with flexible parameter handling.
|
|
|
|
Environment Variables:
|
|
OPC_CONFIG: JSON string containing multiple OPC server configurations
|
|
OPC_ID: OPC server ID (fallback, default: 1)
|
|
OPC_URL: Single OPC server URL (fallback, default: opc.tcp://localhost:4840)
|
|
OPC_SERVER_URI: Single OPC server URI (fallback, default: opc.tcp://localhost:4840)
|
|
OPC_CERT_PATH: Client certificate path (fallback, default: None)
|
|
OPC_PRIVATE_KEY_PATH: Client private key path (fallback, default: None)
|
|
OPC_SERVER_CERT_PATH: Server certificate path (fallback, default: None)
|
|
OPC_RECONNECTION_INTERVAL: Reconnection interval in seconds (fallback, default: 120)
|
|
|
|
Returns:
|
|
dict: OPC server configuration dictionary
|
|
"""
|
|
opc_raw = getenv('OPC_CONFIG', None)
|
|
|
|
if opc_raw:
|
|
return json.loads(opc_raw)
|
|
|
|
return {
|
|
getenv('OPC_ID', '1'): {
|
|
'id': getenv('OPC_ID', '1'),
|
|
'server_name': getenv('OPC_SERVER_NAME', 'default_server'),
|
|
'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'),
|
|
'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'),
|
|
'cert_path': getenv('OPC_CERT_PATH', None),
|
|
'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None),
|
|
'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None),
|
|
'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120')),
|
|
}
|
|
}
|
|
|
|
|
|
def build_minio_config() -> dict[str, Any]:
|
|
"""
|
|
Build MinIO (S3-compatible) configuration from environment variables.
|
|
|
|
Environment Variables:
|
|
MINIO_ENDPOINT: MinIO endpoint including scheme (default: http://localhost:9000)
|
|
MINIO_ACCESS_KEY: Access key (default: minioadmin)
|
|
MINIO_SECRET_KEY: Secret key (default: minioadmin)
|
|
MINIO_REGION: Region name for S3 client (default: us-east-1)
|
|
MINIO_BUCKET_DEFAULT: Default bucket for uploads (default: laborious)
|
|
|
|
Returns:
|
|
dict: MinIO configuration dictionary
|
|
"""
|
|
return {
|
|
'endpoint_url': getenv('MINIO_ENDPOINT_URL', 'http://localhost:9000'),
|
|
'access_key': getenv('MINIO_ACCESS_KEY', 'minioadmin'),
|
|
'secret_key': getenv('MINIO_SECRET_KEY', 'minioadmin'),
|
|
'region_name': getenv('MINIO_REGION_NAME', 'us-east-1'),
|
|
'default_bucket': getenv('MINIO_DEFAULT_BUCKET', 'laborious'),
|
|
}
|