Move modules back to functions

dune_weaver_flask/app.py

@@ -2,10 +2,10 @@ from flask import Flask, request, jsonify, render_template, send_from_directory
 import atexit
 import os
 from datetime import datetime
-from .modules.serial.serial_manager import serial_manager
-from dune_weaver_flask.modules.core.pattern_manager import pattern_manager
-from dune_weaver_flask.modules.core.playlist_manager import playlist_manager
-from .modules.firmware.firmware_manager import firmware_manager
+from .modules.serial import serial_manager
+from dune_weaver_flask.modules.core import pattern_manager
+from dune_weaver_flask.modules.core import playlist_manager
+from .modules.firmware import firmware_manager
 
 app = Flask(__name__)
 

dune_weaver_flask/modules/core/pattern_manager.py

@@ -4,276 +4,277 @@ import time
 import random
 from datetime import datetime
 from tqdm import tqdm
-from dune_weaver_flask.modules.serial.serial_manager import serial_manager
-
-class PatternManager:
-    def __init__(self):
-        self.THETA_RHO_DIR = './patterns'
-        self.CLEAR_PATTERNS = {
-            "clear_from_in":  "./patterns/clear_from_in.thr",
-            "clear_from_out": "./patterns/clear_from_out.thr",
-            "clear_sideway":  "./patterns/clear_sideway.thr"
-        }
-        os.makedirs(self.THETA_RHO_DIR, exist_ok=True)
-
-        # Execution state
-        self.stop_requested = False
-        self.pause_requested = False
-        self.pause_condition = threading.Condition()
-        self.current_playing_file = None
-        self.execution_progress = None
-        self.current_playing_index = None
-        self.current_playlist = None
-        self.is_clearing = False
-
-    def parse_theta_rho_file(self, file_path):
-        """Parse a theta-rho file and return a list of (theta, rho) pairs."""
-        coordinates = []
-        try:
-            with open(file_path, 'r') as file:
-                for line in file:
-                    line = line.strip()
-                    if not line or line.startswith("#"):
-                        continue
-                    try:
-                        theta, rho = map(float, line.split())
-                        coordinates.append((theta, rho))
-                    except ValueError:
-                        print(f"Skipping invalid line: {line}")
-                        continue
-        except Exception as e:
-            print(f"Error reading file: {e}")
-            return coordinates
-
-        # Normalization Step
-        if coordinates:
-            first_theta = coordinates[0][0]
-            normalized = [(theta - first_theta, rho) for theta, rho in coordinates]
-            coordinates = normalized
-
+from dune_weaver_flask.modules.serial import serial_manager
+
+# Global state
+THETA_RHO_DIR = './patterns'
+CLEAR_PATTERNS = {
+    "clear_from_in":  "./patterns/clear_from_in.thr",
+    "clear_from_out": "./patterns/clear_from_out.thr",
+    "clear_sideway":  "./patterns/clear_sideway.thr"
+}
+os.makedirs(THETA_RHO_DIR, exist_ok=True)
+
+# Execution state
+stop_requested = False
+pause_requested = False
+pause_condition = threading.Condition()
+current_playing_file = None
+execution_progress = None
+current_playing_index = None
+current_playlist = None
+is_clearing = False
+
+def parse_theta_rho_file(file_path):
+    """Parse a theta-rho file and return a list of (theta, rho) pairs."""
+    coordinates = []
+    try:
+        with open(file_path, 'r') as file:
+            for line in file:
+                line = line.strip()
+                if not line or line.startswith("#"):
+                    continue
+                try:
+                    theta, rho = map(float, line.split())
+                    coordinates.append((theta, rho))
+                except ValueError:
+                    print(f"Skipping invalid line: {line}")
+                    continue
+    except Exception as e:
+        print(f"Error reading file: {e}")
         return coordinates
 
-    def get_clear_pattern_file(self, clear_pattern_mode, path=None):
-        """Return a .thr file path based on pattern_name."""
-        if not clear_pattern_mode or clear_pattern_mode == 'none':
-            return
-        print("Clear pattern mode: " + clear_pattern_mode)
-        if clear_pattern_mode == "random":
-            return random.choice(list(self.CLEAR_PATTERNS.values()))
-
-        if clear_pattern_mode == 'adaptive':
-            _, first_rho = self.parse_theta_rho_file(path)[0]
-            if first_rho < 0.5:
-                return self.CLEAR_PATTERNS['clear_from_out']
-            else:
-                return random.choice([self.CLEAR_PATTERNS['clear_from_in'], self.CLEAR_PATTERNS['clear_sideway']])
+    # Normalization Step
+    if coordinates:
+        first_theta = coordinates[0][0]
+        normalized = [(theta - first_theta, rho) for theta, rho in coordinates]
+        coordinates = normalized
+
+    return coordinates
+
+def get_clear_pattern_file(clear_pattern_mode, path=None):
+    """Return a .thr file path based on pattern_name."""
+    if not clear_pattern_mode or clear_pattern_mode == 'none':
+        return
+    print("Clear pattern mode: " + clear_pattern_mode)
+    if clear_pattern_mode == "random":
+        return random.choice(list(CLEAR_PATTERNS.values()))
+
+    if clear_pattern_mode == 'adaptive':
+        _, first_rho = parse_theta_rho_file(path)[0]
+        if first_rho < 0.5:
+            return CLEAR_PATTERNS['clear_from_out']
         else:
-            return self.CLEAR_PATTERNS[clear_pattern_mode]
-
-    def schedule_checker(self, schedule_hours):
-        """Pauses/resumes execution based on a given time range."""
-        if not schedule_hours:
-            return
-
-        start_time, end_time = schedule_hours
+            return random.choice([CLEAR_PATTERNS['clear_from_in'], CLEAR_PATTERNS['clear_sideway']])
+    else:
+        return CLEAR_PATTERNS[clear_pattern_mode]
+
+def schedule_checker(schedule_hours):
+    """Pauses/resumes execution based on a given time range."""
+    global pause_requested
+    if not schedule_hours:
+        return
+
+    start_time, end_time = schedule_hours
+    now = datetime.now().time()
+
+    if start_time <= now < end_time:
+        if pause_requested:
+            print("Starting execution: Within schedule.")
+        pause_requested = False
+        with pause_condition:
+            pause_condition.notify_all()
+    else:
+        if not pause_requested:
+            print("Pausing execution: Outside schedule.")
+        pause_requested = True
+        threading.Thread(target=wait_for_start_time, args=(schedule_hours,), daemon=True).start()
+
+def wait_for_start_time(schedule_hours):
+    """Keep checking every 30 seconds if the time is within the schedule to resume execution."""
+    global pause_requested
+    start_time, end_time = schedule_hours
+
+    while pause_requested:
         now = datetime.now().time()
-
         if start_time <= now < end_time:
-            if self.pause_requested:
-                print("Starting execution: Within schedule.")
-            self.pause_requested = False
-            with self.pause_condition:
-                self.pause_condition.notify_all()
+            print("Resuming execution: Within schedule.")
+            pause_requested = False
+            with pause_condition:
+                pause_condition.notify_all()
+            break
         else:
-            if not self.pause_requested:
-                print("Pausing execution: Outside schedule.")
-            self.pause_requested = True
-            threading.Thread(target=self.wait_for_start_time, args=(schedule_hours,), daemon=True).start()
-
-    def wait_for_start_time(self, schedule_hours):
-        """Keep checking every 30 seconds if the time is within the schedule to resume execution."""
-        start_time, end_time = schedule_hours
-
-        while self.pause_requested:
-            now = datetime.now().time()
-            if start_time <= now < end_time:
-                print("Resuming execution: Within schedule.")
-                self.pause_requested = False
-                with self.pause_condition:
-                    self.pause_condition.notify_all()
-                break
-            else:
-                time.sleep(30)
-
-    def run_theta_rho_file(self, file_path, schedule_hours=None):
-        """Run a theta-rho file by sending data in optimized batches with tqdm ETA tracking."""
-        coordinates = self.parse_theta_rho_file(file_path)
-        total_coordinates = len(coordinates)
-
-        if total_coordinates < 2:
-            print("Not enough coordinates for interpolation.")
-            self.current_playing_file = None
-            self.execution_progress = None
-            return
-
-        self.execution_progress = (0, total_coordinates, None)
-        batch_size = 10
-
-        self.stop_actions()
-        with serial_manager.serial_lock:
-            self.current_playing_file = file_path
-            self.execution_progress = (0, 0, None)
-            self.stop_requested = False
-            with tqdm(total=total_coordinates, unit="coords", desc=f"Executing Pattern {file_path}", dynamic_ncols=True, disable=None) as pbar:
-                for i in range(0, total_coordinates, batch_size):
-                    if self.stop_requested:
-                        print("Execution stopped by user after completing the current batch.")
-                        break
+            time.sleep(30)
+
+def run_theta_rho_file(file_path, schedule_hours=None):
+    """Run a theta-rho file by sending data in optimized batches with tqdm ETA tracking."""
+    global current_playing_file, execution_progress, stop_requested
+    coordinates = parse_theta_rho_file(file_path)
+    total_coordinates = len(coordinates)
+
+    if total_coordinates < 2:
+        print("Not enough coordinates for interpolation.")
+        current_playing_file = None
+        execution_progress = None
+        return
+
+    execution_progress = (0, total_coordinates, None)
+    batch_size = 10
+
+    stop_actions()
+    with serial_manager.serial_lock:
+        current_playing_file = file_path
+        execution_progress = (0, 0, None)
+        stop_requested = False
+        with tqdm(total=total_coordinates, unit="coords", desc=f"Executing Pattern {file_path}", dynamic_ncols=True, disable=None) as pbar:
+            for i in range(0, total_coordinates, batch_size):
+                if stop_requested:
+                    print("Execution stopped by user after completing the current batch.")
+                    break
+
+                with pause_condition:
+                    while pause_requested:
+                        print("Execution paused...")
+                        pause_condition.wait()
+
+                batch = coordinates[i:i + batch_size]
+
+                if i == 0:
+                    serial_manager.send_coordinate_batch(batch)
+                    execution_progress = (i + batch_size, total_coordinates, None)
+                    pbar.update(batch_size)
+                    continue
+
+                while True:
+                    schedule_checker(schedule_hours)
+                    if serial_manager.ser.in_waiting > 0:
+                        response = serial_manager.ser.readline().decode().strip()
+                        if response == "R":
+                            serial_manager.send_coordinate_batch(batch)
+                            pbar.update(batch_size)
+                            estimated_remaining_time = pbar.format_dict['elapsed'] / (i + batch_size) * (total_coordinates - (i + batch_size))
+                            execution_progress = (i + batch_size, total_coordinates, estimated_remaining_time)
+                            break
+                        elif response != "IGNORED: FINISHED" and response.startswith("IGNORE"):
+                            print("Received IGNORE. Resending the previous batch...")
+                            print(response)
+                            prev_start = max(0, i - batch_size)
+                            prev_end = i
+                            previous_batch = coordinates[prev_start:prev_end]
+                            serial_manager.send_coordinate_batch(previous_batch)
+                            break
+                        else:
+                            print(f"Arduino response: {response}")
+
+        reset_theta()
+        serial_manager.ser.write("FINISHED\n".encode())
+
+    current_playing_file = None
+    execution_progress = None
+    print("Pattern execution completed.")
+
+def run_theta_rho_files(file_paths, pause_time=0, clear_pattern=None, run_mode="single", shuffle=False, schedule_hours=None):
+    """Run multiple .thr files in sequence with options."""
+    global stop_requested, current_playlist, current_playing_index
+    stop_requested = False
+    
+    if shuffle:
+        random.shuffle(file_paths)
+        print("Playlist shuffled.")
+
+    current_playlist = file_paths
+
+    while True:
+        for idx, path in enumerate(file_paths):
+            print("Upcoming pattern: " + path)
+            current_playing_index = idx
+            schedule_checker(schedule_hours)
+            if stop_requested:
+                print("Execution stopped before starting next pattern.")
+                return
+
+            if clear_pattern:
+                if stop_requested:
+                    print("Execution stopped before running the next clear pattern.")
+                    return
 
-                    with self.pause_condition:
-                        while self.pause_requested:
-                            print("Execution paused...")
-                            self.pause_condition.wait()
-
-                    batch = coordinates[i:i + batch_size]
-
-                    if i == 0:
-                        serial_manager.send_coordinate_batch(batch)
-                        self.execution_progress = (i + batch_size, total_coordinates, None)
-                        pbar.update(batch_size)
-                        continue
-
-                    while True:
-                        self.schedule_checker(schedule_hours)
-                        if serial_manager.ser.in_waiting > 0:
-                            response = serial_manager.ser.readline().decode().strip()
-                            if response == "R":
-                                serial_manager.send_coordinate_batch(batch)
-                                pbar.update(batch_size)
-                                estimated_remaining_time = pbar.format_dict['elapsed'] / (i + batch_size) * (total_coordinates - (i + batch_size))
-                                self.execution_progress = (i + batch_size, total_coordinates, estimated_remaining_time)
-                                break
-                            elif response != "IGNORED: FINISHED" and response.startswith("IGNORE"):
-                                print("Received IGNORE. Resending the previous batch...")
-                                print(response)
-                                prev_start = max(0, i - batch_size)
-                                prev_end = i
-                                previous_batch = coordinates[prev_start:prev_end]
-                                serial_manager.send_coordinate_batch(previous_batch)
-                                break
-                            else:
-                                print(f"Arduino response: {response}")
-
-            self.reset_theta()
-            serial_manager.ser.write("FINISHED\n".encode())
-
-        self.current_playing_file = None
-        self.execution_progress = None
-        print("Pattern execution completed.")
-
-    def run_theta_rho_files(self, file_paths, pause_time=0, clear_pattern=None, run_mode="single", shuffle=False, schedule_hours=None):
-        """Run multiple .thr files in sequence with options."""
-        self.stop_requested = False
-        
-        if shuffle:
-            random.shuffle(file_paths)
-            print("Playlist shuffled.")
+                clear_file_path = get_clear_pattern_file(clear_pattern, path)
+                print(f"Running clear pattern: {clear_file_path}")
+                run_theta_rho_file(clear_file_path, schedule_hours)
 
-        self.current_playlist = file_paths
+            if not stop_requested:
+                print(f"Running pattern {idx + 1} of {len(file_paths)}: {path}")
+                run_theta_rho_file(path, schedule_hours)
 
-        while True:
-            for idx, path in enumerate(file_paths):
-                print("Upcoming pattern: " + path)
-                self.current_playing_index = idx
-                self.schedule_checker(schedule_hours)
-                if self.stop_requested:
-                    print("Execution stopped before starting next pattern.")
+            if idx < len(file_paths) - 1:
+                if stop_requested:
+                    print("Execution stopped before running the next clear pattern.")
                     return
-
-                if clear_pattern:
-                    if self.stop_requested:
-                        print("Execution stopped before running the next clear pattern.")
-                        return
-
-                    clear_file_path = self.get_clear_pattern_file(clear_pattern, path)
-                    print(f"Running clear pattern: {clear_file_path}")
-                    self.run_theta_rho_file(clear_file_path, schedule_hours)
-
-                if not self.stop_requested:
-                    print(f"Running pattern {idx + 1} of {len(file_paths)}: {path}")
-                    self.run_theta_rho_file(path, schedule_hours)
-
-                if idx < len(file_paths) - 1:
-                    if self.stop_requested:
-                        print("Execution stopped before running the next clear pattern.")
-                        return
-                    if pause_time > 0:
-                        print(f"Pausing for {pause_time} seconds...")
-                        time.sleep(pause_time)
-
-            if run_mode == "indefinite":
-                print("Playlist completed. Restarting as per 'indefinite' run mode.")
                 if pause_time > 0:
-                    print(f"Pausing for {pause_time} seconds before restarting...")
+                    print(f"Pausing for {pause_time} seconds...")
                     time.sleep(pause_time)
-                if shuffle:
-                    random.shuffle(file_paths)
-                    print("Playlist reshuffled for the next loop.")
-                continue
-            else:
-                print("Playlist completed.")
-                break
-
-        self.reset_theta()
-        with serial_manager.serial_lock:
-            serial_manager.ser.write("FINISHED\n".encode())
-            
-        print("All requested patterns completed (or stopped).")
-
-    def reset_theta(self):
-        """Reset theta on the Arduino."""
-        with serial_manager.serial_lock:
-            serial_manager.ser.write("RESET_THETA\n".encode())
-            while True:
-                with serial_manager.serial_lock:
-                    if serial_manager.ser.in_waiting > 0:
-                        response = serial_manager.ser.readline().decode().strip()
-                        print(f"Arduino response: {response}")
-                        if response == "THETA_RESET":
-                            print("Theta successfully reset.")
-                            break
-                time.sleep(0.5)
-
-    def stop_actions(self):
-        """Stop all current pattern execution."""
-        with self.pause_condition:
-            self.pause_requested = False
-            self.pause_condition.notify_all()
-            
-        self.stop_requested = True
-        self.current_playing_index = None
-        self.current_playlist = None
-        self.is_clearing = False
-        self.current_playing_file = None
-        self.execution_progress = None
-
-    def get_status(self):
-        """Get the current status of pattern execution."""
-        if self.current_playing_file in self.CLEAR_PATTERNS.values():
-            self.is_clearing = True
+
+        if run_mode == "indefinite":
+            print("Playlist completed. Restarting as per 'indefinite' run mode.")
+            if pause_time > 0:
+                print(f"Pausing for {pause_time} seconds before restarting...")
+                time.sleep(pause_time)
+            if shuffle:
+                random.shuffle(file_paths)
+                print("Playlist reshuffled for the next loop.")
+            continue
         else:
-            self.is_clearing = False
-
-        return {
-            "ser_port": serial_manager.get_port(),
-            "stop_requested": self.stop_requested,
-            "pause_requested": self.pause_requested,
-            "current_playing_file": self.current_playing_file,
-            "execution_progress": self.execution_progress,
-            "current_playing_index": self.current_playing_index,
-            "current_playlist": self.current_playlist,
-            "is_clearing": self.is_clearing
-        }
-
-# Create a global instance
-pattern_manager = PatternManager()
+            print("Playlist completed.")
+            break
+
+    reset_theta()
+    with serial_manager.serial_lock:
+        serial_manager.ser.write("FINISHED\n".encode())
+        
+    print("All requested patterns completed (or stopped).")
+
+def reset_theta():
+    """Reset theta on the Arduino."""
+    with serial_manager.serial_lock:
+        serial_manager.ser.write("RESET_THETA\n".encode())
+        while True:
+            with serial_manager.serial_lock:
+                if serial_manager.ser.in_waiting > 0:
+                    response = serial_manager.ser.readline().decode().strip()
+                    print(f"Arduino response: {response}")
+                    if response == "THETA_RESET":
+                        print("Theta successfully reset.")
+                        break
+            time.sleep(0.5)
+
+def stop_actions():
+    """Stop all current pattern execution."""
+    global pause_requested, stop_requested, current_playing_index, current_playlist, is_clearing, current_playing_file, execution_progress
+    with pause_condition:
+        pause_requested = False
+        stop_requested = True
+        current_playing_index = None
+        current_playlist = None
+        is_clearing = False
+        current_playing_file = None
+        execution_progress = None
+
+def get_status():
+    """Get the current execution status."""
+    global is_clearing
+    # Update is_clearing based on current file
+    if current_playing_file in CLEAR_PATTERNS.values():
+        is_clearing = True
+    else:
+        is_clearing = False
+
+    return {
+        "ser_port": serial_manager.get_port(),
+        "stop_requested": stop_requested,
+        "pause_requested": pause_requested,
+        "current_playing_file": current_playing_file,
+        "execution_progress": execution_progress,
+        "current_playing_index": current_playing_index,
+        "current_playlist": current_playlist,
+        "is_clearing": is_clearing
+    }

dune_weaver_flask/modules/core/playlist_manager.py

@@ -1,99 +1,95 @@
 import json
 import os
 import threading
-from dune_weaver_flask.modules.core.pattern_manager import pattern_manager
+from dune_weaver_flask.modules.core import pattern_manager
 
-class PlaylistManager:
-    def __init__(self):
-        self.PLAYLISTS_FILE = os.path.join(os.getcwd(), "playlists.json")
-        
-        # Ensure the file exists and contains at least an empty JSON object
-        if not os.path.exists(self.PLAYLISTS_FILE):
-            with open(self.PLAYLISTS_FILE, "w") as f:
-                json.dump({}, f, indent=2)
+# Global state
+PLAYLISTS_FILE = os.path.join(os.getcwd(), "playlists.json")
 
-    def load_playlists(self):
-        """Load the entire playlists dictionary from the JSON file."""
-        with open(self.PLAYLISTS_FILE, "r") as f:
-            return json.load(f)
+# Ensure the file exists and contains at least an empty JSON object
+if not os.path.exists(PLAYLISTS_FILE):
+    with open(PLAYLISTS_FILE, "w") as f:
+        json.dump({}, f, indent=2)
 
-    def save_playlists(self, playlists_dict):
-        """Save the entire playlists dictionary back to the JSON file."""
-        with open(self.PLAYLISTS_FILE, "w") as f:
-            json.dump(playlists_dict, f, indent=2)
+def load_playlists():
+    """Load the entire playlists dictionary from the JSON file."""
+    with open(PLAYLISTS_FILE, "r") as f:
+        return json.load(f)
 
-    def list_all_playlists(self):
-        """Returns a list of all playlist names."""
-        playlists_dict = self.load_playlists()
-        return list(playlists_dict.keys())
+def save_playlists(playlists_dict):
+    """Save the entire playlists dictionary back to the JSON file."""
+    with open(PLAYLISTS_FILE, "w") as f:
+        json.dump(playlists_dict, f, indent=2)
 
-    def get_playlist(self, playlist_name):
-        """Get a specific playlist by name."""
-        playlists_dict = self.load_playlists()
-        if playlist_name not in playlists_dict:
-            return None
-        return {
-            "name": playlist_name,
-            "files": playlists_dict[playlist_name]
-        }
+def list_all_playlists():
+    """Returns a list of all playlist names."""
+    playlists_dict = load_playlists()
+    return list(playlists_dict.keys())
 
-    def create_playlist(self, playlist_name, files):
-        """Create or update a playlist."""
-        playlists_dict = self.load_playlists()
-        playlists_dict[playlist_name] = files
-        self.save_playlists(playlists_dict)
-        return True
+def get_playlist(playlist_name):
+    """Get a specific playlist by name."""
+    playlists_dict = load_playlists()
+    if playlist_name not in playlists_dict:
+        return None
+    return {
+        "name": playlist_name,
+        "files": playlists_dict[playlist_name]
+    }
 
-    def modify_playlist(self, playlist_name, files):
-        """Modify an existing playlist."""
-        return self.create_playlist(playlist_name, files)
+def create_playlist(playlist_name, files):
+    """Create or update a playlist."""
+    playlists_dict = load_playlists()
+    playlists_dict[playlist_name] = files
+    save_playlists(playlists_dict)
+    return True
 
-    def delete_playlist(self, playlist_name):
-        """Delete a playlist."""
-        playlists_dict = self.load_playlists()
-        if playlist_name not in playlists_dict:
-            return False
-        del playlists_dict[playlist_name]
-        self.save_playlists(playlists_dict)
-        return True
+def modify_playlist(playlist_name, files):
+    """Modify an existing playlist."""
+    return create_playlist(playlist_name, files)
 
-    def add_to_playlist(self, playlist_name, pattern):
-        """Add a pattern to an existing playlist."""
-        playlists_dict = self.load_playlists()
-        if playlist_name not in playlists_dict:
-            return False
-        playlists_dict[playlist_name].append(pattern)
-        self.save_playlists(playlists_dict)
-        return True
+def delete_playlist(playlist_name):
+    """Delete a playlist."""
+    playlists_dict = load_playlists()
+    if playlist_name not in playlists_dict:
+        return False
+    del playlists_dict[playlist_name]
+    save_playlists(playlists_dict)
+    return True
 
-    def run_playlist(self, playlist_name, pause_time=0, clear_pattern=None, run_mode="single", shuffle=False, schedule_hours=None):
-        """Run a playlist with the given options."""
-        playlists = self.load_playlists()
-        if playlist_name not in playlists:
-            return False, "Playlist not found"
+def add_to_playlist(playlist_name, pattern):
+    """Add a pattern to an existing playlist."""
+    playlists_dict = load_playlists()
+    if playlist_name not in playlists_dict:
+        return False
+    playlists_dict[playlist_name].append(pattern)
+    save_playlists(playlists_dict)
+    return True
 
-        file_paths = playlists[playlist_name]
-        file_paths = [os.path.join(pattern_manager.THETA_RHO_DIR, file) for file in file_paths]
+def run_playlist(playlist_name, pause_time=0, clear_pattern=None, run_mode="single", shuffle=False, schedule_hours=None):
+    """Run a playlist with the given options."""
+    playlists = load_playlists()
+    if playlist_name not in playlists:
+        return False, "Playlist not found"
 
-        if not file_paths:
-            return False, "Playlist is empty"
+    file_paths = playlists[playlist_name]
+    file_paths = [os.path.join(pattern_manager.THETA_RHO_DIR, file) for file in file_paths]
 
-        try:
-            threading.Thread(
-                target=pattern_manager.run_theta_rho_files,
-                args=(file_paths,),
-                kwargs={
-                    'pause_time': pause_time,
-                    'clear_pattern': clear_pattern,
-                    'run_mode': run_mode,
-                    'shuffle': shuffle,
-                    'schedule_hours': schedule_hours
-                },
-                daemon=True
-            ).start()
-            return True, f"Playlist '{playlist_name}' is now running."
-        except Exception as e:
-            return False, str(e)
+    if not file_paths:
+        return False, "Playlist is empty"
 
-# Create a global instance
-playlist_manager = PlaylistManager()
+    try:
+        threading.Thread(
+            target=pattern_manager.run_theta_rho_files,
+            args=(file_paths,),
+            kwargs={
+                'pause_time': pause_time,
+                'clear_pattern': clear_pattern,
+                'run_mode': run_mode,
+                'shuffle': shuffle,
+                'schedule_hours': schedule_hours
+            },
+            daemon=True
+        ).start()
+        return True, f"Playlist '{playlist_name}' is now running."
+    except Exception as e:
+        return False, str(e)

dune_weaver_flask/modules/firmware/firmware_manager.py

@@ -1,226 +1,222 @@
 import os
 import subprocess
-from ..serial.serial_manager import serial_manager
-
-class FirmwareManager:
-    def __init__(self):
-        self.MOTOR_TYPE_MAPPING = {
-            "TMC2209": "./firmware/arduino_code_TMC2209/arduino_code_TMC2209.ino",
-            "DRV8825": "./firmware/arduino_code/arduino_code.ino",
-            "esp32": "./firmware/esp32/esp32.ino",
-            "esp32_TMC2209": "./firmware/esp32_TMC2209/esp32_TMC2209.ino"
+from ..serial import serial_manager
+
+# Global state
+MOTOR_TYPE_MAPPING = {
+    "TMC2209": "./firmware/arduino_code_TMC2209/arduino_code_TMC2209.ino",
+    "DRV8825": "./firmware/arduino_code/arduino_code.ino",
+    "esp32": "./firmware/esp32/esp32.ino",
+    "esp32_TMC2209": "./firmware/esp32_TMC2209/esp32_TMC2209.ino"
+}
+
+def get_ino_firmware_details(ino_file_path):
+    """Extract firmware details from the given .ino file."""
+    try:
+        if not ino_file_path:
+            raise ValueError("Invalid path: ino_file_path is None or empty.")
+
+        firmware_details = {"version": None, "motorType": None}
+
+        with open(ino_file_path, "r") as file:
+            for line in file:
+                if "firmwareVersion" in line:
+                    start = line.find('"') + 1
+                    end = line.rfind('"')
+                    if start != -1 and end != -1 and start < end:
+                        firmware_details["version"] = line[start:end]
+
+                if "motorType" in line:
+                    start = line.find('"') + 1
+                    end = line.rfind('"')
+                    if start != -1 and end != -1 and start < end:
+                        firmware_details["motorType"] = line[start:end]
+
+        if not firmware_details["version"]:
+            print(f"Firmware version not found in file: {ino_file_path}")
+        if not firmware_details["motorType"]:
+            print(f"Motor type not found in file: {ino_file_path}")
+
+        return firmware_details if any(firmware_details.values()) else None
+
+    except FileNotFoundError:
+        print(f"File not found: {ino_file_path}")
+        return None
+    except Exception as e:
+        print(f"Error reading .ino file: {str(e)}")
+        return None
+
+def check_git_updates():
+    """Check for available Git updates."""
+    try:
+        subprocess.run(["git", "fetch", "--tags", "--force"], check=True)
+        latest_remote_tag = subprocess.check_output(
+            ["git", "describe", "--tags", "--abbrev=0", "origin/main"]
+        ).strip().decode()
+        latest_local_tag = subprocess.check_output(
+            ["git", "describe", "--tags", "--abbrev=0"]
+        ).strip().decode()
+
+        tag_behind_count = 0
+        if latest_local_tag != latest_remote_tag:
+            tags = subprocess.check_output(
+                ["git", "tag", "--merged", "origin/main"], text=True
+            ).splitlines()
+
+            found_local = False
+            for tag in tags:
+                if tag == latest_local_tag:
+                    found_local = True
+                elif found_local:
+                    tag_behind_count += 1
+                    if tag == latest_remote_tag:
+                        break
+
+        updates_available = latest_remote_tag != latest_local_tag
+
+        return {
+            "updates_available": updates_available,
+            "tag_behind_count": tag_behind_count,
+            "latest_remote_tag": latest_remote_tag,
+            "latest_local_tag": latest_local_tag,
+        }
+    except subprocess.CalledProcessError as e:
+        print(f"Error checking Git updates: {e}")
+        return {
+            "updates_available": False,
+            "tag_behind_count": 0,
+            "latest_remote_tag": None,
+            "latest_local_tag": None,
         }
 
-    def get_ino_firmware_details(self, ino_file_path):
-        """Extract firmware details from the given .ino file."""
-        try:
-            if not ino_file_path:
-                raise ValueError("Invalid path: ino_file_path is None or empty.")
-
-            firmware_details = {"version": None, "motorType": None}
-
-            with open(ino_file_path, "r") as file:
-                for line in file:
-                    if "firmwareVersion" in line:
-                        start = line.find('"') + 1
-                        end = line.rfind('"')
-                        if start != -1 and end != -1 and start < end:
-                            firmware_details["version"] = line[start:end]
-
-                    if "motorType" in line:
-                        start = line.find('"') + 1
-                        end = line.rfind('"')
-                        if start != -1 and end != -1 and start < end:
-                            firmware_details["motorType"] = line[start:end]
-
-            if not firmware_details["version"]:
-                print(f"Firmware version not found in file: {ino_file_path}")
-            if not firmware_details["motorType"]:
-                print(f"Motor type not found in file: {ino_file_path}")
-
-            return firmware_details if any(firmware_details.values()) else None
-
-        except FileNotFoundError:
-            print(f"File not found: {ino_file_path}")
-            return None
-        except Exception as e:
-            print(f"Error reading .ino file: {str(e)}")
-            return None
-
-    def check_git_updates(self):
-        """Check for available Git updates."""
-        try:
-            subprocess.run(["git", "fetch", "--tags", "--force"], check=True)
-            latest_remote_tag = subprocess.check_output(
-                ["git", "describe", "--tags", "--abbrev=0", "origin/main"]
-            ).strip().decode()
-            latest_local_tag = subprocess.check_output(
-                ["git", "describe", "--tags", "--abbrev=0"]
-            ).strip().decode()
-
-            tag_behind_count = 0
-            if latest_local_tag != latest_remote_tag:
-                tags = subprocess.check_output(
-                    ["git", "tag", "--merged", "origin/main"], text=True
-                ).splitlines()
-
-                found_local = False
-                for tag in tags:
-                    if tag == latest_local_tag:
-                        found_local = True
-                    elif found_local:
-                        tag_behind_count += 1
-                        if tag == latest_remote_tag:
-                            break
-
-            updates_available = latest_remote_tag != latest_local_tag
-
-            return {
-                "updates_available": updates_available,
-                "tag_behind_count": tag_behind_count,
-                "latest_remote_tag": latest_remote_tag,
-                "latest_local_tag": latest_local_tag,
-            }
-        except subprocess.CalledProcessError as e:
-            print(f"Error checking Git updates: {e}")
-            return {
-                "updates_available": False,
-                "tag_behind_count": 0,
-                "latest_remote_tag": None,
-                "latest_local_tag": None,
-            }
-
-    def update_software(self):
-        """Update the software to the latest version."""
-        error_log = []
-
-        def run_command(command, error_message):
-            try:
-                subprocess.run(command, check=True)
-            except subprocess.CalledProcessError as e:
-                print(f"{error_message}: {e}")
-                error_log.append(error_message)
+def update_software():
+    """Update the software to the latest version."""
+    error_log = []
 
+    def run_command(command, error_message):
         try:
-            subprocess.run(["git", "fetch", "--tags"], check=True)
-            latest_remote_tag = subprocess.check_output(
-                ["git", "describe", "--tags", "--abbrev=0", "origin/main"]
-            ).strip().decode()
+            subprocess.run(command, check=True)
         except subprocess.CalledProcessError as e:
-            error_log.append(f"Failed to fetch tags or get latest remote tag: {e}")
-            return False, "Failed to fetch tags or determine the latest version.", error_log
-
-        run_command(["git", "checkout", latest_remote_tag, '--force'], f"Failed to checkout version {latest_remote_tag}")
-        run_command(["docker", "compose", "pull"], "Failed to fetch Docker containers")
-        run_command(["docker", "compose", "up", "-d"], "Failed to restart Docker containers")
-
-        update_status = self.check_git_updates()
-
-        if (
-            update_status["updates_available"] is False
-            and update_status["latest_local_tag"] == update_status["latest_remote_tag"]
-        ):
-            return True, None, None
-        else:
-            return False, "Update incomplete", error_log
-
-    def get_firmware_info(self, motor_type=None):
-        """Get firmware information for the current or specified motor type."""
-        if motor_type and motor_type not in self.MOTOR_TYPE_MAPPING:
-            return False, "Invalid motor type"
-
-        installed_version = serial_manager.firmware_version
-        installed_type = serial_manager.arduino_driver_type
-
-        if motor_type:
-            # POST request with specified motor type
-            ino_path = self.MOTOR_TYPE_MAPPING[motor_type]
-            firmware_details = self.get_ino_firmware_details(ino_path)
+            print(f"{error_message}: {e}")
+            error_log.append(error_message)
+
+    try:
+        subprocess.run(["git", "fetch", "--tags"], check=True)
+        latest_remote_tag = subprocess.check_output(
+            ["git", "describe", "--tags", "--abbrev=0", "origin/main"]
+        ).strip().decode()
+    except subprocess.CalledProcessError as e:
+        error_log.append(f"Failed to fetch tags or get latest remote tag: {e}")
+        return False, "Failed to fetch tags or determine the latest version.", error_log
+
+    run_command(["git", "checkout", latest_remote_tag, '--force'], f"Failed to checkout version {latest_remote_tag}")
+    run_command(["docker", "compose", "pull"], "Failed to fetch Docker containers")
+    run_command(["docker", "compose", "up", "-d"], "Failed to restart Docker containers")
+
+    update_status = check_git_updates()
+
+    if (
+        update_status["updates_available"] is False
+        and update_status["latest_local_tag"] == update_status["latest_remote_tag"]
+    ):
+        return True, None, None
+    else:
+        return False, "Update incomplete", error_log
+
+def get_firmware_info(motor_type=None):
+    """Get firmware information for the current or specified motor type."""
+    if motor_type and motor_type not in MOTOR_TYPE_MAPPING:
+        return False, "Invalid motor type"
+
+    installed_version = serial_manager.firmware_version
+    installed_type = serial_manager.arduino_driver_type
+
+    if motor_type:
+        # POST request with specified motor type
+        ino_path = MOTOR_TYPE_MAPPING[motor_type]
+        firmware_details = get_ino_firmware_details(ino_path)
+
+        if not firmware_details:
+            return False, "Failed to retrieve .ino firmware details"
+
+        return True, {
+            "installedVersion": 'Unknown',
+            "installedType": motor_type,
+            "inoVersion": firmware_details["version"],
+            "inoType": firmware_details["motorType"],
+            "updateAvailable": True
+        }
+    else:
+        # GET request for current firmware info
+        if installed_version != 'Unknown' and installed_type != 'Unknown':
+            ino_path = MOTOR_TYPE_MAPPING.get(installed_type)
+            firmware_details = get_ino_firmware_details(ino_path)
 
-            if not firmware_details:
+            if not firmware_details or not firmware_details.get("version") or not firmware_details.get("motorType"):
                 return False, "Failed to retrieve .ino firmware details"
 
-            return True, {
-                "installedVersion": 'Unknown',
-                "installedType": motor_type,
-                "inoVersion": firmware_details["version"],
-                "inoType": firmware_details["motorType"],
-                "updateAvailable": True
-            }
-        else:
-            # GET request for current firmware info
-            if installed_version != 'Unknown' and installed_type != 'Unknown':
-                ino_path = self.MOTOR_TYPE_MAPPING.get(installed_type)
-                firmware_details = self.get_ino_firmware_details(ino_path)
-
-                if not firmware_details or not firmware_details.get("version") or not firmware_details.get("motorType"):
-                    return False, "Failed to retrieve .ino firmware details"
-
-                update_available = (
-                    installed_version != firmware_details["version"] or
-                    installed_type != firmware_details["motorType"]
-                )
-
-                return True, {
-                    "installedVersion": installed_version,
-                    "installedType": installed_type,
-                    "inoVersion": firmware_details["version"],
-                    "inoType": firmware_details["motorType"],
-                    "updateAvailable": update_available
-                }
+            update_available = (
+                installed_version != firmware_details["version"] or
+                installed_type != firmware_details["motorType"]
+            )
 
             return True, {
                 "installedVersion": installed_version,
                 "installedType": installed_type,
-                "updateAvailable": False
+                "inoVersion": firmware_details["version"],
+                "inoType": firmware_details["motorType"],
+                "updateAvailable": update_available
             }
 
-    def flash_firmware(self, motor_type):
-        """Flash firmware for the specified motor type."""
-        if not motor_type or motor_type not in self.MOTOR_TYPE_MAPPING:
-            return False, "Invalid or missing motor type"
-
-        if not serial_manager.is_connected():
-            return False, "No device connected or connection lost"
+        return True, {
+            "installedVersion": installed_version,
+            "installedType": installed_type,
+            "updateAvailable": False
+        }
 
-        try:
-            ino_file_path = self.MOTOR_TYPE_MAPPING[motor_type]
-            hex_file_path = f"{ino_file_path}.hex"
-            bin_file_path = f"{ino_file_path}.bin"
-
-            if motor_type.lower() in ["esp32", "esp32_tmc2209"]:
-                if not os.path.exists(bin_file_path):
-                    return False, f"Firmware binary not found: {bin_file_path}"
-
-                flash_command = [
-                    "esptool.py",
-                    "--chip", "esp32",
-                    "--port", serial_manager.get_port(),
-                    "--baud", "115200",
-                    "write_flash", "-z", "0x1000", bin_file_path
-                ]
-            else:
-                if not os.path.exists(hex_file_path):
-                    return False, f"Hex file not found: {hex_file_path}"
-
-                flash_command = [
-                    "avrdude",
-                    "-v",
-                    "-c", "arduino",
-                    "-p", "atmega328p",
-                    "-P", serial_manager.get_port(),
-                    "-b", "115200",
-                    "-D",
-                    "-U", f"flash:w:{hex_file_path}:i"
-                ]
-
-            flash_process = subprocess.run(flash_command, capture_output=True, text=True)
-            if flash_process.returncode != 0:
-                return False, flash_process.stderr
-
-            return True, "Firmware flashed successfully"
-        except Exception as e:
-            return False, str(e)
-
-# Create a global instance
-firmware_manager = FirmwareManager()
+def flash_firmware(motor_type):
+    """Flash firmware for the specified motor type."""
+    if not motor_type or motor_type not in MOTOR_TYPE_MAPPING:
+        return False, "Invalid or missing motor type"
+
+    if not serial_manager.is_connected():
+        return False, "No device connected or connection lost"
+
+    try:
+        ino_file_path = MOTOR_TYPE_MAPPING[motor_type]
+        hex_file_path = f"{ino_file_path}.hex"
+        bin_file_path = f"{ino_file_path}.bin"
+
+        if motor_type.lower() in ["esp32", "esp32_tmc2209"]:
+            if not os.path.exists(bin_file_path):
+                return False, f"Firmware binary not found: {bin_file_path}"
+
+            flash_command = [
+                "esptool.py",
+                "--chip", "esp32",
+                "--port", serial_manager.get_port(),
+                "--baud", "115200",
+                "write_flash", "-z", "0x1000", bin_file_path
+            ]
+        else:
+            if not os.path.exists(hex_file_path):
+                return False, f"Hex file not found: {hex_file_path}"
+
+            flash_command = [
+                "avrdude",
+                "-v",
+                "-c", "arduino",
+                "-p", "atmega328p",
+                "-P", serial_manager.get_port(),
+                "-b", "115200",
+                "-D",
+                "-U", f"flash:w:{hex_file_path}:i"
+            ]
+
+        flash_process = subprocess.run(flash_command, capture_output=True, text=True)
+        if flash_process.returncode != 0:
+            return False, flash_process.stderr
+
+        return True, "Firmware flashed successfully"
+    except Exception as e:
+        return False, str(e)

dune_weaver_flask/modules/serial/serial_manager.py

@@ -3,107 +3,105 @@ import serial.tools.list_ports
 import threading
 import time
 
-class SerialManager:
-    def __init__(self):
-        self.ser = None
-        self.ser_port = None
-        self.serial_lock = threading.RLock()
-        self.IGNORE_PORTS = ['/dev/cu.debug-console', '/dev/cu.Bluetooth-Incoming-Port']
-        
-        # Device information
-        self.arduino_table_name = None
-        self.arduino_driver_type = 'Unknown'
-        self.firmware_version = 'Unknown'
-
-    def list_serial_ports(self):
-        """Return a list of available serial ports."""
-        ports = serial.tools.list_ports.comports()
-        return [port.device for port in ports if port.device not in self.IGNORE_PORTS]
-
-    def connect_to_serial(self, port=None, baudrate=115200):
-        """Automatically connect to the first available serial port or a specified port."""
-        try:
-            if port is None:
-                ports = self.list_serial_ports()
-                if not ports:
-                    print("No serial port connected")
-                    return False
-                port = ports[0]  # Auto-select the first available port
-
-            with self.serial_lock:
-                if self.ser and self.ser.is_open:
-                    self.ser.close()
-                self.ser = serial.Serial(port, baudrate, timeout=2)
-                self.ser_port = port
-
-            print(f"Connected to serial port: {port}")
-            time.sleep(2)  # Allow time for the connection to establish
-
-            # Read initial startup messages from Arduino
-            while self.ser.in_waiting > 0:
-                line = self.ser.readline().decode().strip()
-                print(f"Arduino: {line}")
-
-                # Store the device details based on the expected messages
-                if "Table:" in line:
-                    self.arduino_table_name = line.replace("Table: ", "").strip()
-                elif "Drivers:" in line:
-                    self.arduino_driver_type = line.replace("Drivers: ", "").strip()
-                elif "Version:" in line:
-                    self.firmware_version = line.replace("Version: ", "").strip()
-
-            print(f"Detected Table: {self.arduino_table_name or 'Unknown'}")
-            print(f"Detected Drivers: {self.arduino_driver_type or 'Unknown'}")
-
-            return True
-        except serial.SerialException as e:
-            print(f"Failed to connect to serial port {port}: {e}")
-            self.ser_port = None
-
-        print("Max retries reached. Could not connect to a serial port.")
-        return False
-
-    def disconnect_serial(self):
-        """Disconnect the current serial connection."""
-        if self.ser and self.ser.is_open:
-            self.ser.close()
-            self.ser = None
-        self.ser_port = None
-
-    def restart_serial(self, port, baudrate=115200):
-        """Restart the serial connection."""
-        self.disconnect_serial()
-        return self.connect_to_serial(port, baudrate)
-
-    def send_coordinate_batch(self, coordinates):
-        """Send a batch of theta-rho pairs to the Arduino."""
-        batch_str = ";".join(f"{theta:.5f},{rho:.5f}" for theta, rho in coordinates) + ";\n"
-        with self.serial_lock:
-            self.ser.write(batch_str.encode())
-
-    def send_command(self, command):
-        """Send a single command to the Arduino."""
-        with self.serial_lock:
-            self.ser.write(f"{command}\n".encode())
-            print(f"Sent: {command}")
-
-            # Wait for "R" acknowledgment from Arduino
-            while True:
-                with self.serial_lock:
-                    if self.ser.in_waiting > 0:
-                        response = self.ser.readline().decode().strip()
-                        print(f"Arduino response: {response}")
-                        if response == "R":
-                            print("Command execution completed.")
-                            break
-
-    def is_connected(self):
-        """Check if serial connection is established and open."""
-        return self.ser is not None and self.ser.is_open
-
-    def get_port(self):
-        """Get the current serial port."""
-        return self.ser_port
-
-# Create a global instance
-serial_manager = SerialManager()
+# Global state
+ser = None
+ser_port = None
+serial_lock = threading.RLock()
+IGNORE_PORTS = ['/dev/cu.debug-console', '/dev/cu.Bluetooth-Incoming-Port']
+
+# Device information
+arduino_table_name = None
+arduino_driver_type = 'Unknown'
+firmware_version = 'Unknown'
+
+def list_serial_ports():
+    """Return a list of available serial ports."""
+    ports = serial.tools.list_ports.comports()
+    return [port.device for port in ports if port.device not in IGNORE_PORTS]
+
+def connect_to_serial(port=None, baudrate=115200):
+    """Automatically connect to the first available serial port or a specified port."""
+    global ser, ser_port, arduino_table_name, arduino_driver_type, firmware_version
+    try:
+        if port is None:
+            ports = list_serial_ports()
+            if not ports:
+                print("No serial port connected")
+                return False
+            port = ports[0]  # Auto-select the first available port
+
+        with serial_lock:
+            if ser and ser.is_open:
+                ser.close()
+            ser = serial.Serial(port, baudrate, timeout=2)
+            ser_port = port
+
+        print(f"Connected to serial port: {port}")
+        time.sleep(2)  # Allow time for the connection to establish
+
+        # Read initial startup messages from Arduino
+        while ser.in_waiting > 0:
+            line = ser.readline().decode().strip()
+            print(f"Arduino: {line}")
+
+            # Store the device details based on the expected messages
+            if "Table:" in line:
+                arduino_table_name = line.replace("Table: ", "").strip()
+            elif "Drivers:" in line:
+                arduino_driver_type = line.replace("Drivers: ", "").strip()
+            elif "Version:" in line:
+                firmware_version = line.replace("Version: ", "").strip()
+
+        print(f"Detected Table: {arduino_table_name or 'Unknown'}")
+        print(f"Detected Drivers: {arduino_driver_type or 'Unknown'}")
+
+        return True
+    except serial.SerialException as e:
+        print(f"Failed to connect to serial port {port}: {e}")
+        ser_port = None
+
+    print("Max retries reached. Could not connect to a serial port.")
+    return False
+
+def disconnect_serial():
+    """Disconnect the current serial connection."""
+    global ser, ser_port
+    if ser and ser.is_open:
+        ser.close()
+        ser = None
+    ser_port = None
+
+def restart_serial(port, baudrate=115200):
+    """Restart the serial connection."""
+    disconnect_serial()
+    return connect_to_serial(port, baudrate)
+
+def send_coordinate_batch(coordinates):
+    """Send a batch of theta-rho pairs to the Arduino."""
+    batch_str = ";".join(f"{theta:.5f},{rho:.5f}" for theta, rho in coordinates) + ";\n"
+    with serial_lock:
+        ser.write(batch_str.encode())
+
+def send_command(command):
+    """Send a single command to the Arduino."""
+    with serial_lock:
+        ser.write(f"{command}\n".encode())
+        print(f"Sent: {command}")
+
+        # Wait for "R" acknowledgment from Arduino
+        while True:
+            with serial_lock:
+                if ser.in_waiting > 0:
+                    response = ser.readline().decode().strip()
+                    print(f"Arduino response: {response}")
+                    if response == "R":
+                        print("Command execution completed.")
+                        break
+
+def is_connected():
+    """Check if serial connection is established and open."""
+    return ser is not None and ser.is_open
+
+def get_port():
+    """Get the current serial port."""
+    return ser_port