Create app.py

app.py

@@ -0,0 +1,205 @@
+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]
+        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]
+        return []
+
+    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():
+        # Get and refresh device lists
+        ezviz_devices = automation.get_ezviz_devices()
+        tuya_devices = automation.get_tuya_devices()
+        
+        ezviz_choices = [f"{dev['name']} ({dev['id']})" for dev in ezviz_devices]
+        tuya_choices = [f"{dev['name']} ({dev['id']})" for dev in tuya_devices]
+        
+        return gr.Dropdown.update(choices=ezviz_choices), gr.Dropdown.update(choices=tuya_choices)
+
+    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")
+
+        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"))
+        
+        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()