Skip to content

API Reference

Jump to bottom
Brian Rodriguez edited this page Dec 9, 2025 · 2 revisions

API Reference

Core Modules

app.py

Main application entry point using Waitress WSGI production server.

if __name__ == '__main__':
    serve(server, host='0.0.0.0', port=8050, threads=8)

Starts:

  1. Database initialization and cleanup
  2. Modbus reader thread (background daemon)
  3. Waitress production server (8 threads, cross-platform)

database.py

SQLite database operations.

Constants

DB_FILE: str  # Database file path from environment (default: src/modbus_sensor_data.db)
DATA_RETENTION_DAYS: int  # Data retention period (0 = infinite)
DEBUG_LOGGING: bool  # Enable debug output

Functions

def ensure_table_exists(cursor, table_name: str) -> None:
    """
    Create table-per-day partition if it doesn't exist.
    
    Args:
        cursor: SQLite cursor
        table_name: Name of table (format: measurements_YYYYMMDD)
    
    Columns:
        - id: INTEGER PRIMARY KEY AUTOINCREMENT
        - timestamp: INTEGER NOT NULL (Unix timestamp)
        - temperature: REAL NOT NULL
        - humidity: REAL NOT NULL
        - dewpoint: REAL (calculated)
        - absolute_humidity: REAL (calculated, g/m³)
    """

def get_table_name(date=None) -> str:
    """
    Get table name for specific date (table-per-day partitioning).
    
    Args:
        date: datetime object (default: now)
    
    Returns:
        Table name in format measurements_YYYYMMDD
    """

def build_union_query(cursor, columns, start_timestamp=None, end_timestamp=None) -> tuple:
    """
    Build optimized UNION ALL query across relevant daily tables.
    
    Args:
        cursor: SQLite cursor
        columns: List of column names to select
        start_timestamp: Optional Unix timestamp filter start
        end_timestamp: Optional Unix timestamp filter end
    
    Returns:
        Tuple of (query_string, table_count)
        Returns (None, None) if no tables found
    
    Features:
        - Only queries relevant tables for time range
        - Automatic table detection and sorting
        - Optimized with indexed timestamp filtering
    """

def cleanup_old_data() -> None:
    """
    Delete daily tables older than DATA_RETENTION_DAYS.
    
    Runs automatically:
        - At startup (via app.py)
        - Every 24 hours (via modbus_reader.py)
    
    Skipped if DATA_RETENTION_DAYS = 0 (infinite retention)
    """

def init_database() -> None:
    """
    Initialize database with optimized settings.
    
    Actions:
        - Creates database directory if needed
        - Enables WAL mode for better concurrency
        - Sets PRAGMA optimizations (NORMAL sync, memory temp_store)
        - Creates today's measurement table
        - Prints configuration summary
    """

def get_all_measurement_tables(cursor) -> list:
    """
    Get all measurement tables sorted by date.
    
    Returns:
        List of table names (measurements_YYYYMMDD format)
    """

modbus_reader.py

Modbus RTU communication and data logging.

Configuration

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 register

Functions

def _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.py

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

translations.py

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

layout.py

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
    """

callbacks.py

Dash callback functions for interactivity.

@app.callback(...)
def update_language(lang: str) -> tuple:
    """
    Update all UI labels when language changes.
    
    Updates:
        - All text labels (14 outputs)
        - Dropdown options with translations
        - Tooltip content
    """

@app.callback(...)
def update_graph(n, time_range_minutes, relayout_data, lang, stored_relayout) -> tuple:
    """
    Update main graphs and current values.
    
    Features:
        - 5 synchronized graphs (temperature, humidity, dewpoint, abs_humidity, comfort_score)
        - Adaptive downsampling for performance (1min to 1day intervals)
        - Comfort level calculation using Humidex formula
        - Dynamic Y-axis ranges with padding
        - Preserves zoom/pan state between updates
    
    Returns:
        Tuple of (figure, data_count, current_temp, current_humidity, 
                  current_dewpoint, current_abs_humidity, comfort_text, 
                  comfort_score, comfort_icon, relayout_data)
    """

@app.callback(...)
def update_psychrometric_chart(n: int, lang: str) -> figure:
    """
    Update psychrometric (Mollier) chart with current conditions.
    
    Shows:
        - Current temperature/humidity point
        - Relative humidity curves (10-100%)
        - Comfort zone (20-26°C, 30-60% RH)
        - Saturation line
    """

@app.callback(...)
def toggle_modal(open_clicks, close_clicks, apply_clicks, current_style) -> dict:
    """Toggle historical data range selection modal."""

@app.callback(...)
def handle_range_selection(...) -> tuple:
    """
    Handle time range selection (preset buttons or custom range).
    
    Presets: 5min, 10min, 30min, 1h, 2h, 6h, 12h, 24h, 2d, week, month
    Custom: User-specified start/end date and time
    """

@app.callback(...)
def load_historical_data(range_value, lang) -> tuple:
    """
    Load and prepare historical data for psychrometric replay.
    
    Features:
        - Resamples to 1-minute intervals if range > 1 minute
        - Generates slider marks based on time range
        - Stores data for slider interaction
    
    Returns:
        Tuple of (data_store, slider_min, slider_max, slider_value, marks, style)
    """

@app.callback(...)
def update_historical_view(slider_value, drag_value, lang, stored_data) -> tuple:
    """
    Update psychrometric chart with historical point during slider movement.
    
    Features:
        - Live update during drag
        - Shows timestamp and conditions for selected point
        - Uses specialized historical chart variant
    """

All callbacks:

  • Handle errors gracefully with try-except
  • Support multi-language UI
  • Return no_update when appropriate
  • Use DEBUG_LOGGING for troubleshooting
  • Preserve UI state where appropriate

Environment Variables

Complete list with defaults:

# Modbus Configuration
MODBUS_PORT=COM11                    # Serial port (COM11, /dev/ttyUSB0, etc.)
MODBUS_SLAVE_ID=1                    # Modbus slave address (1-247)
MODBUS_BAUDRATE=9600                 # Baud rate
MODBUS_BYTESIZE=8                    # Data bits (5-8)
MODBUS_PARITY=N                      # Parity: N=None, E=Even, O=Odd
MODBUS_STOPBITS=1                    # Stop bits (1, 1.5, 2)
MODBUS_TIMEOUT=1                     # Timeout in seconds
MODBUS_FUNCTION_CODE=4               # 3=Holding, 4=Input registers
MODBUS_REGISTER_TEMP=1               # Temperature register address
MODBUS_REGISTER_HUMIDITY=2           # Humidity register address

# Database Configuration
DATABASE_FILE=src/modbus_sensor_data.db  # SQLite database path
DATA_RETENTION_DAYS=0                    # 0 = infinite, >0 = days to keep

# Application Configuration
APP_HOST=127.0.0.1                   # Web server host (0.0.0.0 for all interfaces)
APP_PORT=8050                        # Web server port
APP_DEBUG=False                      # Flask debug mode (use False in production)
DEBUG_LOGGING=False                  # Enable detailed debug output
TIMEZONE=Europe/Amsterdam            # Timezone for timestamps
DEFAULT_LANGUAGE=EN                  # Default UI language (NL/EN/DE/FR/ES)

Note: App now uses Waitress production server, not Flask development server.

Data Models

Database Schema

Table name format: measurements_YYYYMMDD (table-per-day partitioning)

CREATE TABLE IF NOT EXISTS measurements_20251209 (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    timestamp INTEGER NOT NULL,           -- Unix timestamp (seconds since epoch)
    temperature REAL NOT NULL,            -- Celsius
    humidity REAL NOT NULL,               -- Percentage (0-100)
    dewpoint REAL,                        -- Celsius (calculated)
    absolute_humidity REAL                -- g/m³ (calculated)
);

CREATE INDEX IF NOT EXISTS idx_measurements_20251209_timestamp 
ON measurements_20251209(timestamp);

Database Optimizations:

  • WAL (Write-Ahead Logging) mode for better concurrency
  • PRAGMA synchronous=NORMAL for performance
  • PRAGMA temp_store=MEMORY
  • Table-per-day partitioning for efficient cleanup and queries

Data Types

# Timestamp format (stored as Unix integer)
timestamp: int = 1733757045  # Seconds since epoch

# Sensor values
temperature: float  # Celsius, range -50 to 100 (validated)
humidity: float     # Percent, range 0 to 100 (validated)

# Calculated properties (stored in database)
dewpoint: float           # Celsius, formula: T - ((100 - RH) / 5.0)
absolute_humidity: float  # g/m³, formula: (6.112 × e^((17.67×T)/(T+243.5)) × RH × 2.1674) / (273.15+T)

# Calculated properties (not stored, computed on-demand)
humidex: float            # Comfort index using Humidex formula
comfort_score: int        # 0-6 rating based on Humidex ranges

Comfort Score System

Based on Humidex (Environment Canada scale):

Score Humidex Range Description Icon
0 < 20 or ≥ 54 Unhealthy (too cold or heatstroke risk) 🥶 / ⚠️
1 54-46 Unhealthy (heat cramps possible) 🔥
2 46-40 Uncomfortable (avoid exertion) 😟
3 40-35 Slightly uncomfortable 😓
4 27-20 Comfortable (cool) 🙂
5 30-27 Comfortable 😊
6 35-30 Optimal comfort ✨

Humidex Formula:

Dewpoint (Td) = T - ((100 - RH) / 5.0)
e = 6.11 × exp(5417.7530 × (1/273.16 - 1/(273.15+Td)))
Humidex = T + 0.5555 × (e - 10)

Performance Optimization

Adaptive Downsampling

The application implements intelligent downsampling for time ranges > 1 hour or datasets > 5000 points:

Time Range Resample Interval Typical Point Count
< 1 hour No downsampling ~3,600 max
1-6 hours 1 minute ~360 points
6-24 hours 5 minutes ~288 points
1-7 days 15 minutes ~672 points
1 month 1 hour ~720 points
3 months 3 hours ~720 points
6 months 6 hours ~720 points
> 6 months 1 day ~365 points/year

"All Data" Mode (Dynamic):

  • 1 year: 1 day intervals

  • 3-12 months: 6 hour intervals
  • 1-3 months: 2 hour intervals
  • < 1 month: 1 hour intervals

Performance Impact:

  • 6 hours: ~18,720 → 360 points (98% reduction)
  • 1 year: ~27 million → 365 points (99.999% reduction)
  • Visual quality preserved - trends fully visible
  • Browser rendering instant regardless of data volume

Implementation:

df.set_index('timestamp').resample(interval).mean()

Uses pandas' optimized time-series resampling with mean aggregation.

Database Query Optimization

  • Union Query Builder: Only queries tables within requested time range
  • Indexed Timestamps: Fast filtering on timestamp column
  • WAL Mode: Better read/write concurrency
  • Table Partitioning: Daily tables for efficient cleanup and queries

Batch Operations

Modbus Reader:

  • Buffers 30 measurements before database write
  • Reduces write operations by 30x
  • Commit interval: 30 seconds

Database Connection:

  • Persistent connection in Modbus thread
  • Reused across measurements
  • Proper transaction management

Error Handling

All modules implement try-except blocks:

try:
    # Normal operation
except Exception as e:
    print(f"Error: {e}")
    # Fallback behavior
    # Continue operation if possible

Errors logged to terminal output. Application continues running on errors. Database operations use WAL mode for robustness. Modbus thread auto-reconnects on communication errors. Invalid sensor readings are validated and skipped.

Clone this wiki locally