95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
import os
|
|
import requests
|
|
import gradio as gr
|
|
import subprocess
|
|
|
|
# FHEM URL
|
|
FHEM_URL = "https://fhem.auwiesen2.de/fhem"
|
|
|
|
# Parameters including CSRF token
|
|
PARAMS = {
|
|
"cmd": "jsonlist2",
|
|
"XHR": "1",
|
|
"fwcsrf": "csrf_611440676390392" # CSRF token as a parameter
|
|
}
|
|
|
|
# Headers including CSRF token
|
|
HEADERS = {
|
|
"X-FHEM-csrfToken": "csrf_611440676390392",
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"Accept": "application/json"
|
|
}
|
|
|
|
def get_device_list():
|
|
try:
|
|
# Send the request with both parameters and headers
|
|
response = requests.get(FHEM_URL, params=PARAMS, headers=HEADERS, timeout=10)
|
|
|
|
# Debugging: Check the status code and response content
|
|
print("Response Status Code:", response.status_code)
|
|
print("Response Content:", response.text)
|
|
|
|
response.raise_for_status() # Raise an HTTPError for bad responses (4xx and 5xx)
|
|
|
|
# Attempt to parse the response as JSON
|
|
data = response.json()
|
|
print("Parsed JSON Data:", data) # Debugging: Print the parsed JSON data
|
|
|
|
devices = data.get('Results', [])
|
|
|
|
# Prepare data for Gradio (convert dictionaries to tuples)
|
|
device_list = []
|
|
for device in devices:
|
|
device_tuple = (device['Name'], device['Internals'].get('STATE', 'Unknown'))
|
|
device_list.append(device_tuple)
|
|
|
|
return device_list
|
|
except requests.exceptions.JSONDecodeError:
|
|
return "Error: Failed to decode JSON response."
|
|
except KeyError:
|
|
return "Error: Expected keys not found in the JSON data."
|
|
except requests.exceptions.RequestException as e:
|
|
return f"Error: {str(e)}"
|
|
|
|
def display_devices():
|
|
device_list = get_device_list()
|
|
if isinstance(device_list, str): # Check if an error message was returned
|
|
return device_list
|
|
else:
|
|
# Return the device list as a Gradio Dataframe component
|
|
return gr.Dataframe(device_list, headers=["Device", "State"])
|
|
|
|
# Create a Gradio interface
|
|
interface = gr.Interface(fn=display_devices, inputs=[], outputs=gr.Dataframe(headers=["Device", "State"]))
|
|
|
|
# List of ports to try
|
|
ports_to_try = [8081, 8082]
|
|
|
|
# Function to terminate processes using a specific port
|
|
def kill_process_on_port(port):
|
|
try:
|
|
pid = subprocess.check_output(f"lsof -t -i:{port}", shell=True).decode().strip()
|
|
if pid:
|
|
os.system(f"kill {pid}")
|
|
print(f"Terminated process {pid} on port {port}")
|
|
except subprocess.CalledProcessError:
|
|
print(f"No process found on port {port}")
|
|
|
|
# Terminate any process that might be using the desired ports
|
|
for port in ports_to_try:
|
|
kill_process_on_port(port)
|
|
|
|
# Try each port until one is available
|
|
for port in ports_to_try:
|
|
try:
|
|
# Launch the interface with the specified port
|
|
interface.launch(
|
|
server_name="0.0.0.0", # Bind to all IPs inside the Docker container
|
|
server_port=port, # Try the current port
|
|
share=False, # Do not create a public link since you use a custom domain
|
|
inbrowser=False # Do not automatically open a browser
|
|
)
|
|
break # If launch is successful, break out of the loop
|
|
except OSError:
|
|
print(f"Port {port} is not available. Trying next port...")
|