app.py

import gradio as gr
import requests
import time
from datetime import datetime
import json
import threading
import queue

class EzvizTuyaAutomation:
    def __init__(self):
        self.ezviz_base_url = "https://open.ezviz.com/api"
        self.tuya_base_url = "https://openapi.tuyaus.com"
        self.credentials = {
            'ezviz_app_key': '',
            'ezviz_app_secret': '',
            'tuya_client_id': '',
            'tuya_secret': ''
        }
        self.ezviz_access_token = None
        self.tuya_access_token = None
        self.selected_camera = None
        self.selected_light = None
        self.is_running = False
        self.status_queue = queue.Queue()

    def get_ezviz_devices(self):
        """Get list of EZVIZ cameras"""
        if not self.ezviz_access_token:
            return []
        
        url = f"{self.ezviz_base_url}/devices"
        headers = {"Authorization": f"Bearer {self.ezviz_access_token}"}
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            devices = response.json().get("devices", [])
            return [{"name": dev["name"], "id": dev["deviceSerial"]} for dev in devices]
        print(f"EZVIZ API Response: {response.text}")  # Debugging
        return []

    def get_tuya_devices(self):
        """Get list of Tuya devices"""
        if not self.tuya_access_token:
            return []
        
        url = f"{self.tuya_base_url}/v1.0/devices"
        headers = {
            "Authorization": f"Bearer {self.tuya_access_token}",
            "Content-Type": "application/json"
        }
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            devices = response.json().get("devices", [])
            return [{"name": dev["name"], "id": dev["id"]} for dev in devices]
        print(f"Tuya API Response: {response.text}")  # Debugging
        return []

    def validate_tokens(self):
        if not self.ezviz_access_token:
            return "EZVIZ access token is missing or invalid."
        if not self.tuya_access_token:
            return "Tuya access token is missing or invalid."
        return "Tokens validated successfully."

    def check_motion(self):
        """Check for motion events from selected EZVIZ camera"""
        if not self.ezviz_access_token or not self.selected_camera:
            return False

        url = f"{self.ezviz_base_url}/devices/{self.selected_camera}/events"
        headers = {"Authorization": f"Bearer {self.ezviz_access_token}"}
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            events = response.json().get("events", [])
            current_time = time.time()
            for event in events:
                if (event.get("type") == "motion" and 
                    current_time - event.get("timestamp", 0) <= 30):
                    return True
        return False

    def control_light(self, turn_on=True):
        """Control selected Tuya device"""
        if not self.tuya_access_token or not self.selected_light:
            return False

        url = f"{self.tuya_base_url}/v1.0/devices/{self.selected_light}/commands"
        headers = {
            "Authorization": f"Bearer {self.tuya_access_token}",
            "Content-Type": "application/json"
        }
        command = {
            "commands": [
                {
                    "code": "switch_led",
                    "value": turn_on
                }
            ]
        }
        response = requests.post(url, headers=headers, json=command)
        return response.status_code == 200

    def automation_loop(self):
        """Main automation loop"""
        last_motion_time = 0
        while self.is_running:
            try:
                if self.check_motion():
                    current_time = time.time()
                    if current_time - last_motion_time > 30:
                        message = f"Motion detected at {datetime.now()}"
                        self.status_queue.put(message)
                        if self.control_light(True):
                            self.status_queue.put("Light turned on successfully")
                            last_motion_time = current_time
                        else:
                            self.status_queue.put("Failed to control light")
                time.sleep(5)
            except Exception as e:
                self.status_queue.put(f"Error: {str(e)}")
                time.sleep(30)

def create_ui():
    automation = EzvizTuyaAutomation()
    
    def update_credentials(ezviz_key, ezviz_secret, tuya_id, tuya_secret):
        automation.credentials.update({
            'ezviz_app_key': ezviz_key,
            'ezviz_app_secret': ezviz_secret,
            'tuya_client_id': tuya_id,
            'tuya_secret': tuya_secret
        })
        return "Credentials updated"

    def get_devices():
        # Attempt to fetch devices
        try:
            ezviz_devices = automation.get_ezviz_devices()
            tuya_devices = automation.get_tuya_devices()
            
            if not ezviz_devices:
                ezviz_choices = ["No devices found or invalid credentials."]
            else:
                ezviz_choices = [f"{dev['name']} ({dev['id']})" for dev in ezviz_devices]
            
            if not tuya_devices:
                tuya_choices = ["No devices found or invalid credentials."]
            else:
                tuya_choices = [f"{dev['name']} ({dev['id']})" for dev in tuya_devices]
            
            return ezviz_choices, tuya_choices
        
        except Exception as e:
            print(f"Error fetching devices: {e}")
            return ["Error fetching EZVIZ devices"], ["Error fetching Tuya devices"]

    def start_automation(camera_choice, light_choice):
        if not camera_choice or not light_choice:
            return "Please select both a camera and a light device"
        
        automation.selected_camera = camera_choice.split('(')[1].rstrip(')')
        automation.selected_light = light_choice.split('(')[1].rstrip(')')
        automation.is_running = True
        
        # Start automation in a separate thread
        thread = threading.Thread(target=automation.automation_loop)
        thread.daemon = True
        thread.start()
        
        return "Automation started"

    def stop_automation():
        automation.is_running = False
        return "Automation stopped"

    def get_status():
        status_messages = []
        while not automation.status_queue.empty():
            status_messages.append(automation.status_queue.get())
        return "\n".join(status_messages) if status_messages else "No new updates"

    with gr.Blocks(title="EZVIZ-Tuya Automation") as app:
        with gr.Row():
            with gr.Column():
                gr.Markdown("### API Credentials")
                ezviz_key = gr.Textbox(label="EZVIZ App Key")
                ezviz_secret = gr.Textbox(label="EZVIZ App Secret", type="password")
                tuya_id = gr.Textbox(label="Tuya Client ID")
                tuya_secret = gr.Textbox(label="Tuya Secret", type="password")
                save_btn = gr.Button("Save Credentials")
                validate_btn = gr.Button("Validate Tokens")
                validate_status = gr.Textbox(label="Token Validation Status", interactive=False)

        with gr.Row():
            refresh_btn = gr.Button("Refresh Device Lists")
            camera_dropdown = gr.Dropdown(label="Select EZVIZ Camera", choices=[])
            light_dropdown = gr.Dropdown(label="Select Tuya Light", choices=[])

        with gr.Row():
            start_btn = gr.Button("Start Automation")
            stop_btn = gr.Button("Stop Automation")

        status_text = gr.Textbox(label="Status", interactive=False)
        
        # Update status every 5 seconds
        status_text.change(get_status, None, status_text, every=5)
        
        # Button click handlers
        save_btn.click(update_credentials, 
                      inputs=[ezviz_key, ezviz_secret, tuya_id, tuya_secret],
                      outputs=gr.Textbox(label="Status"))
        
        validate_btn.click(automation.validate_tokens, None, validate_status)
        
        refresh_btn.click(get_devices, 
                         None, 
                         outputs=[camera_dropdown, light_dropdown])
        
        start_btn.click(start_automation,
                       inputs=[camera_dropdown, light_dropdown],
                       outputs=gr.Textbox(label="Status"))
        
        stop_btn.click(stop_automation,
                      None,
                      outputs=gr.Textbox(label="Status"))

    return app

if __name__ == "__main__":
    app = create_ui()
    app.launch()