explore_device_data.py

#!/usr/bin/env python3
"""
Explore INKBIRD device data points and capabilities
Step 1: Understanding what the device can do
"""

import tinytuya
import json
import time
from datetime import datetime

# Working configuration from our tests
INKBIRD_IP = "192.168.1.170"
DEVICE_ID = "ebe68bf20a45e83945l5q6"
LOCAL_KEY = "MiPEx}<M:Zj@8<X!"
PROTOCOL_VERSION = 3.4

def create_device():
    """Create and return configured device instance"""
    device = tinytuya.Device(
        dev_id=DEVICE_ID,
        address=INKBIRD_IP,
        local_key=LOCAL_KEY,
        version=PROTOCOL_VERSION
    )
    device.set_socketTimeout(10)
    return device

def explore_basic_status():
    """Get basic device status - what we already know works"""
    print("🔍 Getting Basic Device Status")
    print("-" * 40)
    
    device = create_device()
    
    try:
        status = device.status()
        print(f"✅ Device Status: {json.dumps(status, indent=2)}")
        
        if 'dps' in status:
            print(f"\n📊 Available Data Points:")
            for dps_id, value in status['dps'].items():
                print(f"   DPS {dps_id}: {value} (type: {type(value).__name__})")
                
        return status
        
    except Exception as e:
        print(f"❌ Error getting status: {e}")
        return None

def discover_all_dps():
    """Try to discover all available data points"""
    print("\n🔍 Discovering All Available Data Points")
    print("-" * 40)
    
    device = create_device()
    
    try:
        # Use tinytuya's built-in DPS discovery
        dps_data = device.detect_available_dps()
        print(f"✅ DPS Discovery Result: {json.dumps(dps_data, indent=2)}")
        
        return dps_data
        
    except Exception as e:
        print(f"❌ Error discovering DPS: {e}")
        return None

def analyze_dps_meanings():
    """Analyze what each DPS might mean based on the schema from logs"""
    print("\n🧠 Analyzing DPS Meanings (from APK analysis)")
    print("-" * 40)
    
    # From the logs, we found schema information
    known_schema = {
        "101": {
            "name": "real_time_data", 
            "description": "当前探头温度和风扇输出 (Current probe temperature and fan output)",
            "mode": "ro",  # read-only
            "type": "raw",
            "maxlen": 128
        },
        "107": {
            "name": "setting_para",
            "description": "设置参数 (Setting parameters)", 
            "mode": "rw",  # read-write
            "type": "raw",
            "maxlen": 128
        },
        "108": {
            "name": "alarm",
            "description": "报警 (Alarm status)",
            "mode": "ro",  # read-only
            "type": "bitmap",
            "alarms": [
                "p1_temp_ah_alm", "p2_temp_ah_alm", "p3_temp_ah_alm", "p4_temp_ah_alm",
                "p1_temp_al_alm", "p2_temp_al_alm", "p3_temp_al_alm", "p4_temp_al_alm", 
                "p1_lost_alm", "p2_lost_alm", "p3_lost_alm", "p4_lost_alm",
                "in_temp_ah_alm", "grill_open_alm"
            ]
        },
        "116": {
            "name": "Probe_Preset_Extra_1",
            "description": "预设值附加数据1 (Preset additional data 1)",
            "mode": "rw",
            "type": "raw", 
            "maxlen": 128
        },
        "117": {
            "name": "Probe_Preset_Extra_2", 
            "description": "预设值附加数据2 (Preset additional data 2)",
            "mode": "rw",
            "type": "raw",
            "maxlen": 128
        },
        "120": {
            "name": "unknown_boolean",
            "description": "Unknown boolean value - possibly power/enable state",
            "mode": "rw",
            "type": "boolean"
        }
    }
    
    for dps_id, info in known_schema.items():
        print(f"📍 DPS {dps_id}: {info['name']}")
        print(f"   Description: {info['description']}")
        print(f"   Mode: {'Read/Write' if info['mode'] == 'rw' else 'Read-Only'}")
        print(f"   Type: {info['type']}")
        print()

def test_multiple_reads():
    """Take several readings to see if values change"""
    print("\n📈 Taking Multiple Readings (watching for changes)")
    print("-" * 40)
    
    device = create_device()
    readings = []
    
    for i in range(5):
        try:
            status = device.status()
            timestamp = datetime.now().isoformat()
            
            if status and 'dps' in status:
                reading = {
                    'timestamp': timestamp,
                    'reading': i + 1,
                    'dps': status['dps']
                }
                readings.append(reading)
                
                print(f"📊 Reading {i+1}: {status['dps']}")
                
                # Wait between readings
                if i < 4:
                    time.sleep(10)
                    
            else:
                print(f"❌ Reading {i+1} failed: {status}")
                
        except Exception as e:
            print(f"❌ Reading {i+1} error: {e}")
    
    # Analyze changes
    print(f"\n📊 Analysis of {len(readings)} readings:")
    if len(readings) > 1:
        for dps_id in readings[0]['dps'].keys():
            values = [r['dps'].get(dps_id) for r in readings]
            unique_values = list(set(str(v) for v in values))
            
            if len(unique_values) > 1:
                print(f"   DPS {dps_id}: CHANGING values: {unique_values}")
            else:
                print(f"   DPS {dps_id}: STABLE value: {values[0]}")
    
    return readings

def test_simple_commands():
    """Try some simple commands to see what the device accepts"""
    print("\n🎛️ Testing Simple Commands")
    print("-" * 40)
    
    device = create_device()
    
    # Test commands based on what we know
    test_commands = [
        {"dps": 120, "value": True, "description": "Set DPS 120 to True"},
        {"dps": 120, "value": False, "description": "Set DPS 120 to False"},
        # Don't test read-only DPS like 108 (alarms) or 101 (real-time data)
    ]
    
    for cmd in test_commands:
        try:
            print(f"🔧 Testing: {cmd['description']}")
            
            # Get current value first
            current_status = device.status()
            if current_status and 'dps' in current_status:
                current_value = current_status['dps'].get(str(cmd['dps']))
                print(f"   Current DPS {cmd['dps']}: {current_value}")
            
            # Send command
            result = device.set_value(cmd['dps'], cmd['value'])
            print(f"   Command result: {result}")
            
            # Check new value
            time.sleep(2)
            new_status = device.status() 
            if new_status and 'dps' in new_status:
                new_value = new_status['dps'].get(str(cmd['dps']))
                print(f"   New DPS {cmd['dps']}: {new_value}")
                
                if str(new_value) != str(current_value):
                    print(f"   ✅ Command worked! Changed from {current_value} to {new_value}")
                else:
                    print(f"   ❓ Value unchanged - command may not have worked")
            
            print()
            
        except Exception as e:
            print(f"   ❌ Command error: {e}")
            print()

def main():
    print("🚀 INKBIRD Device Data Point Exploration")
    print("=" * 60)
    print(f"Device ID: {DEVICE_ID}")
    print(f"Device IP: {INKBIRD_IP}")
    print(f"Protocol: {PROTOCOL_VERSION}")
    print()
    
    # Step 1: Basic status
    basic_status = explore_basic_status()
    
    # Step 2: DPS discovery 
    dps_discovery = discover_all_dps()
    
    # Step 3: Analyze meanings
    analyze_dps_meanings()
    
    # Step 4: Multiple readings
    readings = test_multiple_reads()
    
    # Step 5: Test commands
    test_simple_commands()
    
    # Save results
    results = {
        'timestamp': datetime.now().isoformat(),
        'basic_status': basic_status,
        'dps_discovery': dps_discovery,
        'multiple_readings': readings,
        'device_config': {
            'device_id': DEVICE_ID,
            'ip': INKBIRD_IP,
            'protocol': PROTOCOL_VERSION
        }
    }
    
    with open('device_exploration_results.json', 'w') as f:
        json.dump(results, f, indent=2)
    
    print("💾 Results saved to device_exploration_results.json")
    print("\n🎯 Next Steps:")
    print("1. Analyze the results to understand temperature data format")  
    print("2. Test commands to control fan speed")
    print("3. Build Python controller class")
    print("4. Create Flask web interface")

if __name__ == "__main__":
    main()