Name | maxsmart JSON |
Version |
2.0.3
JSON |
| download |
home_page | None |
Summary | A comprehensive Python library for controlling Revogi-based Max Hauri MaxSmart PowerStrips and Smart Plugs |
upload_time | 2025-07-20 16:25:35 |
maintainer | None |
docs_url | None |
author | None |
requires_python | >=3.7 |
license | MIT |
keywords |
maxsmart
revogi
smart
plug
power
strip
home
automation
iot
|
VCS |
 |
bugtrack_url |
|
requirements |
No requirements were recorded.
|
Travis-CI |
No Travis.
|
coveralls test coverage |
No coveralls.
|
# MaxSmart Python Module
[](https://pypi.org/project/maxsmart/)
[](https://pypi.org/project/maxsmart/)
[](https://opensource.org/licenses/MIT)
**Version:** 2.0.3
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.
## 🎯 What's New in v2.0
- **Full async/await architecture** - Modern Python async throughout
- **Intelligent firmware detection** - Auto-detects and adapts to different firmware versions
- **Adaptive polling system** - Mimics official app behavior (5s normal, 2s burst)
- **Real-time monitoring** - Live consumption and state change detection
- **Enhanced error handling** - Robust retry logic with localized messages
- **Session management** - Connection pooling and automatic cleanup
- **Advanced statistics** - Hourly, daily, monthly data with visualization support
- **Device time access** - Real-time clock management
## 🔧 Supported Hardware
### Firmware Compatibility
- **v1.30**: Full feature support including port name management
- **v2.11+**: Basic control and monitoring (port renaming not supported)
- **Auto-detection**: Module automatically adapts to your firmware version
### Device Models
- Max Hauri MaxSmart Power Station (6 ports)
- Max Hauri MaxSmart Smart Plug (1 port)
- Revogi Smart Power Strip
- Extel Soky Power Strip
- MCL DOM-PPS06I
### 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
- **Integer watts** (5) → Direct watt values
## 📦 Installation
### From PyPI
```bash
pip install maxsmart
```
### From Source
```bash
git clone https://github.com/superkikim/maxsmart.git
cd maxsmart
pip install .
```
### Dependencies
- Python 3.7+
- aiohttp (async HTTP client)
- matplotlib (for example scripts)
## 🚀 Quick Start
```python
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"Data format: {device._watt_format}")
# 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())
```
## 🔍 Device Discovery
### Network Scanning
```python
from maxsmart import MaxSmartDiscovery
# Discover all devices on local network
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"Ports: {device['pname']}")
```
### Targeted Discovery
```python
# 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 Control
### Basic Port Operations
```python
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
```python
# 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
```python
# 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
```python
# 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
```python
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
```python
# 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
```python
# 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
```python
# 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
```python
# 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
```python
# 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
```python
# 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
```python
# Always clean up resources
try:
# ... device operations
pass
finally:
await device.close() # Stops polling and closes HTTP session
```
## 🛠️ Error Handling
### Exception Types
```python
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
- `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
```bash
# Install visualization dependencies
pip install -r example_scripts/requirements.txt
# Run comprehensive test suite
python example_scripts/maxsmart_tests_async.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
```python
# 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
```python
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
```python
# 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
```python
# 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**
```python
# 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")
```
**Firmware compatibility errors**
```python
# Check device firmware version
await device.initialize_device()
print(f"Firmware: {device.version}")
print(f"Data format: {device._watt_format}")
# Firmware capabilities:
# v1.30: Full support including port renaming
# v2.11+: Basic control only (no port renaming)
```
**Connection timeouts**
```python
# 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
```python
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
```python
# 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](https://github.com/altery/mh-maxsmart-powerstation) who documented the MaxSmart communication protocol.
## 📄 License
MIT License - see [LICENSE](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
Raw data
{
"_id": null,
"home_page": null,
"name": "maxsmart",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": "Superkikim <your-email@example.com>",
"keywords": "maxsmart, revogi, smart, plug, power, strip, home, automation, iot",
"author": null,
"author_email": "Superkikim <your-email@example.com>",
"download_url": "https://files.pythonhosted.org/packages/40/f1/ffb5306549472968a13898757aa2dbe856f94e17367c4e0f35ed1e4dacf5/maxsmart-2.0.3.tar.gz",
"platform": null,
"description": "# MaxSmart Python Module\n\n[](https://pypi.org/project/maxsmart/)\n[](https://pypi.org/project/maxsmart/)\n[](https://opensource.org/licenses/MIT)\n\n**Version:** 2.0.3\n\nA 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.\n\n## \ud83c\udfaf What's New in v2.0\n\n- **Full async/await architecture** - Modern Python async throughout\n- **Intelligent firmware detection** - Auto-detects and adapts to different firmware versions\n- **Adaptive polling system** - Mimics official app behavior (5s normal, 2s burst)\n- **Real-time monitoring** - Live consumption and state change detection\n- **Enhanced error handling** - Robust retry logic with localized messages\n- **Session management** - Connection pooling and automatic cleanup\n- **Advanced statistics** - Hourly, daily, monthly data with visualization support\n- **Device time access** - Real-time clock management\n\n## \ud83d\udd27 Supported Hardware\n\n### Firmware Compatibility\n- **v1.30**: Full feature support including port name management\n- **v2.11+**: Basic control and monitoring (port renaming not supported)\n- **Auto-detection**: Module automatically adapts to your firmware version\n\n### Device Models\n- Max Hauri MaxSmart Power Station (6 ports)\n- Max Hauri MaxSmart Smart Plug (1 port) \n- Revogi Smart Power Strip\n- Extel Soky Power Strip\n- MCL DOM-PPS06I\n\n### Data Format Detection\nThe module automatically detects your device's data format:\n- **String floats** (\"5.20\") \u2192 Direct watt values\n- **Integer milliwatts** (5200) \u2192 Converted to watts\n- **Integer watts** (5) \u2192 Direct watt values\n\n## \ud83d\udce6 Installation\n\n### From PyPI\n```bash\npip install maxsmart\n```\n\n### From Source\n```bash\ngit clone https://github.com/superkikim/maxsmart.git\ncd maxsmart\npip install .\n```\n\n### Dependencies\n- Python 3.7+\n- aiohttp (async HTTP client)\n- matplotlib (for example scripts)\n\n## \ud83d\ude80 Quick Start\n\n```python\nimport asyncio\nfrom maxsmart import MaxSmartDiscovery, MaxSmartDevice\n\nasync def main():\n # Discover devices on network\n devices = await MaxSmartDiscovery.discover_maxsmart()\n if not devices:\n print(\"No devices found\")\n return\n \n # Connect to first device\n device = MaxSmartDevice(devices[0]['ip'])\n await device.initialize_device()\n \n print(f\"Connected to {device.name} (FW: {device.version})\")\n print(f\"Data format: {device._watt_format}\")\n \n # Control ports\n await device.turn_on(1) # Turn on port 1\n state = await device.check_state(1) # Check port 1 state\n power = await device.get_power_data(1) # Get power consumption\n \n print(f\"Port 1: {'ON' if state else 'OFF'}, {power['watt']}W\")\n \n # Cleanup\n await device.close()\n\nasyncio.run(main())\n```\n\n## \ud83d\udd0d Device Discovery\n\n### Network Scanning\n```python\nfrom maxsmart import MaxSmartDiscovery\n\n# Discover all devices on local network\ndiscovery = MaxSmartDiscovery()\ndevices = await discovery.discover_maxsmart()\n\nfor device in devices:\n print(f\"Name: {device['name']}\")\n print(f\"IP: {device['ip']}\")\n print(f\"Serial: {device['sn']}\")\n print(f\"Firmware: {device['ver']}\")\n print(f\"Ports: {device['pname']}\")\n```\n\n### Targeted Discovery\n```python\n# Query specific IP address\ndevices = await MaxSmartDiscovery.discover_maxsmart(ip=\"192.168.1.100\")\n\n# With custom timeout and locale\ndevices = await MaxSmartDiscovery.discover_maxsmart(\n ip=\"192.168.1.100\",\n user_locale=\"fr\",\n timeout=5.0\n)\n```\n\n## \ud83c\udf9b\ufe0f Device Control\n\n### Basic Port Operations\n```python\ndevice = MaxSmartDevice('192.168.1.100')\nawait device.initialize_device()\n\n# Individual ports (1-6)\nawait device.turn_on(1) # Turn on port 1\nawait device.turn_off(3) # Turn off port 3\n\n# Master control (port 0 = all ports)\nawait device.turn_on(0) # Turn on all ports\nawait device.turn_off(0) # Turn off all ports\n\n# State checking\nport1_state = await device.check_state(1) # Single port: 0 or 1\nall_states = await device.check_state() # All ports: [0,1,0,1,1,0]\n```\n\n### Device Information\n```python\n# Device properties (available after initialization)\nprint(f\"Device name: {device.name}\")\nprint(f\"IP address: {device.ip}\")\nprint(f\"Serial number: {device.sn}\")\nprint(f\"Firmware version: {device.version}\")\nprint(f\"Strip name: {device.strip_name}\")\nprint(f\"Port names: {device.port_names}\")\n```\n\n### Port Name Management\n```python\n# Retrieve current port names\nport_mapping = await device.retrieve_port_names()\nprint(port_mapping)\n# Output: {'Port 0': 'Living Room Strip', 'Port 1': 'TV', 'Port 2': 'Lamp', ...}\n\n# Change port names (firmware v1.30 only)\ntry:\n await device.change_port_name(1, \"Smart TV\")\n await device.change_port_name(0, \"Entertainment Center\") # Strip name\n print(\"Port names updated successfully\")\nexcept FirmwareError as e:\n print(f\"Port renaming not supported: {e}\")\n```\n\n## \u26a1 Advanced Polling & Monitoring\n\n### Adaptive Polling System\n```python\n# Start intelligent polling (mimics official app)\nawait device.start_adaptive_polling()\n\n# Polling behavior:\n# - Normal: polls every 5 seconds\n# - Burst: polls every 2 seconds for 4 cycles after commands\n# - Auto-triggers burst mode on turn_on/turn_off operations\n\n# Register callback for poll events\ndef poll_handler(poll_data):\n mode = poll_data['mode'] # 'normal' or 'burst'\n count = poll_data['poll_count'] # Poll sequence number\n data = poll_data['device_data'] # Switch states and watt values\n print(f\"Poll #{count} ({mode}): {data}\")\n\ndevice.register_poll_callback(\"monitor\", poll_handler)\n\n# Commands automatically trigger burst mode\nawait device.turn_on(1) # Triggers 2-second polling for 8 seconds\n\n# Stop polling\nawait device.stop_adaptive_polling()\n```\n\n### Real-time Change Detection\n```python\nasync def on_consumption_change(data):\n \"\"\"Called when power consumption changes >1W\"\"\"\n port = data['port']\n port_name = data['port_name']\n change = data['change']\n current = data['current_watt']\n print(f\"{port_name} (Port {port}): {current:.1f}W ({change:+.1f}W)\")\n\nasync def on_state_change(data):\n \"\"\"Called when port switches on/off\"\"\"\n port = data['port']\n port_name = data['port_name'] \n state = data['state_text'] # 'ON' or 'OFF'\n print(f\"{port_name} (Port {port}): {state}\")\n\n# Setup monitoring with automatic change detection\nawait device.setup_realtime_monitoring(\n consumption_callback=on_consumption_change,\n state_callback=on_state_change\n)\n\n# Start polling to enable monitoring\nawait device.start_adaptive_polling()\n```\n\n## \ud83d\udcca Power Monitoring\n\n### Real-time Data\n```python\n# Get current power consumption for specific port\npower_data = await device.get_power_data(1)\nprint(f\"Port 1: {power_data['watt']}W\")\n\n# Get comprehensive device data\nall_data = await device.get_data()\nprint(f\"Switch states: {all_data['switch']}\") # [0,1,0,1,1,0]\nprint(f\"Power values: {all_data['watt']}\") # [0.0,15.2,0.0,8.7,122.5,0.0]\n```\n\n### Historical Statistics\n```python\n# Statistics types:\n# 0 = Hourly (last 24 hours)\n# 1 = Daily (last 30 days) \n# 2 = Monthly (last 12 months)\n\n# Get hourly data for all ports combined\nhourly_stats = await device.get_statistics(0, 0) # port 0, type 0\nprint(f\"Last 24 hours: {hourly_stats['watt']}\")\nprint(f\"Period ending: {hourly_stats['year']}-{hourly_stats['month']}-{hourly_stats['day']} {hourly_stats['hour']}:00\")\n\n# Get daily data for specific port\ndaily_stats = await device.get_statistics(1, 1) # port 1, type 1\nprint(f\"Port 1 daily consumption: {daily_stats['watt']}\")\n\n# Statistics include cost information if configured\nif daily_stats['cost'] > 0:\n print(f\"Estimated cost: {daily_stats['cost']} {daily_stats['currency']}/kWh\")\n```\n\n### Data Visualization\n```python\n# The example scripts include comprehensive visualization\n# See: example_scripts/maxsmart_tests_async.py\n\nimport matplotlib.pyplot as plt\n\n# Get monthly data\nmonthly_data = await device.get_statistics(0, 2)\nwatt_values = monthly_data['watt']\n\n# Plot consumption over time\nplt.figure(figsize=(12, 6))\nplt.plot(watt_values)\nplt.title('Monthly Power Consumption')\nplt.ylabel('Power (W)')\nplt.xlabel('Month')\nplt.show()\n```\n\n## \ud83d\udd50 Device Time Management\n```python\n# Get device's real-time clock\ntime_data = await device.get_device_time()\nprint(f\"Device time: {time_data['time']}\")\n# Output: \"2024-07-17,14:30:25\"\n```\n\n## \ud83d\udd27 Session Management & Health Checks\n\n### Context Manager Usage\n```python\n# Automatic initialization and cleanup\nasync with MaxSmartDevice('192.168.1.100') as device:\n await device.turn_on(1)\n power = await device.get_power_data(1)\n print(f\"Power: {power['watt']}W\")\n# Device automatically closed on exit\n```\n\n### Health Monitoring\n```python\n# Check device connectivity and performance\nhealth = await device.health_check()\nprint(f\"Status: {health['status']}\")\nprint(f\"Response time: {health['response_time']}ms\")\nprint(f\"Polling active: {health['polling_active']}\")\n\nif health['status'] == 'unhealthy':\n print(f\"Error: {health['error']}\")\n```\n\n### Manual Cleanup\n```python\n# Always clean up resources\ntry:\n # ... device operations\n pass\nfinally:\n await device.close() # Stops polling and closes HTTP session\n```\n\n## \ud83d\udee0\ufe0f Error Handling\n\n### Exception Types\n```python\nfrom maxsmart.exceptions import (\n DiscoveryError, # Device discovery failures\n ConnectionError, # Network connectivity issues\n CommandError, # Command execution failures\n StateError, # Device state inconsistencies\n FirmwareError, # Firmware compatibility issues\n DeviceTimeoutError # Device response timeouts\n)\n\ntry:\n devices = await MaxSmartDiscovery.discover_maxsmart()\n device = MaxSmartDevice(devices[0]['ip'])\n await device.initialize_device()\n await device.turn_on(1)\n \nexcept DiscoveryError as e:\n print(f\"Discovery failed: {e}\")\nexcept ConnectionError as e:\n print(f\"Network error: {e}\")\nexcept FirmwareError as e:\n print(f\"Firmware limitation: {e}\")\nexcept Exception as e:\n print(f\"Unexpected error: {e}\")\n```\n\n### Retry Logic\nThe module includes built-in retry mechanisms:\n- **Discovery**: 3 attempts with exponential backoff\n- **Commands**: 3 retries with 1-second delays\n- **State verification**: 3 verification attempts\n- **Session management**: Automatic reconnection on failures\n\n## \ud83d\udcc1 Example Scripts\n\nThe `example_scripts/` directory contains comprehensive examples:\n\n### Basic Examples\n- `test_discovery_async.py` - Device discovery with validation\n- `get_port_names_async.py` - Port name retrieval and display\n- `retrieve_states_async.py` - Port state checking\n- `show_consumption.py` - Real-time power monitoring\n\n### Advanced Examples \n- `maxsmart_tests_async.py` - Complete feature testing with visualization\n- `test_adaptive_polling.py` - Polling system demonstration\n- `rename_ports.py` - Port name management (v1.30 firmware)\n- `test_device_timestamps.py` - Device time synchronization\n\n### Running Examples\n```bash\n# Install visualization dependencies\npip install -r example_scripts/requirements.txt\n\n# Run comprehensive test suite\npython example_scripts/maxsmart_tests_async.py\n\n# Test adaptive polling\npython example_scripts/test_adaptive_polling.py\n\n# Monitor real-time consumption\npython example_scripts/show_consumption.py\n```\n\n## \u2699\ufe0f Configuration Options\n\n### Discovery Configuration\n```python\n# Custom timeout and retry settings\ndevices = await MaxSmartDiscovery.discover_maxsmart(\n ip=None, # None for broadcast, IP for unicast\n user_locale=\"en\", # Locale for error messages\n timeout=3.0, # Discovery timeout in seconds\n max_attempts=3 # Maximum discovery attempts\n)\n```\n\n### Device Configuration\n```python\ndevice = MaxSmartDevice(\n ip_address=\"192.168.1.100\",\n auto_polling=False # Start polling automatically after init\n)\n\n# Custom session timeouts\ndevice.DEFAULT_TIMEOUT = 15.0 # Command timeout\ndevice.RETRY_COUNT = 5 # Command retry attempts\n```\n\n### Polling Configuration\n```python\n# Customize polling intervals\ndevice.NORMAL_INTERVAL = 10.0 # Normal polling interval (default: 5s)\ndevice.BURST_INTERVAL = 1.0 # Burst polling interval (default: 2s) \ndevice.BURST_CYCLES = 6 # Burst cycle count (default: 4)\n```\n\n## \ud83d\udd12 Security Considerations\n\n### Network Security\n- **Unencrypted HTTP**: All communication is in plain text\n- **No authentication**: Devices don't require credentials\n- **Local network only**: Devices must be on same network as client\n- **Trusted networks**: Only use on secure, trusted networks\n\n### Best Practices\n```python\n# Use connection limits to prevent resource exhaustion\ndevice.DEFAULT_CONNECTOR_LIMIT = 10\ndevice.DEFAULT_CONNECTOR_LIMIT_PER_HOST = 5\n\n# Always clean up resources\nasync with MaxSmartDevice(ip) as device:\n # ... operations\n pass # Automatic cleanup\n\n# Handle exceptions gracefully\ntry:\n await device.turn_on(1)\nexcept Exception as e:\n logging.error(f\"Command failed: {e}\")\n```\n\n## \ud83d\udc1b Troubleshooting\n\n### Common Issues\n\n**Device not found during discovery**\n```python\n# Try targeted discovery\ndevices = await MaxSmartDiscovery.discover_maxsmart(ip=\"192.168.1.100\")\n\n# Check network connectivity\nimport socket\ntry:\n socket.create_connection((\"192.168.1.100\", 8888), timeout=3)\n print(\"Device reachable\")\nexcept:\n print(\"Device unreachable\")\n```\n\n**Firmware compatibility errors**\n```python\n# Check device firmware version\nawait device.initialize_device()\nprint(f\"Firmware: {device.version}\")\nprint(f\"Data format: {device._watt_format}\")\n\n# Firmware capabilities:\n# v1.30: Full support including port renaming\n# v2.11+: Basic control only (no port renaming)\n```\n\n**Connection timeouts**\n```python\n# Increase timeouts for slow networks\ndevice.DEFAULT_TIMEOUT = 30.0\ndevice.SESSION_TIMEOUT = 60.0\n\n# Reduce retry count to fail faster\ndevice.RETRY_COUNT = 1\n```\n\n### Debug Logging\n```python\nimport logging\nlogging.basicConfig(level=logging.DEBUG)\n\n# Enable detailed HTTP logging\nlogging.getLogger('aiohttp').setLevel(logging.DEBUG)\n```\n\n## \ud83d\udcc8 Performance Notes\n\n### Efficient Usage\n- **Reuse device instances**: Don't recreate for each operation\n- **Use adaptive polling**: More efficient than manual polling\n- **Batch operations**: Minimize individual command calls\n- **Context managers**: Ensure proper cleanup\n\n### Resource Management\n```python\n# Good: Reuse device instance\ndevice = MaxSmartDevice(ip)\nawait device.initialize_device()\nfor port in range(1, 7):\n await device.turn_on(port)\nawait device.close()\n\n# Bad: Create new instances\nfor port in range(1, 7):\n device = MaxSmartDevice(ip) # Inefficient\n await device.initialize_device()\n await device.turn_on(port)\n await device.close()\n```\n\n## \ud83e\udd1d Credits\n\nThis module builds upon reverse engineering work by [@altery](https://github.com/altery/mh-maxsmart-powerstation) who documented the MaxSmart communication protocol.\n\n## \ud83d\udcc4 License\n\nMIT License - see [LICENSE](LICENSE) file for details.\n\n## \ud83d\udd17 Links\n\n- **GitHub Repository**: https://github.com/superkikim/maxsmart\n- **PyPI Package**: https://pypi.org/project/maxsmart/\n- **Issues & Support**: https://github.com/superkikim/maxsmart/issues\n- **Example Scripts**: https://github.com/superkikim/maxsmart/tree/main/example_scripts\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A comprehensive Python library for controlling Revogi-based Max Hauri MaxSmart PowerStrips and Smart Plugs",
"version": "2.0.3",
"project_urls": {
"Bug Tracker": "https://github.com/superkikim/maxsmart/issues",
"Changelog": "https://github.com/superkikim/maxsmart/blob/main/CHANGELOG.md",
"Documentation": "https://github.com/superkikim/maxsmart/blob/main/README.md",
"Homepage": "https://github.com/superkikim/maxsmart",
"Repository": "https://github.com/superkikim/maxsmart.git"
},
"split_keywords": [
"maxsmart",
" revogi",
" smart",
" plug",
" power",
" strip",
" home",
" automation",
" iot"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "1579489e28319c9f76af9154a980a12411079322dd89e9686f1d65daa085b1d6",
"md5": "cbf68c6d1429230c8b51c655ff235383",
"sha256": "d4f949fa1ee59cdef3c75b75f8fcb6309145458545fe1d555dc8028a6d671198"
},
"downloads": -1,
"filename": "maxsmart-2.0.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "cbf68c6d1429230c8b51c655ff235383",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 38271,
"upload_time": "2025-07-20T16:25:34",
"upload_time_iso_8601": "2025-07-20T16:25:34.181563Z",
"url": "https://files.pythonhosted.org/packages/15/79/489e28319c9f76af9154a980a12411079322dd89e9686f1d65daa085b1d6/maxsmart-2.0.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "40f1ffb5306549472968a13898757aa2dbe856f94e17367c4e0f35ed1e4dacf5",
"md5": "8febccbcc818ad9f64936502a1beb8b9",
"sha256": "2d03edb81396c2f73006198792366796cf56bbf80fff835860bf706a3902fffa"
},
"downloads": -1,
"filename": "maxsmart-2.0.3.tar.gz",
"has_sig": false,
"md5_digest": "8febccbcc818ad9f64936502a1beb8b9",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 38267,
"upload_time": "2025-07-20T16:25:35",
"upload_time_iso_8601": "2025-07-20T16:25:35.912972Z",
"url": "https://files.pythonhosted.org/packages/40/f1/ffb5306549472968a13898757aa2dbe856f94e17367c4e0f35ed1e4dacf5/maxsmart-2.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-20 16:25:35",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "superkikim",
"github_project": "maxsmart",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "maxsmart"
}