Repository navigation
API Reference
Brian Rodriguez edited this page Dec 5, 2025
·
2 revisions
Main application entry point.
def main():
"""Initialize and run the Dash application with Modbus thread."""Starts:
- Modbus reader thread
- Dash web server on configured HOST:PORT
SQLite database operations.
DB_FILE: str # Database file path from environment
DATA_RETENTION_DAYS: int # Data retention perioddef ensure_table_exists(table_name: str) -> None:
"""
Create table if it doesn't exist.
Args:
table_name: Name of table to create (format: data_YYYY_MM)
Columns:
- id: INTEGER PRIMARY KEY AUTOINCREMENT
- timestamp: TEXT NOT NULL
- temperature: REAL NOT NULL
- humidity: REAL NOT NULL
"""
def get_table_name() -> str:
"""
Get current month's table name.
Returns:
Table name in format data_YYYY_MM
"""
def cleanup_old_data() -> None:
"""
Delete tables older than DATA_RETENTION_DAYS.
Runs automatically every 24 hours.
"""
def fetch_data() -> pandas.DataFrame:
"""
Fetch all data from all tables.
Returns:
DataFrame with columns: timestamp, temperature, humidity
Empty DataFrame if no data
"""Modbus RTU communication and data logging.
All values from environment variables:
MODBUS_PORT: str # Serial port path
MODBUS_SLAVE_ID: int # Slave address (1-247)
MODBUS_BAUDRATE: int # Baud rate
MODBUS_BYTESIZE: int # Data bits
MODBUS_PARITY: str # Parity (N/E/O)
MODBUS_STOPBITS: int # Stop bits
MODBUS_TIMEOUT: int # Timeout seconds
MODBUS_FUNCTION_CODE: int # Function code (3/4)
MODBUS_REGISTER_TEMP: int # Temperature register
MODBUS_REGISTER_HUMIDITY: int # Humidity registerdef _initialize_instrument() -> minimalmodbus.Instrument:
"""
Initialize Modbus instrument (lazy).
Called on first use, not during import.
Returns:
Configured Modbus instrument
"""
def read_modbus_data() -> None:
"""
Main thread function for reading sensor data.
Reads sensor every 1 second.
Batch writes to database every 30 seconds.
Runs cleanup every 24 hours.
Handles errors and reconnection automatically.
"""
def start_modbus_thread() -> threading.Thread:
"""
Start Modbus reader in background thread.
Returns:
Thread object (daemon=True)
"""Psychrometric calculations using psychrolib.
def calculate_psychrometric_properties(
temperature: float,
humidity: float
) -> tuple[float, float, float]:
"""
Calculate psychrometric properties.
Args:
temperature: Dry-bulb temperature (Celsius)
humidity: Relative humidity (0-100%)
Returns:
Tuple of (dew_point, absolute_humidity, enthalpy)
- dew_point: Dew point temperature (Celsius)
- absolute_humidity: Water vapor density (g/m³)
- enthalpy: Specific enthalpy (kJ/kg)
Raises:
ValueError: If inputs are out of valid range
"""Valid ranges:
- Temperature: -50 to 100 degrees C
- Humidity: 0 to 100 percent
Multi-language support.
TRANSLATIONS: dict # All translation strings
def t(key: str, lang: str = "nl") -> str:
"""
Get translated string.
Args:
key: Translation key
lang: Language code (nl/en/de/fr/es)
Returns:
Translated string
Falls back to Dutch if language not found
Returns key if key not found
"""Translation keys include:
- UI labels: title, current_values, temperature, humidity
- Graph labels: temp_graph, hum_graph, time, degrees, percent
- Psychrometric: dew_point, absolute_humidity, enthalpy
- Table: data_table, timestamp
- Language: language
Dash UI layout components.
def create_layout(app: dash.Dash) -> html.Div:
"""
Create main dashboard layout.
Args:
app: Dash application instance
Returns:
Layout div containing all components
Components:
- Language selector (dropdown)
- Current values display (cards)
- Psychrometric properties (cards)
- Temperature graph (plotly)
- Humidity graph (plotly)
- Data table (dash_table)
- Hidden interval components for updates
"""Dash callback functions for interactivity.
@app.callback(...)
def update_current_values(n: int, lang: str) -> tuple:
"""Update current temperature and humidity display."""
@app.callback(...)
def update_psychrometric(n: int, lang: str) -> tuple:
"""Update psychrometric calculations display."""
@app.callback(...)
def update_graphs(n: int, period: str, lang: str) -> tuple:
"""Update temperature and humidity graphs."""
@app.callback(...)
def update_table(n: int, period: str, lang: str) -> tuple:
"""Update data table."""
@app.callback(...)
def update_ui_language(lang: str) -> tuple:
"""Update all UI text when language changes."""All callbacks:
- Accept interval trigger (n)
- Accept language selection (lang)
- Return updated component properties
- Handle errors gracefully
- Return empty/default on no data
Complete list with defaults:
# Modbus Configuration
MODBUS_PORT=COM11
MODBUS_SLAVE_ID=1
MODBUS_BAUDRATE=9600
MODBUS_BYTESIZE=8
MODBUS_PARITY=N
MODBUS_STOPBITS=1
MODBUS_TIMEOUT=1
MODBUS_FUNCTION_CODE=4
MODBUS_REGISTER_TEMP=1
MODBUS_REGISTER_HUMIDITY=2
# Database Configuration
DB_FILE=sensor_data.db
DATA_RETENTION_DAYS=30
# Web Server Configuration
HOST=0.0.0.0
PORT=8050
DEBUG=FalseTable name format: data_YYYY_MM
CREATE TABLE IF NOT EXISTS data_2025_12 (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
temperature REAL NOT NULL,
humidity REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_timestamp
ON data_2025_12(timestamp);# Timestamp format
timestamp: str = "2025-12-05 14:30:45"
# Sensor values
temperature: float # Celsius, range -50 to 100
humidity: float # Percent, range 0 to 100
# Psychrometric properties
dew_point: float # Celsius
absolute_humidity: float # g/m³
enthalpy: float # kJ/kgAll modules implement try-except blocks:
try:
# Normal operation
except Exception as e:
print(f"Error: {e}")
# Fallback behavior
# Continue operation if possibleErrors logged to terminal output. Application continues running on errors. Database operations use WAL mode for robustness.