diff --git a/ratbag_emu/device.py b/ratbag_emu/device.py index dd6db1d..2d8d23a 100644 --- a/ratbag_emu/device.py +++ b/ratbag_emu/device.py @@ -35,6 +35,9 @@ def __init__(self, name: str, info: Tuple[int, int, int], for i, r in enumerate(rdescs): self.endpoints.append(Endpoint(self, r, i)) + self._action_endpoint: Dict[ActionType, int] = {} + self._map_endpoint_actions() + self.report_rate = 100 self.fw = Firmware(self) self.hw: Dict[str, HWComponent] = {} @@ -75,6 +78,13 @@ def actuators(self, val: List[Actuator]) -> None: self._actuators = val + def _map_endpoint_actions(self) -> None: + for i, endpoint in enumerate(self.endpoints): + for features in endpoint.parsed_rdesc.input_reports.values(): + for feature in features: + if feature.usage_name in ['X', 'Y']: + self._action_endpoint[ActionType.XY] = i + def destroy(self) -> None: for endpoint in self.endpoints: endpoint.destroy() @@ -96,7 +106,7 @@ def transform_action(self, data: Dict[str, Any]) -> Dict[str, Any]: return hid_data - def send_hid_action(self, action: object) -> None: + def send_hid_action(self, endpoint_number: int, action: object) -> None: ''' Sends a HID action @@ -104,15 +114,24 @@ def send_hid_action(self, action: object) -> None: keyboard, button, etc.) so we send the action to all endpoints. The endpoint will only act on the action if it supports it. - :param action: HID action + :param endpoint_number: Number of the endpoint from where to send the action + :param action: HID action ''' - for endpoint in self.endpoints: - endpoint.send(endpoint.create_report(action)) - - def _simulate_action_xy(self, action: Dict[str, Any], packets: List[EventData], report_count: int) -> None: - # FIXME: Read max size from the report descriptor - axis_max = 127 - axis_min = -127 + endpoint = self.endpoints[endpoint_number] + endpoint.send(endpoint.create_report(action)) + + def _simulate_action_xy(self, action: Dict[str, Any], endpoint_number: int, packets: List[EventData], + report_count: int) -> None: + endpoint = self.endpoints[endpoint_number] + min = max = {} + for features in endpoint.parsed_rdesc.input_reports.values(): + for feature in features: + if feature.usage_name in ['X', 'Y']: + axis = feature.usage_name.lower() + min[axis] = feature.logical_min + max[axis] = feature.logical_max + + assert 'x' in min and 'x' in max and 'y' in min and 'y' in max, 'Endpoint does not support XY movement' # We assume a linear motion dot_buffer = {} @@ -139,6 +158,7 @@ def _simulate_action_xy(self, action: Dict[str, Any], packets: List[EventData], dot_buffer = self.transform_action(action['data']) for attr in ['x', 'y']: + axis_max = max[attr] assert dot_buffer[attr] <= axis_max * report_count step[attr] = dot_buffer[attr] / report_count @@ -149,6 +169,8 @@ def _simulate_action_xy(self, action: Dict[str, Any], packets: List[EventData], break for attr in ['x', 'y']: + axis_min = min[attr] + axis_max = max[attr] dot_buffer[attr] -= step[attr] diff = int(round(real_dot_buffer[attr] - dot_buffer[attr])) ''' @@ -180,6 +202,7 @@ def simulate_action(self, action: Dict[str, Any], type: int = None) -> None: packets: List[EventData] = [] report_count = int(round(ms2s(action['duration']) * self.report_rate)) + endpoint = self._action_endpoint[action['type']] if not report_count: report_count = 1 @@ -188,14 +211,15 @@ def simulate_action(self, action: Dict[str, Any], type: int = None) -> None: packets.append(EventData()) if action['type'] == ActionType.XY: - self._simulate_action_xy(action, packets, report_count) + self._simulate_action_xy(action, endpoint, packets, report_count) elif action['type'] == ActionType.BUTTON: - self._simulate_action_xy(action, packets, report_count) + pass s = sched.scheduler(time.time, time.sleep) next_time = 0.0 for packet in packets: - s.enter(next_time, 1, self.send_hid_action, - kwargs={'action': packet}) + s.enter(next_time, 1, self.send_hid_action, kwargs={ + 'endpoint_number': endpoint, + 'action': packet}) next_time += 1 / self.report_rate s.run() diff --git a/tests/test_device.py b/tests/test_device.py index a6bc0b2..c2a157d 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -84,7 +84,7 @@ def test_mouse_report(self, device, event_data): expected = EventData(x, y) - device.send_hid_action(expected) + device.send_hid_action(endpoint_number=0, action=expected) time.sleep(0.1) # give time for the kernel to proccess all events assert expected.x <= event_data.x <= expected.x