Reducing the amount of nested try/exceptions

Added (module-level helpers):
_FAILED sentinel + safe_call() - one try/except that serves all call sites
_playing_state() context manager - handles flag lifecycle via with blocks
_cleanup_audio_monitoring() - extracted from inline try/except

Refactored (try/excepts eliminated or simplified):

run() startup from 2 nested try/excepts to safe_call() + conditional
Audio monitoring cleanup from 15-line try/except block to safe_call(self._cleanup_audio_monitoring, ...)
_stop_audio_input() from try/except wrapping join+stop to 2 safe_call() lines
_play_test_movement fallback from Nested try/except in except to safe_call(..., label="fallback movement")
_play_emotion from Manual flag set + try/except/finally to with self._playing_state('_emotion_playing') + safe_call for fallback
_go_to_sleep_wrapper from a 7-line try/except to safe_call() + conditional
_disable_motors_wrapper from 7-line try/except to safe_call() + conditional
_park_robot from 6-line try/except to a Single safe_call() line
_test_antenna_movement from try/except wrapping entire body to Clean body, safe_call at thread call site
stop_movement route from Unnecessary try/except around Event.set() to Removed entirely
File number parsing from try/except with continue to with suppress(IndexError, ValueError)

Result: 27 try blocks to 21 total (19 in app code, 2 in helpers).
wrapped in a helper function making it more readable.

wake_me_up/main.py

@@ -1,6 +1,7 @@
 import threading
 import time
 import json
+from contextlib import contextmanager, suppress
 from pathlib import Path
 from datetime import datetime
 
@@ -14,6 +15,19 @@ from reachy_mini import ReachyMini, ReachyMiniApp
 from reachy_mini.utils import create_head_pose
 from reachy_mini.motion.recorded_move import RecordedMove, RecordedMoves
 
+# Sentinel for safe_call failure detection
+_FAILED = object()
+
+
+def safe_call(fn, *args, label="operation", **kwargs):
+    """Call fn, log and return _FAILED on failure."""
+    try:
+        return fn(*args, **kwargs)
+    except Exception as e:
+        print(f"Failed to {label}: {e}")
+        return _FAILED
+
+
 RECORDINGS_DIR = Path(__file__).parent / "recordings"
 RECORDINGS_DIR.mkdir(parents=True, exist_ok=True)
 
@@ -99,30 +113,55 @@ class WakeMeUp(ReachyMiniApp):
 
         self._register_routes()
 
+    @contextmanager
+    def _playing_state(self, *flags):
+        """Set state flags True on entry, False on exit."""
+        with self._state_lock:
+            for flag in flags:
+                setattr(self, flag, True)
+        try:
+            yield
+        finally:
+            with self._state_lock:
+                for flag in flags:
+                    setattr(self, flag, False)
+
+    def _cleanup_audio_monitoring(self, reachy_mini):
+        """Flush buffer and stop recording after monitoring ends."""
+        time.sleep(0.2)
+        flushed = 0
+        for _ in range(100):
+            sample = reachy_mini.media.get_audio_sample()
+            if sample is None:
+                break
+            flushed += 1
+        reachy_mini.media.stop_recording()
+        print(f"DEBUG: Monitoring thread cleanup complete, flushed {flushed} samples")
+
     def run(self, reachy_mini: ReachyMini, stop_event: threading.Event):
         self._reachy_mini = reachy_mini
 
         try:
             # Pre-load emotions library (so no wait during diagnostic)
             print("Pre-loading emotions library...")
-            try:
-                self._emotions = RecordedMoves("pollen-robotics/reachy-mini-emotions-library")
+            self._emotions = safe_call(
+                RecordedMoves, "pollen-robotics/reachy-mini-emotions-library",
+                label="pre-load emotions",
+            )
+            if self._emotions is not _FAILED:
                 print("Emotions loaded successfully")
-            except Exception as e:
-                print(f"Failed to pre-load emotions: {e}")
+            else:
+                self._emotions = None
 
             # Put robot in sleeping pose at startup
             print("Setting sleeping pose...")
-            try:
-                reachy_mini.goto_sleep()
+            if safe_call(reachy_mini.goto_sleep, label="set sleeping pose") is not _FAILED:
                 with self._state_lock:
                     self._is_sleeping = True
                     # Auto-advance to step 1 when sleeping
                     if self._current_step == 0:
                         self._current_step = 1
                 print("Sleeping pose set, advancing to step 1")
-            except Exception as e:
-                print(f"Failed to set sleeping pose: {e}")
 
             # Note: Audio monitoring will start when entering step 3
 
@@ -277,24 +316,7 @@ class WakeMeUp(ReachyMiniApp):
                     time.sleep(0.1)
 
             print(f"DEBUG: Audio monitoring loop ended, performing cleanup...")
-
-            # Cleanup: flush buffer and stop recording
-            try:
-                time.sleep(0.2)  # Wait for any pending samples
-
-                # Flush any remaining samples from buffer
-                flushed = 0
-                for _ in range(100):
-                    sample = reachy_mini.media.get_audio_sample()
-                    if sample is None:
-                        break
-                    flushed += 1
-
-                # Stop recording
-                reachy_mini.media.stop_recording()
-                print(f"DEBUG: Monitoring thread cleanup complete, flushed {flushed} samples")
-            except Exception as e:
-                print(f"ERROR during monitoring cleanup: {e}")
+            safe_call(self._cleanup_audio_monitoring, reachy_mini, label="monitoring cleanup")
 
         # Start monitoring in background thread
         import threading
@@ -305,14 +327,10 @@ class WakeMeUp(ReachyMiniApp):
     def _stop_audio_input(self):
         self._audio_monitoring_active = False
         if hasattr(self, '_audio_thread') and self._audio_thread:
-            try:
-                self._audio_thread.join(timeout=1.0)
-                # Stop recording
-                if self._reachy_mini:
-                    self._reachy_mini.media.stop_recording()
-                print("Audio monitoring stopped")
-            except Exception as e:
-                print(f"Error stopping audio: {e}")
+            safe_call(self._audio_thread.join, timeout=1.0, label="join audio thread")
+            if self._reachy_mini:
+                safe_call(self._reachy_mini.media.stop_recording, label="stop recording")
+            print("Audio monitoring stopped")
 
     def _manage_step_transition(self, current_step: int, next_step: int) -> dict:
         """
@@ -617,12 +635,8 @@ class WakeMeUp(ReachyMiniApp):
 
         except Exception as e:
             print(f"Failed to play movement: {e}")
-            # Fallback
-            try:
-                if not self._cancel_movement.is_set():
-                    self._play_simple_movement(reachy_mini)
-            except Exception as e2:
-                print(f"Fallback movement also failed: {e2}")
+            if not self._cancel_movement.is_set():
+                safe_call(self._play_simple_movement, reachy_mini, label="fallback movement")
         finally:
             with self._state_lock:
                 self._movement_playing = False
@@ -731,85 +745,72 @@ class WakeMeUp(ReachyMiniApp):
         with self._state_lock:
             is_first = self._is_first_emotion
             is_sleeping = self._is_sleeping
-            self._emotion_playing = True
 
-        try:
-            # Check if robot is sleeping and wake it up if needed
-            if is_sleeping:
-                print(f"Robot is sleeping, waking up before playing {emotion_name}...")
+        with self._playing_state('_emotion_playing'):
+            try:
+                # Check if robot is sleeping and wake it up if needed
+                if is_sleeping:
+                    print(f"Robot is sleeping, waking up before playing {emotion_name}...")
+                    self._safe_enable_motors(reachy_mini)
+                    reachy_mini.goto_target(
+                        head=create_head_pose(pitch=0.0, yaw=0.0, roll=0.0),
+                        duration=1.0
+                    )
+                    with self._state_lock:
+                        self._is_sleeping = False
+                    time.sleep(0.3)
+                    print(f"Robot awake, playing {emotion_name}...")
+                elif is_first:
+                    print(f"First emotion play, starting {emotion_name} directly...")
+                else:
+                    print(f"Playing {emotion_name}...")
+
+                # Mark that first emotion has been played
+                with self._state_lock:
+                    self._is_first_emotion = False
+
+                # Use pre-loaded emotions (no wait time!)
+                if self._emotions is None:
+                    print("Emotions not loaded, loading now...")
+                    self._emotions = RecordedMoves("pollen-robotics/reachy-mini-emotions-library")
+
+                emotion_move = self._emotions.get(emotion_name)
+
                 self._safe_enable_motors(reachy_mini)
+                reachy_mini.play_move(emotion_move, initial_goto_duration=0.8, sound=True)
+
+                # Return to neutral position after emotion
+                print(f"{emotion_name} completed, returning to neutral position...")
                 reachy_mini.goto_target(
                     head=create_head_pose(pitch=0.0, yaw=0.0, roll=0.0),
-                    duration=1.0
+                    antennas=[0, 0],
+                    duration=0.8
                 )
                 with self._state_lock:
                     self._is_sleeping = False
-                time.sleep(0.3)
-                print(f"Robot awake, playing {emotion_name}...")
-            elif is_first:
-                print(f"First emotion play, starting {emotion_name} directly...")
-            else:
-                print(f"Playing {emotion_name}...")
-
-            # Mark that first emotion has been played
-            with self._state_lock:
-                self._is_first_emotion = False
-
-            # Use pre-loaded emotions (no wait time!)
-            if self._emotions is None:
-                print("Emotions not loaded, loading now...")
-                self._emotions = RecordedMoves("pollen-robotics/reachy-mini-emotions-library")
-
-            emotion_move = self._emotions.get(emotion_name)
-
-            self._safe_enable_motors(reachy_mini)
-            reachy_mini.play_move(emotion_move, initial_goto_duration=0.8, sound=True)
-
-            # Return to neutral position after emotion
-            print(f"{emotion_name} completed, returning to neutral position...")
-            reachy_mini.goto_target(
-                head=create_head_pose(pitch=0.0, yaw=0.0, roll=0.0),
-                antennas=[0, 0],
-                duration=0.8
-            )
-            with self._state_lock:
-                self._is_sleeping = False
-            print("Robot in neutral position (motors still enabled)")
+                print("Robot in neutral position (motors still enabled)")
 
-        except Exception as e:
-            print(f"Failed to play {emotion_name}: {e}")
-            # Fallback: just play sound (only for happy/success2)
-            if emotion_name == "success2":
-                try:
-                    reachy_mini.media.play_sound("wake_up.wav")
-                    print("Played fallback sound")
-                except Exception as e2:
-                    print(f"Failed to play fallback sound: {e2}")
-        finally:
-            with self._state_lock:
-                self._emotion_playing = False
+            except Exception as e:
+                print(f"Failed to play {emotion_name}: {e}")
+                if emotion_name == "success2":
+                    if safe_call(reachy_mini.media.play_sound, "wake_up.wav", label="fallback sound") is not _FAILED:
+                        print("Played fallback sound")
 
     def _go_to_sleep_wrapper(self, reachy_mini):
         """Put robot to sleep (for going back from step 4 to step 3)"""
-        try:
-            print("Going to sleep (back navigation)...")
-            reachy_mini.goto_sleep()
+        print("Going to sleep (back navigation)...")
+        if safe_call(reachy_mini.goto_sleep, label="go to sleep") is not _FAILED:
             with self._state_lock:
                 self._is_sleeping = True
             print("Robot is now sleeping")
-        except Exception as e:
-            print(f"Failed to go to sleep: {e}")
 
     def _disable_motors_wrapper(self, reachy_mini):
         """Disable motors for manual positioning (step 2)"""
-        try:
-            print("Disabling motors for manual positioning...")
-            reachy_mini.disable_motors()
+        print("Disabling motors for manual positioning...")
+        if safe_call(reachy_mini.disable_motors, label="disable motors") is not _FAILED:
             with self._state_lock:
                 self._is_sleeping = False
             print("Motors disabled")
-        except Exception as e:
-            print(f"Failed to disable motors: {e}")
 
     def _goto_current_pose(self, reachy_mini: ReachyMini, duration: float = 0.05) -> None:
         """Command robot to go to its current position (used before enabling motors)"""
@@ -829,51 +830,42 @@ class WakeMeUp(ReachyMiniApp):
 
     def _park_robot(self, reachy_mini: ReachyMini) -> None:
         """Ensure motors are enabled when exiting app (to prevent robot from falling)"""
-        try:
-            print("Parking robot before exit...")
-            # Enable motors safely (idempotent - safe to call even if already on)
-            self._safe_enable_motors(reachy_mini)
-            print("Motors enabled for safe exit")
-        except Exception as e:
-            print(f"Failed to enable motors on exit: {e}")
+        print("Parking robot before exit...")
+        safe_call(self._safe_enable_motors, reachy_mini, label="enable motors on exit")
 
     def _test_antenna_movement(self, reachy_mini, side):
         """Move one antenna to test if it's correctly connected"""
-        try:
-            print(f"Testing {side} antenna...")
-            self._safe_enable_motors(reachy_mini)
-
-            # First raise head to neutral (0 degrees pitch)
-            reachy_mini.goto_target(
-                head=create_head_pose(pitch=0.0, yaw=0.0, roll=0.0),
-                antennas=[0, 0],
-                duration=0.8
-            )
-            time.sleep(0.8)
-
-            # Move the specified antenna with a small shake (4 repetitions)
-            # Note: antenna indices are swapped - right is index 0, left is index 1
-            if side == "left":
-                # Left antenna (index 1) - fast shake 4 times
-                head=create_head_pose(pitch=0.0, yaw=0.0, roll=0.0),
-                for _ in range(4):
-                    reachy_mini.goto_target(antennas=np.radians([0, 10]), duration=0.2)
-                    reachy_mini.goto_target(antennas=np.radians([0, -10]), duration=0.2)
-                # Return to neutral
-                reachy_mini.goto_target(antennas=np.radians([0, 0]), duration=0.2)
-
-            elif side == "right":
-                # Right antenna (index 0) - fast shake 4 times
-                for _ in range(4):
-                    reachy_mini.goto_target(antennas=np.radians([10, 0]), duration=0.2)
-                    reachy_mini.goto_target(antennas=np.radians([-10, 0]), duration=0.2)
-                # Return to neutral
-                reachy_mini.goto_target(antennas=np.radians([0, 0]), duration=0.2)
-
-            print(f"{side} antenna test complete")
+        print(f"Testing {side} antenna...")
+        self._safe_enable_motors(reachy_mini)
 
-        except Exception as e:
-            print(f"Failed to test {side} antenna: {e}")
+        # First raise head to neutral (0 degrees pitch)
+        reachy_mini.goto_target(
+            head=create_head_pose(pitch=0.0, yaw=0.0, roll=0.0),
+            antennas=[0, 0],
+            duration=0.8
+        )
+        time.sleep(0.8)
+
+        # Move the specified antenna with a small shake (4 repetitions)
+        # Note: antenna indices are swapped - right is index 0, left is index 1
+        if side == "left":
+            # Left antenna (index 1) - fast shake 4 times
+            head=create_head_pose(pitch=0.0, yaw=0.0, roll=0.0),
+            for _ in range(4):
+                reachy_mini.goto_target(antennas=np.radians([0, 10]), duration=0.2)
+                reachy_mini.goto_target(antennas=np.radians([0, -10]), duration=0.2)
+            # Return to neutral
+            reachy_mini.goto_target(antennas=np.radians([0, 0]), duration=0.2)
+
+        elif side == "right":
+            # Right antenna (index 0) - fast shake 4 times
+            for _ in range(4):
+                reachy_mini.goto_target(antennas=np.radians([10, 0]), duration=0.2)
+                reachy_mini.goto_target(antennas=np.radians([-10, 0]), duration=0.2)
+            # Return to neutral
+            reachy_mini.goto_target(antennas=np.radians([0, 0]), duration=0.2)
+
+        print(f"{side} antenna test complete")
 
     def _detect_motor_swaps(self, motors_data):
         """Detect if motors have been swapped during assembly
@@ -1138,14 +1130,9 @@ class WakeMeUp(ReachyMiniApp):
             if self._reachy_mini is None:
                 return {"status": "error", "message": "Robot not initialized"}
 
-            try:
-                # Signal the movement thread to stop
-                self._cancel_movement.set()
-                print("Movement cancellation requested")
-                return {"status": "stopping"}
-            except Exception as e:
-                print(f"Failed to stop movement: {e}")
-                return {"status": "error", "message": str(e)}
+            self._cancel_movement.set()
+            print("Movement cancellation requested")
+            return {"status": "stopping"}
 
         @self.settings_app.post("/api/play_happy")
         def play_happy():
@@ -1283,17 +1270,11 @@ class WakeMeUp(ReachyMiniApp):
 
                     # Find next available number
                     existing_files = list(RECORDINGS_DIR.glob("record_*.wav"))
-                    if existing_files:
-                        numbers = []
-                        for f in existing_files:
-                            try:
-                                num = int(f.stem.split('_')[1])
-                                numbers.append(num)
-                            except (IndexError, ValueError):
-                                continue
-                        next_num = max(numbers) + 1 if numbers else 1
-                    else:
-                        next_num = 1
+                    numbers = []
+                    for f in existing_files:
+                        with suppress(IndexError, ValueError):
+                            numbers.append(int(f.stem.split('_')[1]))
+                    next_num = max(numbers) + 1 if numbers else 1
 
                     filename = f"record_{next_num}.wav"
                     filepath = RECORDINGS_DIR / filename
@@ -1376,8 +1357,10 @@ class WakeMeUp(ReachyMiniApp):
                 return {"status": "error", "message": "Invalid antenna side"}
 
             threading.Thread(
-                target=self._test_antenna_movement,
-                args=(self._reachy_mini, antenna_side),
+                target=lambda side=antenna_side: safe_call(
+                    self._test_antenna_movement, self._reachy_mini, side,
+                    label=f"test {side} antenna",
+                ),
                 daemon=True
             ).start()