Working input recorder

Author: danic85

config/input_recorder.yml

@@ -1,5 +1,5 @@
 input_recorder:
-  enabled: false
+  enabled: true
   path: modules.input_recorder.InputRecorder
   config:
     animations_dir: animations/recorded

modules/controller.py

@@ -58,30 +58,31 @@ def listen(self):
             except FileNotFoundError:
                 self.log(f"No gamepad found at {self.device}. Waiting for connection...", level='warning')
                 time.sleep(1)
-            except Exception as e:
-                self.log(f"Error: {e}", level='error')
-                time.sleep(1)
 
     def _handle_js_event(self, time_ms, value, type_, number):
         """Route joystick events to appropriate handlers."""
         is_init = (type_ & self.JS_EVENT_INIT) != 0
         event_type = type_ & ~self.JS_EVENT_INIT
 
-        topic = None
-        args = {}
-
+        topics_args = []
         if event_type == self.JS_EVENT_BUTTON:
-            topic, args = self._handle_button_event(value, number)
+            topics_args = self._handle_button_event(value, number, collect_topics_args=True)
         elif event_type == self.JS_EVENT_AXIS:
-            topic, args = self._handle_axis_event(value, number)
-
-        self.publish('controller/event', event=(time_ms, value, type_, number, topic, args))
-
-    def _handle_button_event(self, value, number):
+            topics_args = self._handle_axis_event(value, number, collect_topics_args=True)
+        self.publish('controller/event', event={
+            'time_ms': time_ms,
+            'value': value,
+            'type': type_,
+            'number': number,
+            'topics_args': topics_args
+        })
+
+    def _handle_button_event(self, value, number, collect_topics_args=False):
         """Handle button press/release events."""
         button = self.button_names.get(number, f'BTN_{number}')
         topic = None
         args = {}
+        topics_args = []
 
         if value == 1:
             self.pressed_buttons.add(button)
@@ -98,17 +99,29 @@ def _handle_button_event(self, value, number):
                 args = mapping.get('args', {})
                 self.publish(topic, **args)
                 self.log(f"Published to topic {topic} with args {args} (jsdev)")
+                topics_args.append({'topic': topic, 'args': args})
 
+        # If collect_topics_args is set, return the list, else preserve old behavior
+        import inspect
+        frame = inspect.currentframe().f_back
+        if frame and 'collect_topics_args' in frame.f_locals and frame.f_locals['collect_topics_args']:
+            return topics_args
         return topic, args
 
-    def _handle_axis_event(self, value, number):
+    def _handle_axis_event(self, value, number, collect_topics_args=False):
         """Handle axis movement events."""
         axis = self.axis_names.get(number, f'AXIS_{number}')
         topic = None
         args = {}
+        topics_args = []
 
         button_map = self._get_active_mapping()
         if axis not in button_map:
+            # If collecting, return empty list
+            import inspect
+            frame = inspect.currentframe().f_back
+            if frame and 'collect_topics_args' in frame.f_locals and frame.f_locals['collect_topics_args']:
+                return topics_args
             return topic, args
 
         for mapping in button_map[axis]:
@@ -141,7 +154,13 @@ def _handle_axis_event(self, value, number):
 
             self.publish(topic, **args)
             self.log(f"Published to topic {topic} with args {args} (jsdev)")
+            topics_args.append({'topic': topic, 'args': args})
 
+        # If collect_topics_args is set, return the list, else preserve old behavior
+        import inspect
+        frame = inspect.currentframe().f_back
+        if frame and 'collect_topics_args' in frame.f_locals and frame.f_locals['collect_topics_args']:
+            return topics_args
         return topic, args
 
     # Keep public method for backwards compatibility with tests

modules/input_recorder.py

@@ -13,10 +13,16 @@ def __init__(self, **kwargs):
         self.filename = None
 
     def setup_messaging(self):
-        self.subscribe('record/start', self.toggle_recording(enable=True))
-        self.subscribe('record/stop', lambda: self.toggle_recording(enable=False))
+        self.subscribe('record/start', self.start_recording )
+        self.subscribe('record/stop', self.stop_recording )
         self.subscribe('controller/event', self.handle_event)
 
+    def start_recording(self):
+        self.toggle_recording(enable=True)
+        
+    def stop_recording(self):
+        self.toggle_recording(enable=False)
+        
     def toggle_recording(self, enable=True, filename=None):
         if enable and not self.recording:
             self.recording = True
@@ -28,7 +34,7 @@ def toggle_recording(self, enable=True, filename=None):
             self.save_events()
             self.log(f"Recording stopped and saved: {self.filename}")
 
-    def handle_event(self, **event):
+    def handle_event(self, event):
         if self.recording:
             event['timestamp'] = time.time()
             self.events.append(event)

tests/test_input_recorder.py

@@ -13,17 +13,35 @@ def test_recording_toggle_and_event(self):
         self.recorder.toggle_recording(enable=True, filename='test.json')
         self.assertTrue(self.recorder.recording)
         self.assertEqual(self.recorder.filename, 'test.json')
-        event = {'ev_type': 'Key', 'code': 'BTN_A', 'state': 1}
-        self.recorder.handle_event(**event)
+        # Simulate a controller event as per new format
+        event = {
+            'time_ms': 123456,
+            'value': 1,
+            'type': 1,
+            'number': 7,
+            'topics_args': [
+                {'topic': 'tts', 'args': {'msg': 'Test Default'}}
+            ]
+        }
+        self.recorder.handle_event(event)
         self.assertEqual(len(self.recorder.events), 1)
+        self.assertIn('timestamp', self.recorder.events[0])
         self.recorder.toggle_recording(enable=False)
         self.assertFalse(self.recorder.recording)
         self.recorder.save_events.assert_called()
 
     def test_event_timestamp(self):
         self.recorder.toggle_recording(enable=True, filename='test2.json')
-        event = {'ev_type': 'Key', 'code': 'BTN_B', 'state': 1}
-        self.recorder.handle_event(**event)
+        event = {
+            'time_ms': 654321,
+            'value': 0,
+            'type': 1,
+            'number': 7,
+            'topics_args': [
+                {'topic': 'tts', 'args': {'msg': 'Test Default'}}
+            ]
+        }
+        self.recorder.handle_event(event)
         self.assertIn('timestamp', self.recorder.events[0])
         self.recorder.toggle_recording(enable=False)