Compare commits

..

4 Commits

Author SHA1 Message Date
1a28d18234 Format code 2026-03-10 21:52:41 -06:00
38f4d5c5e2 feat: Track last cast IP and validate stop requests
Co-authored-by: aider (gemini/gemini-2.5-pro) <aider@aider.chat>
2026-03-10 21:48:24 -06:00
f06e728513 refactor: Add LAST_CAST_ADDRESS and streamline subprocess error handling 2026-03-10 21:48:21 -06:00
5b14666381 refactor: Make /cast API accept IP and use a single cast function
Co-authored-by: aider (gemini/gemini-2.5-pro) <aider@aider.chat>
2026-03-10 21:43:50 -06:00

125
main.py
View File

@@ -6,63 +6,52 @@ logging.basicConfig(
from flask import Flask, request from flask import Flask, request
# Constants
AUTO_STOP_TIMEOUT_SECONDS = 3600 # 1 hour AUTO_STOP_TIMEOUT_SECONDS = 3600 # 1 hour
LAST_CAST_ADDRESS = None
app = Flask(__name__) app = Flask(__name__)
auto_stop_timer = None # Timer for automatic VNC stop auto_stop_timer = None # Timer for automatic VNC stop
@app.route('/cast', methods=['POST']) @app.route('/cast', methods=['POST'])
def cast_spell(): def cast_spell():
machine = request.form.get('machine') global auto_stop_timer, LAST_CAST_ADDRESS
logging.info(f"Received POST request on /cast. Requested machine: {machine}")
global auto_stop_timer ip_address = request.form.get('ip_address')
logging.info(f"Received POST request on /cast. Requested ip_address: {ip_address}")
if machine == "trotec": if not ip_address:
logging.info("Casting to Trotec.") logging.warning(f"Invalid or missing ip_address parameter: {ip_address}")
kill_vnc() # Kill any existing VNC session first return f"Invalid or missing 'ip_address' parameter.", 400
cast_trotec()
# Restart auto-stop timer logging.info(f"Casting to {ip_address}.")
if auto_stop_timer is not None and auto_stop_timer.is_alive(): kill_vnc() # Kill any existing VNC session first
auto_stop_timer.cancel() cast_machine(ip_address)
logging.info("Cancelled previous auto-stop timer.") LAST_CAST_ADDRESS = ip_address
auto_stop_timer = threading.Timer(AUTO_STOP_TIMEOUT_SECONDS, _auto_stop_vnc)
auto_stop_timer.start() # Restart auto-stop timer
logging.info(f"Scheduled auto-stop for VNC in {AUTO_STOP_TIMEOUT_SECONDS} seconds.") if auto_stop_timer is not None and auto_stop_timer.is_alive():
return f"Successfully cast to Trotec.", 200 auto_stop_timer.cancel()
elif machine == "thunder": logging.info("Cancelled previous auto-stop timer.")
logging.info("Casting to Thunder.") auto_stop_timer = threading.Timer(AUTO_STOP_TIMEOUT_SECONDS, _auto_stop_vnc)
kill_vnc() # Kill any existing VNC session first auto_stop_timer.start()
cast_thunder() logging.info(f"Scheduled auto-stop for VNC in {AUTO_STOP_TIMEOUT_SECONDS} seconds.")
# Restart auto-stop timer
if auto_stop_timer is not None and auto_stop_timer.is_alive(): return f"Successfully cast to {ip_address}.", 200
auto_stop_timer.cancel()
logging.info("Cancelled previous auto-stop timer.")
auto_stop_timer = threading.Timer(AUTO_STOP_TIMEOUT_SECONDS, _auto_stop_vnc)
auto_stop_timer.start()
logging.info(f"Scheduled auto-stop for VNC in {AUTO_STOP_TIMEOUT_SECONDS} seconds.")
return f"Successfully cast to Thunder.", 200
elif machine == "xtool":
logging.info("Casting to XTool.")
kill_vnc() # Kill any existing VNC session first
cast_xtool()
# Restart auto-stop timer
if auto_stop_timer is not None and auto_stop_timer.is_alive():
auto_stop_timer.cancel()
logging.info("Cancelled previous auto-stop timer.")
auto_stop_timer = threading.Timer(AUTO_STOP_TIMEOUT_SECONDS, _auto_stop_vnc)
auto_stop_timer.start()
logging.info(f"Scheduled auto-stop for VNC in {AUTO_STOP_TIMEOUT_SECONDS} seconds.")
return f"Successfully cast to XTool.", 200
else:
logging.warning(f"Invalid or missing machine parameter: {machine}")
return f"Invalid or missing 'machine' parameter. Use 'trotec' or 'thunder'.", 400
@app.route('/stop', methods=['POST']) @app.route('/stop', methods=['POST'])
def stop_cast(): def stop_cast():
global auto_stop_timer global auto_stop_timer, LAST_CAST_ADDRESS
logging.info("Received POST request on /stop.")
ip_address = request.form.get('ip_address')
logging.info(f"Received POST request on /stop for ip_address: {ip_address}")
if not ip_address:
logging.warning(f"Invalid or missing ip_address parameter: {ip_address}")
return f"Invalid or missing 'ip_address' parameter.", 400
if ip_address != LAST_CAST_ADDRESS:
logging.warning(f"Stop request for {ip_address} does not match last cast address {LAST_CAST_ADDRESS}.")
return f"Stop request for {ip_address} does not match last cast address.", 403
if auto_stop_timer is not None and auto_stop_timer.is_alive(): if auto_stop_timer is not None and auto_stop_timer.is_alive():
auto_stop_timer.cancel() auto_stop_timer.cancel()
@@ -70,6 +59,8 @@ def stop_cast():
logging.info("Manual stop: auto-stop timer cancelled.") logging.info("Manual stop: auto-stop timer cancelled.")
kill_vnc() kill_vnc()
LAST_CAST_ADDRESS = None
return "Attempted to stop VNC viewers.", 200 return "Attempted to stop VNC viewers.", 200
def _auto_stop_vnc(): def _auto_stop_vnc():
@@ -79,50 +70,16 @@ def _auto_stop_vnc():
kill_vnc() kill_vnc()
auto_stop_timer = None # Clear the timer reference auto_stop_timer = None # Clear the timer reference
def cast_trotec(): def cast_machine(ip_address):
"""Executes the xtightvncviewer command.""" """Executes the xtightvncviewer command for a given IP address."""
command = "DISPLAY=:1 xtightvncviewer -viewonly -fullscreen 172.17.17.214" command = f"DISPLAY=:1 xtightvncviewer -viewonly -fullscreen {ip_address}"
try: try:
logging.info(f"Launching command: {command}") logging.info(f"Launching command: {command}")
# Use Popen to run the command in the background (non-blocking) # Use Popen to run the command in the background (non-blocking)
subprocess.Popen(command, shell=True) subprocess.Popen(command, shell=True)
logging.info(f"Command '{command}' launched successfully.") logging.info(f"Command '{command}' launched successfully.")
except FileNotFoundError:
# This error is more likely if the shell itself (e.g., /bin/sh) is not found,
# or if 'xtightvncviewer' is not in PATH and the shell fails to find it.
logging.error(f"Failed to launch VNC for Trotec: Shell or VNC command might not be found. Ensure xtightvncviewer is in PATH and shell is available.")
except Exception as e: # Catch other potential errors during Popen except Exception as e: # Catch other potential errors during Popen
logging.error(f"An error occurred while launching VNC for Trotec with Popen: {e}") logging.error(f"An error occurred while launching VNC for {ip_address} with Popen: {e}")
def cast_thunder():
"""Executes the xtightvncviewer command for Thunder."""
command = "DISPLAY=:1 xtightvncviewer -viewonly -fullscreen 172.17.17.215"
try:
logging.info(f"Launching command: {command}")
# Use Popen to run the command in the background (non-blocking)
subprocess.Popen(command, shell=True)
logging.info(f"Command '{command}' launched successfully.")
except FileNotFoundError:
# This error is more likely if the shell itself (e.g., /bin/sh) is not found,
# or if 'xtightvncviewer' is not in PATH and the shell fails to find it.
logging.error(f"Failed to launch VNC for Thunder: Shell or VNC command might not be found. Ensure xtightvncviewer is in PATH and shell is available.")
except Exception as e: # Catch other potential errors during Popen
logging.error(f"An error occurred while launching VNC for Thunder with Popen: {e}")
def cast_xtool():
"""Executes the xtightvncviewer command for XTool."""
command = "DISPLAY=:1 xtightvncviewer -viewonly -fullscreen 172.17.17.221"
try:
logging.info(f"Launching command: {command}")
# Use Popen to run the command in the background (non-blocking)
subprocess.Popen(command, shell=True)
logging.info(f"Command '{command}' launched successfully.")
except FileNotFoundError:
# This error is more likely if the shell itself (e.g., /bin/sh) is not found,
# or if 'xtightvncviewer' is not in PATH and the shell fails to find it.
logging.error(f"Failed to launch VNC for XTool: Shell or VNC command might not be found. Ensure xtightvncviewer is in PATH and shell is available.")
except Exception as e: # Catch other potential errors during Popen
logging.error(f"An error occurred while launching VNC for XTool with Popen: {e}")
def kill_vnc(): def kill_vnc():
"""Executes the killall command for xtightvncviewer.""" """Executes the killall command for xtightvncviewer."""
@@ -141,8 +98,6 @@ def kill_vnc():
if "no process found" in result.stderr.lower(): if "no process found" in result.stderr.lower():
logging.info("No xtightvncviewer processes were found to kill.") logging.info("No xtightvncviewer processes were found to kill.")
except FileNotFoundError:
logging.error(f"Command not found: killall. Please ensure it is installed and in PATH.")
except Exception as e: except Exception as e:
logging.error(f"An unexpected error occurred while trying to run killall: {e}") logging.error(f"An unexpected error occurred while trying to run killall: {e}")