MaxSmart Python Module

Version: 2.1.0

A comprehensive Python library for controlling Revogi-based Max Hauri MaxSmart PowerStrips and Smart Plugs over local network. Features intelligent auto-detection, adaptive polling, real-time monitoring, and robust async architecture with simplified device identification.

☕ Support My Work

If this plugin helps you, consider supporting its development:

🎯 What's New in v2.1.0

🔄 Protocol Transparency - Unified API works seamlessly with both HTTP and UDP V3 devices
🤖 Automatic Protocol Detection - No need to specify protocol, automatically detects HTTP or UDP V3
📡 UDP V3 Support - Full support for UDP-only devices (commands 20 and 90). Tested on Revogi FW 5.11
🛡️ Robust Error Handling - Same retry logic and timeout handling for both protocols
🎯 Simplified Usage - Same methods (turn_on, turn_off, get_data) work regardless of device protocol
⚡ Enhanced Performance - Optimized command routing and reduced overhead

🔧 Supported Hardware

Firmware Compatibility & Testing Status

Brand	Device Type	Firmware Versions
Max Hauri	Power Station (6 ports)	v1.10, v1.30, v2.11
Max Hauri	Smart Plug (1 port)	v1.10, v1.30, v2.11
Revogi	Power Strip (6 ports)	v3.36, v3.49, v5.11
CoCoSo	Power Strip (6 ports)	v1.06
Extel	Soky Power Strip	Various
MCL	DOM-PPS06I	Various

Device Models

Max Hauri MaxSmart Power Station (6 ports)
Max Hauri MaxSmart Smart Plug (1 port)
Revogi Smart Power Strip (SOW323)
CoCoSo Smart Power Strip (SOW323)
Extel Soky Power Strip
MCL DOM-PPS06I

🤝 Help expand compatibility! If you have Revogi-based devices with different firmware versions, please test them and let us know - we'd love to add them to the supported list!

Data Format Detection

The module automatically detects your device's data format:

String floats ("5.20") → Direct watt values
Integer milliwatts (5200) → Converted to watts (×0.001)
Integer watts (5) → Direct watt values

📦 Installation

From PyPI

pip install maxsmart

From Source

git clone https://github.com/superkikim/maxsmart.git
cd maxsmart
pip install .

Dependencies

Python 3.7+
aiohttp (async HTTP client)
getmac (MAC address discovery)
matplotlib (for example scripts)

🚀 Quick Start

import asyncio
from maxsmart import MaxSmartDiscovery, MaxSmartDevice

async def main():
    # Discover devices on network
    devices = await MaxSmartDiscovery.discover_maxsmart()
    if not devices:
        print("No devices found")
        return
        
    # Connect to first device
    device = MaxSmartDevice(devices[0]['ip'])
    await device.initialize_device()
    
    print(f"Connected to {device.name} (FW: {device.version})")
    print(f"Serial: {devices[0]['sn']}")
    print(f"CPU ID: {devices[0]['cpuid']}")
    
    # Control ports
    await device.turn_on(1)                    # Turn on port 1
    state = await device.check_state(1)       # Check port 1 state
    power = await device.get_power_data(1)    # Get power consumption
    
    print(f"Port 1: {'ON' if state else 'OFF'}, {power['watt']}W")
    
    # Cleanup
    await device.close()

asyncio.run(main())

🔄 Protocol Transparency

Unified API for All Devices

The module automatically detects and supports both HTTP and UDP V3 protocols transparently:

# Same code works for both HTTP and UDP V3 devices!
device = MaxSmartDevice('192.168.1.100')  # Auto-detects protocol
await device.initialize_device()

# These methods work regardless of protocol:
await device.turn_on(1)                    # ✅ HTTP or UDP V3
await device.turn_off(3)                   # ✅ HTTP or UDP V3
state = await device.check_state(1)        # ✅ HTTP or UDP V3
power = await device.get_power_data(1)     # ✅ HTTP or UDP V3
data = await device.get_data()             # ✅ HTTP or UDP V3

print(f"Protocol: {device.protocol}")      # Shows 'http' or 'udp_v3'

Firmware Support Matrix

Firmware Version	HTTP Support	UDP V3 Support	Auto-Detection Result	Recommended
1.30	✅ Full	❌ None	`HTTP`	HTTP (only option)
2.11	✅ Full	⚠️ Partial*	`HTTP`	HTTP (more features)
5.xx+	❌ None	✅ Full	`UDP V3`	UDP V3 (only option)

*Firmware 2.11 supports UDP V3 commands (20, 90) but with limited data response

Protocol Support Matrix

Feature	HTTP Devices	UDP V3 Devices	Notes
Port Control	✅	✅	`turn_on()`, `turn_off()`
State Checking	✅	✅	`check_state()`, `get_data()`
Power Monitoring	✅	✅	`get_power_data()`
Statistics	✅	❌	`get_statistics()` HTTP only
Port Naming	✅	❌	`change_port_name()` HTTP only
Device Time	✅	❌	`get_device_time()` HTTP only

Manual Protocol Selection

# Force specific protocol if needed
device_http = MaxSmartDevice('192.168.1.100', protocol='http')
device_udp = MaxSmartDevice('192.168.1.101', protocol='udp_v3')

# Auto-detection (recommended)
device_auto = MaxSmartDevice('192.168.1.102')  # protocol=None

🔍 Device Discovery

Network Scanning

from maxsmart import MaxSmartDiscovery

# Discover all devices on local network (simplified format)
discovery = MaxSmartDiscovery()
devices = await discovery.discover_maxsmart()

for device in devices:
    print(f"Name: {device['name']}")
    print(f"IP: {device['ip']}")
    print(f"Serial: {device['sn']}")
    print(f"Firmware: {device['ver']}")
    print(f"MAC: {device.get('mac', 'N/A')}")
    print(f"Protocol: {device.get('protocol', 'unknown')}")
    print(f"Ports: {device.get('nr_of_ports', 6)}")

Essential Discovery Format

The module now returns a clean, essential format:

{
    "sn": "SWP6023002003697",           # Serial number (primary identifier)
    "name": "Living Room Strip",        # Device name
    "pname": ["TV", "Lamp", ...],       # Port names (empty for newer firmware)
    "ip": "192.168.1.100",              # IP address
    "ver": "1.30",                      # Firmware version
    "mac": "AA:BB:CC:DD:EE:FF",         # MAC address via ARP
    "server": "www.maxsmart.ch"         # Cloud server endpoint
}

Targeted Discovery

# Query specific IP address
devices = await MaxSmartDiscovery.discover_maxsmart(ip="192.168.1.100")

# With custom timeout and locale
devices = await MaxSmartDiscovery.discover_maxsmart(
    ip="192.168.1.100",
    user_locale="fr",
    timeout=5.0
)

🆔 Device Identification

Network Identifiers

device = MaxSmartDevice('192.168.1.100')
await device.initialize_device()

# Get MAC address via ARP (best effort)
mac_address = await device.get_mac_address_via_arp()
print(f"ARP MAC: {mac_address}")

Identification Strategy

The module uses serial number as primary identifier with hardware identifiers for additional validation:

Serial Number (primary) - Device serial from UDP discovery
CPU ID (validation) - Hardware-based unique identifier
MAC Address (network) - Network hardware identifier via ARP

🎛️ Device Control

Basic Port Operations

device = MaxSmartDevice('192.168.1.100')
await device.initialize_device()

# Individual ports (1-6)
await device.turn_on(1)      # Turn on port 1
await device.turn_off(3)     # Turn off port 3

# Master control (port 0 = all ports)
await device.turn_on(0)      # Turn on all ports
await device.turn_off(0)     # Turn off all ports

# State checking
port1_state = await device.check_state(1)    # Single port: 0 or 1
all_states = await device.check_state()      # All ports: [0,1,0,1,1,0]

Device Information

# Device properties (available after initialization)
print(f"Device name: {device.name}")
print(f"IP address: {device.ip}")
print(f"Serial number: {device.sn}")
print(f"Firmware version: {device.version}")
print(f"Strip name: {device.strip_name}")
print(f"Port names: {device.port_names}")

Port Name Management

# Retrieve current port names
port_mapping = await device.retrieve_port_names()
print(port_mapping)
# Output: {'Port 0': 'Living Room Strip', 'Port 1': 'TV', 'Port 2': 'Lamp', ...}

# Change port names (firmware v1.30 only)
try:
    await device.change_port_name(1, "Smart TV")
    await device.change_port_name(0, "Entertainment Center")  # Strip name
    print("Port names updated successfully")
except FirmwareError as e:
    print(f"Port renaming not supported: {e}")

⚡ Advanced Polling & Monitoring

Adaptive Polling System

# Start intelligent polling (mimics official app)
await device.start_adaptive_polling()

# Polling behavior:
# - Normal: polls every 5 seconds
# - Burst: polls every 2 seconds for 4 cycles after commands
# - Auto-triggers burst mode on turn_on/turn_off operations

# Register callback for poll events
def poll_handler(poll_data):
    mode = poll_data['mode']        # 'normal' or 'burst'
    count = poll_data['poll_count'] # Poll sequence number
    data = poll_data['device_data'] # Switch states and watt values
    print(f"Poll #{count} ({mode}): {data}")

device.register_poll_callback("monitor", poll_handler)

# Commands automatically trigger burst mode
await device.turn_on(1)  # Triggers 2-second polling for 8 seconds

# Stop polling
await device.stop_adaptive_polling()

Real-time Change Detection

async def on_consumption_change(data):
    """Called when power consumption changes >1W"""
    port = data['port']
    port_name = data['port_name']
    change = data['change']
    current = data['current_watt']
    print(f"{port_name} (Port {port}): {current:.1f}W ({change:+.1f}W)")

async def on_state_change(data):
    """Called when port switches on/off"""
    port = data['port']
    port_name = data['port_name'] 
    state = data['state_text']  # 'ON' or 'OFF'
    print(f"{port_name} (Port {port}): {state}")

# Setup monitoring with automatic change detection
await device.setup_realtime_monitoring(
    consumption_callback=on_consumption_change,
    state_callback=on_state_change
)

# Start polling to enable monitoring
await device.start_adaptive_polling()

📊 Power Monitoring

Real-time Data

# Get current power consumption for specific port
power_data = await device.get_power_data(1)
print(f"Port 1: {power_data['watt']}W")

# Get comprehensive device data
all_data = await device.get_data()
print(f"Switch states: {all_data['switch']}")  # [0,1,0,1,1,0]
print(f"Power values: {all_data['watt']}")     # [0.0,15.2,0.0,8.7,122.5,0.0]

Historical Statistics

# Statistics types:
# 0 = Hourly (last 24 hours)
# 1 = Daily (last 30 days)  
# 2 = Monthly (last 12 months)

# Get hourly data for all ports combined
hourly_stats = await device.get_statistics(0, 0)  # port 0, type 0
print(f"Last 24 hours: {hourly_stats['watt']}")
print(f"Period ending: {hourly_stats['year']}-{hourly_stats['month']}-{hourly_stats['day']} {hourly_stats['hour']}:00")

# Get daily data for specific port
daily_stats = await device.get_statistics(1, 1)   # port 1, type 1
print(f"Port 1 daily consumption: {daily_stats['watt']}")

# Statistics include cost information if configured
if daily_stats['cost'] > 0:
    print(f"Estimated cost: {daily_stats['cost']} {daily_stats['currency']}/kWh")

Data Visualization

# The example scripts include comprehensive visualization
# See: example_scripts/maxsmart_tests_async.py

import matplotlib.pyplot as plt

# Get monthly data
monthly_data = await device.get_statistics(0, 2)
watt_values = monthly_data['watt']

# Plot consumption over time
plt.figure(figsize=(12, 6))
plt.plot(watt_values)
plt.title('Monthly Power Consumption')
plt.ylabel('Power (W)')
plt.xlabel('Month')
plt.show()

🕐 Device Time Management

# Get device's real-time clock
time_data = await device.get_device_time()
print(f"Device time: {time_data['time']}")
# Output: "2024-07-17,14:30:25"

🔧 Session Management & Health Checks

Context Manager Usage

# Automatic initialization and cleanup
async with MaxSmartDevice('192.168.1.100') as device:
    await device.turn_on(1)
    power = await device.get_power_data(1)
    print(f"Power: {power['watt']}W")
# Device automatically closed on exit

Health Monitoring

# Check device connectivity and performance
health = await device.health_check()
print(f"Status: {health['status']}")
print(f"Response time: {health['response_time']}ms")
print(f"Polling active: {health['polling_active']}")

if health['status'] == 'unhealthy':
    print(f"Error: {health['error']}")

Manual Cleanup

# Always clean up resources
try:
    # ... device operations
    pass
finally:
    await device.close()  # Stops polling and closes HTTP session

🛠️ Error Handling

Exception Types

from maxsmart.exceptions import (
    DiscoveryError,        # Device discovery failures
    ConnectionError,       # Network connectivity issues
    CommandError,          # Command execution failures
    StateError,           # Device state inconsistencies
    FirmwareError,        # Firmware compatibility issues
    DeviceTimeoutError    # Device response timeouts
)

try:
    devices = await MaxSmartDiscovery.discover_maxsmart()
    device = MaxSmartDevice(devices[0]['ip'])
    await device.initialize_device()
    await device.turn_on(1)
    
except DiscoveryError as e:
    print(f"Discovery failed: {e}")
except ConnectionError as e:
    print(f"Network error: {e}")
except FirmwareError as e:
    print(f"Firmware limitation: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Retry Logic

The module includes built-in retry mechanisms:

Discovery: 3 attempts with exponential backoff
Commands: 3 retries with 1-second delays
State verification: 3 verification attempts
Session management: Automatic reconnection on failures

📁 Example Scripts

The example_scripts/ directory contains comprehensive examples:

Basic Examples

test_discovery_async.py - Device discovery with validation
test_device_identification.py - Hardware identification testing
get_port_names_async.py - Port name retrieval and display
retrieve_states_async.py - Port state checking
show_consumption.py - Real-time power monitoring

Advanced Examples

maxsmart_tests_async.py - Complete feature testing with visualization
test_adaptive_polling.py - Polling system demonstration
rename_ports.py - Port name management (v1.30 firmware)
test_device_timestamps.py - Device time synchronization

Running Examples

# Install visualization dependencies
pip install -r example_scripts/requirements.txt

# Run comprehensive test suite
python example_scripts/maxsmart_tests_async.py

# Test device identification
python example_scripts/test_device_identification.py

# Test adaptive polling
python example_scripts/test_adaptive_polling.py

# Monitor real-time consumption
python example_scripts/show_consumption.py

⚙️ Configuration Options

Discovery Configuration

# Custom timeout and retry settings
devices = await MaxSmartDiscovery.discover_maxsmart(
    ip=None,                    # None for broadcast, IP for unicast
    user_locale="en",           # Locale for error messages
    timeout=3.0,                # Discovery timeout in seconds
    max_attempts=3             # Maximum discovery attempts
)

Device Configuration

device = MaxSmartDevice(
    ip_address="192.168.1.100",
    auto_polling=False          # Start polling automatically after init
)

# Custom session timeouts
device.DEFAULT_TIMEOUT = 15.0  # Command timeout
device.RETRY_COUNT = 5         # Command retry attempts

Polling Configuration

# Customize polling intervals
device.NORMAL_INTERVAL = 10.0  # Normal polling interval (default: 5s)
device.BURST_INTERVAL = 1.0    # Burst polling interval (default: 2s)  
device.BURST_CYCLES = 6        # Burst cycle count (default: 4)

🔒 Security Considerations

Network Security

Unencrypted HTTP: All communication is in plain text
No authentication: Devices don't require credentials
Local network only: Devices must be on same network as client
Trusted networks: Only use on secure, trusted networks

Best Practices

# Use connection limits to prevent resource exhaustion
device.DEFAULT_CONNECTOR_LIMIT = 10
device.DEFAULT_CONNECTOR_LIMIT_PER_HOST = 5

# Always clean up resources
async with MaxSmartDevice(ip) as device:
    # ... operations
    pass  # Automatic cleanup

# Handle exceptions gracefully
try:
    await device.turn_on(1)
except Exception as e:
    logging.error(f"Command failed: {e}")

🐛 Troubleshooting

Common Issues

Device not found during discovery

# Try targeted discovery
devices = await MaxSmartDiscovery.discover_maxsmart(ip="192.168.1.100")

# Check network connectivity
import socket
try:
    socket.create_connection(("192.168.1.100", 8888), timeout=3)
    print("Device reachable")
except:
    print("Device unreachable")

Power scaling issues

# Check device firmware and data format
await device.initialize_device()
print(f"Firmware: {device.version}")
print(f"Data format: {device._watt_format}")
print(f"Conversion factor: {device._watt_multiplier}")

# Test with raw data
raw_data = await device.get_data()
print(f"Raw power values: {raw_data['watt']}")

Firmware compatibility errors

# Check device firmware version
await device.initialize_device()
print(f"Firmware: {device.version}")
print(f"Data format: {device._watt_format}")

# Firmware capabilities:
# v1.06, v1.10, v1.30: Full support including port renaming
# v2.11, v3.36, v3.49: Full control, basic port renaming

Connection timeouts

# Increase timeouts for slow networks
device.DEFAULT_TIMEOUT = 30.0
device.SESSION_TIMEOUT = 60.0

# Reduce retry count to fail faster
device.RETRY_COUNT = 1

Debug Logging

import logging
logging.basicConfig(level=logging.DEBUG)

# Enable detailed HTTP logging
logging.getLogger('aiohttp').setLevel(logging.DEBUG)

📈 Performance Notes

Efficient Usage

Reuse device instances: Don't recreate for each operation
Use adaptive polling: More efficient than manual polling
Batch operations: Minimize individual command calls
Context managers: Ensure proper cleanup

Resource Management

# Good: Reuse device instance
device = MaxSmartDevice(ip)
await device.initialize_device()
for port in range(1, 7):
    await device.turn_on(port)
await device.close()

# Bad: Create new instances
for port in range(1, 7):
    device = MaxSmartDevice(ip)  # Inefficient
    await device.initialize_device()
    await device.turn_on(port)
    await device.close()

🤝 Credits

This module builds upon reverse engineering work by @altery who documented the MaxSmart communication protocol.

📄 License

MIT License - see LICENSE file for details.

🔗 Links

GitHub Repository: https://github.com/superkikim/maxsmart
PyPI Package: https://pypi.org/project/maxsmart/
Issues & Support: https://github.com/superkikim/maxsmart/issues
Example Scripts: https://github.com/superkikim/maxsmart/tree/main/example_scripts

Name		Name	Last commit message	Last commit date
Latest commit History 132 Commits
.github/workflows		.github/workflows
example_scripts		example_scripts
maxsmart		maxsmart
.gitignore		.gitignore
LICENSE		LICENSE
README.md		README.md
RELEASE_NOTES.md		RELEASE_NOTES.md
pyproject.toml		pyproject.toml

License

Superkikim/maxsmart

Folders and files

Latest commit

History

Repository files navigation