Spaces:
Sleeping
Sleeping
| 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() |