diff --git a/README.md b/README.md index f6898e0..648c3a7 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,9 @@ This package requires Python 3.8+ to be installed on your system (due to demands ``` pip install ltamp +python +>>> from ltamp import LtAmp +>>> LtAmp().connect() ``` > [!IMPORTANT] @@ -29,9 +32,6 @@ For full documentation of the LtAmp and LtAmpAsync classes, refer to the [wiki]( **Demonstration loop:** -> [!WARNING] -> On some linux distros, `sudo`-mode is required for running python files which interact with USB devices. - ```python import time import ltamp @@ -50,6 +50,9 @@ for i in range(5): amp.disconnect() ``` +> [!WARNING] +> On some linux distros, `sudo`-mode is required for running python files which interact with USB devices. + ## 🗺️ Roadmap Compatibility with LtAmp protocol: @@ -76,11 +79,21 @@ Other TODOs: - [x] Publish to Pypi - [x] Continuous Deployment - [x] LtAmpAsync/Base Classes +- [x] Document classes in Wiki - [ ] Unit Testing -- [ ] Customize timeout +- [x] Customize timeout - [ ] Complete Protocol - [x] Other LT Amps -- [ ] Multiple Amps +- [ ] Multiple Connected Devices + +Fully-tested LT-series amps: + +- [x] LT25 +- [ ] LT40S +- [ ] LT50 +- [ ] LTX100? +- [ ] LTX50? +- [ ] R15-R100? ## 🛠️ Contributing diff --git a/ltamp/__init__.py b/ltamp/__init__.py index b550208..c6e0c3a 100644 --- a/ltamp/__init__.py +++ b/ltamp/__init__.py @@ -1,7 +1,7 @@ -from .hidWrapper import HIDWrapper, HIDDevice, HIDBackendNotFound +from .hidWrapper import HIDWrapper, HIDDevice from .ltAmpSync import LtAmp from .ltAmpAsync import LtAmpAsync __all__ = [ - 'LtAmp', 'LtAmpAsync', 'HIDBackendNotFound', 'HIDWrapper', 'HIDDevice' + 'LtAmp', 'LtAmpAsync', 'HIDWrapper', 'HIDDevice' ] diff --git a/ltamp/hidWrapper.py b/ltamp/hidWrapper.py index 9486bc3..5c5299c 100644 --- a/ltamp/hidWrapper.py +++ b/ltamp/hidWrapper.py @@ -2,9 +2,6 @@ import platform from .hidDevice import HIDDevice -class HIDBackendNotFound(Exception): - pass - class HIDWrapper: """unified HID interface that works across platforms""" diff --git a/ltamp/ltAmpAsync.py b/ltamp/ltAmpAsync.py index dfd9721..c834bce 100644 --- a/ltamp/ltAmpAsync.py +++ b/ltamp/ltAmpAsync.py @@ -38,7 +38,6 @@ class LtAmpAsync(LtAmpBase): request_product_id() get product ID - - Data: -- last_message Last parsed message - device Current HID device connection - hid_wrapper HID wrapper instance for backend operations """ @@ -68,7 +67,7 @@ class LtAmpAsync(LtAmpBase): self._cs_event.clear() request_connection_status(self.device) try: - await asyncio.wait_for(self._cs_event.wait(), timeout=2.0) + await asyncio.wait_for(self._cs_event.wait(), timeout=self.timeout) return True except asyncio.TimeoutError: return False @@ -78,7 +77,7 @@ class LtAmpAsync(LtAmpBase): self._ps_event.clear() request_current_preset(self.device) try: - await asyncio.wait_for(self._ps_event.wait(), timeout=2.0) + await asyncio.wait_for(self._ps_event.wait(), timeout=self.timeout) return self._last_preset except asyncio.TimeoutError: raise TimeoutError("No current preset response received.") @@ -89,7 +88,7 @@ class LtAmpAsync(LtAmpBase): from .protocol import request_firmware_version request_firmware_version(self.device) try: - await asyncio.wait_for(self._fw_event.wait(), timeout=2.0) + await asyncio.wait_for(self._fw_event.wait(), timeout=self.timeout) return self._last_firmware_version except asyncio.TimeoutError: raise TimeoutError("No firmware version response received.") @@ -99,7 +98,7 @@ class LtAmpAsync(LtAmpBase): self._qa_event.clear() request_qa_slots(self.device) try: - await asyncio.wait_for(self._qa_event.wait(), timeout=10.0) + await asyncio.wait_for(self._qa_event.wait(), timeout=self.timeout) return self._last_qa_slots except asyncio.TimeoutError: raise TimeoutError("No QA slots response received.") @@ -109,7 +108,7 @@ class LtAmpAsync(LtAmpBase): self._aud_event.clear() request_audition_state(self.device) try: - await asyncio.wait_for(self._aud_event.wait(), timeout=2.0) + await asyncio.wait_for(self._aud_event.wait(), timeout=self.timeout) return self._last_audition_state except asyncio.TimeoutError: raise TimeoutError("No audition state response received.") @@ -119,7 +118,7 @@ class LtAmpAsync(LtAmpBase): self._mem_event.clear() request_memory_usage(self.device) try: - await asyncio.wait_for(self._mem_event.wait(), timeout=2.0) + await asyncio.wait_for(self._mem_event.wait(), timeout=self.timeout) return self._last_memory_state except asyncio.TimeoutError: raise TimeoutError("No memory state response received.") @@ -129,7 +128,7 @@ class LtAmpAsync(LtAmpBase): self._pu_event.clear() request_processor_utilization(self.device) try: - await asyncio.wait_for(self._pu_event.wait(), timeout=2.0) + await asyncio.wait_for(self._pu_event.wait(), timeout=self.timeout) return self._last_processor_utilization except asyncio.TimeoutError: raise TimeoutError("No processor utilization response received.") @@ -139,7 +138,7 @@ class LtAmpAsync(LtAmpBase): self._gain_event.clear() request_usb_gain(self.device) try: - await asyncio.wait_for(self._gain_event.wait(), timeout=2.0) + await asyncio.wait_for(self._gain_event.wait(), timeout=self.timeout) return self._last_gain_state except asyncio.TimeoutError: raise TimeoutError("No usb gain state response received.") @@ -149,7 +148,7 @@ class LtAmpAsync(LtAmpBase): self._pid_event.clear() request_product_id(self.device) try: - await asyncio.wait_for(self._pid_event.wait(), timeout=2.0) + await asyncio.wait_for(self._pid_event.wait(), timeout=self.timeout) return self._last_product_id except asyncio.TimeoutError: raise TimeoutError("No product ID response received.") diff --git a/ltamp/ltAmpBase.py b/ltamp/ltAmpBase.py index 2431414..a681b6c 100644 --- a/ltamp/ltAmpBase.py +++ b/ltamp/ltAmpBase.py @@ -4,7 +4,7 @@ import platform import sys import os -from .hidWrapper import HIDWrapper, HIDDevice, HIDBackendNotFound +from .hidWrapper import HIDWrapper, HIDDevice # protobuf imports protocol_path = os.path.join(os.path.dirname(__file__), 'protocol') @@ -20,14 +20,13 @@ class LtAmpBase: """ base class for LT amplifier communication """ - def __init__(self, debug=False): + def __init__(self, debug=False, timeout=2.0): self.debug = debug + self.timeout = timeout self.hid_wrapper = HIDWrapper(self.debug) self.device = None - self.msg_buffer = bytearray() - self.stop_event = threading.Event() - self.last_message = None - self._input_thread = None + self._msg_buffer = bytearray() + self._stop_event = threading.Event() self._cs_event = threading.Event() # connection status self._fw_event = threading.Event() # firmware self._ps_event = threading.Event() # preset @@ -63,12 +62,12 @@ class LtAmpBase: return True def disconnect(self): - self.stop_event.set() + self._stop_event.set() if self.device: self.device.close() def _input_thread_proc(self): - while not self.stop_event.is_set(): + while not self._stop_event.is_set(): try: data = self.device.read(64, 100) if data and any(b != 0 for b in data): @@ -93,14 +92,14 @@ class LtAmpBase: length = data_list[3] value = data_list[4:4 + length] if tag == 0x33: - self.msg_buffer = bytearray(value) + self._msg_buffer = bytearray(value) elif tag == 0x34: - self.msg_buffer += bytearray(value) + self._msg_buffer += bytearray(value) elif tag == 0x35: - self.msg_buffer += bytearray(value) + self._msg_buffer += bytearray(value) try: msg = FenderMessageLT() - msg.ParseFromString(self.msg_buffer) + msg.ParseFromString(self._msg_buffer) if self.debug: print("====== DEBUG =====\r\n", msg) @@ -151,7 +150,7 @@ class LtAmpBase: self._set_event(self._pid_event) except Exception: pass - self.msg_buffer = bytearray() + self._msg_buffer = bytearray() except Exception as e: import traceback tb = traceback.format_exc() diff --git a/ltamp/ltAmpSync.py b/ltamp/ltAmpSync.py index e475288..7c3d14c 100644 --- a/ltamp/ltAmpSync.py +++ b/ltamp/ltAmpSync.py @@ -37,7 +37,6 @@ class LtAmp(LtAmpBase): - request_product_id() get product ID - - Data: -- last_message Last parsed message - device Current HID device connection - hid_wrapper HID wrapper instance for backend operations """ @@ -45,7 +44,7 @@ class LtAmp(LtAmpBase): def request_connection_status(self): self._cs_event.clear() request_connection_status(self.device) - if self._cs_event.wait(timeout=3.0): + if self._cs_event.wait(timeout=self.timeout): return True else: return False @@ -54,7 +53,7 @@ class LtAmp(LtAmpBase): self._last_preset = None self._ps_event.clear() request_current_preset(self.device) - if self._ps_event.wait(timeout=2.0): + if self._ps_event.wait(timeout=self.timeout): return self._last_preset else: raise TimeoutError("No current preset response received.") @@ -64,7 +63,7 @@ class LtAmp(LtAmpBase): self._fw_event.clear() from .protocol import request_firmware_version request_firmware_version(self.device) - if self._fw_event.wait(timeout=2.0): + if self._fw_event.wait(timeout=self.timeout): return self._last_firmware_version else: raise TimeoutError("No firmware version response received.") @@ -73,7 +72,7 @@ class LtAmp(LtAmpBase): self._last_qa_slots = None self._qa_event.clear() request_qa_slots(self.device) - if self._qa_event.wait(timeout=5.0): + if self._qa_event.wait(timeout=self.timeout): return self._last_qa_slots else: raise TimeoutError("No QA slots response received.") @@ -82,7 +81,7 @@ class LtAmp(LtAmpBase): self._last_audition_state = None self._aud_event.clear() request_audition_state(self.device) - if self._aud_event.wait(timeout=2.0): + if self._aud_event.wait(timeout=self.timeout): return self._last_audition_state else: raise TimeoutError("No audition state response received.") @@ -91,7 +90,7 @@ class LtAmp(LtAmpBase): self._last_memory_state = None self._mem_event.clear() request_memory_usage(self.device) - if self._mem_event.wait(timeout=2.0): + if self._mem_event.wait(timeout=self.timeout): return self._last_memory_state else: raise TimeoutError("No memory state response received.") @@ -100,7 +99,7 @@ class LtAmp(LtAmpBase): self._last_processor_utilization = None self._pu_event.clear() request_processor_utilization(self.device) - if self._pu_event.wait(timeout=2.0): + if self._pu_event.wait(timeout=self.timeout): return self._last_processor_utilization else: raise TimeoutError("No processor utilization response received.") @@ -109,7 +108,7 @@ class LtAmp(LtAmpBase): self._last_gain_state = None self._gain_event.clear() request_usb_gain(self.device) - if self._gain_event.wait(timeout=2.0): + if self._gain_event.wait(timeout=self.timeout): return self._last_gain_state else: raise TimeoutError("No usb gain state response received.") @@ -118,7 +117,7 @@ class LtAmp(LtAmpBase): self._last_ftsw_state = None self._ftsw_event.clear() request_footswitch_mode(self.device) - if self._ftsw_event.wait(timeout=2.0): + if self._ftsw_event.wait(timeout=self.timeout): return self._last_ftsw_state else: raise TimeoutError("No footswitch mode response received. (Maybe not on an LT4?)") @@ -127,7 +126,7 @@ class LtAmp(LtAmpBase): self._last_product_id = None self._pid_event.clear() request_product_id(self.device) - if self._pid_event.wait(timeout=2.0): + if self._pid_event.wait(timeout=self.timeout): return self._last_product_id else: raise TimeoutError("No product ID response received.") diff --git a/pyproject.toml b/pyproject.toml index 6d3e305..ccdae40 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "ltamp" -version = "0.2.4" +version = "0.2.5" description = "🎸 Cross-platform python interface for LT guitar/bass amplifiers" readme = "README.md" license = {text = "MIT"}