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()