import time class AudioPlayerError(Exception): pass class PlayStatus: STOPPED = 0 PLAYING = 1 PAUSED = 2 ERROR = 0xFF class PlayMode: ALL_LOOP = 0 SINGLE_LOOP = 1 FOLDER_LOOP = 2 RANDOM = 3 SINGLE_STOP = 4 ALL_ONCE = 5 FOLDER_ONCE = 6 ERROR = 0xFF class StorageDevice: UDISK = 1 SD = 2 FLASH = 3 UDISK_OR_SD = 4 FLASH_OR_UDISK = 5 FLASH_OR_SD = 6 ERROR = 0xFF class PlayDevice: UDISK = 0 SD = 1 FLASH = 2 ERROR = 0xFF class AudioPlayer: DEFAULT_TIMEOUT_MS = 500 def __init__(self, uart, debug=False): self.uart = uart self.debug = debug self._last_command = None self._expected = None # --------------------------------------------------------- # Low level # --------------------------------------------------------- def _flush_rx(self): while self.uart.in_waiting: self.uart.read() def _build_frame(self, command, data=b""): frame = bytearray() frame.append(command) frame.append((~command) & 0xFF) frame.append(len(data)) frame.extend(data) checksum = sum(frame) & 0xFF frame.append(checksum) return frame def _send_command( self, command, data=b"", expected=None, ): self._flush_rx() frame = self._build_frame(command, data) self._last_command = command self._expected = expected if self.debug: print( "TX:", " ".join(f"{b:02X}" for b in frame) ) self.uart.write(frame) def _read_response(self): if self.uart.in_waiting < 6: return None raw = self.uart.read(min(self.uart.in_waiting, 32)) if raw is None: return None data = bytes(raw) if self.debug: print( "RX:", " ".join(f"{b:02X}" for b in data) ) cmd_ok = ( ( data[0] == self._last_command and data[1] == ((~self._last_command) & 0xFF) ) or ( self._last_command == 0x05 and data[0] == 0x0A ) ) if not cmd_ok: return None if self._expected is not None: if len(data) < 5: return None if ( data[2] != self._expected[0] or data[3] != self._expected[1] ): return None checksum = sum(data[:-1]) & 0xFF if checksum != data[-1]: return None payload = data[4:-1] return payload def _transaction( self, command, data=b"", expected=None, timeout_ms=DEFAULT_TIMEOUT_MS, ): self._send_command( command, data, expected, ) start = time.monotonic() while True: payload = self._read_response() if payload is not None: return payload elapsed_ms = ( time.monotonic() - start ) * 1000 if elapsed_ms > timeout_ms: raise AudioPlayerError( "Response timeout" ) time.sleep(0.01) @staticmethod def _u16(payload): return (payload[0] << 8) | payload[1] # --------------------------------------------------------- # Device checks # --------------------------------------------------------- def get_storage_device(self): payload = self._transaction( 0x04, bytes([0x08]), expected=(0x02, 0x08), ) return payload[0] def is_ready(self): return self.get_storage_device() == StorageDevice.SD # --------------------------------------------------------- # Playback control # --------------------------------------------------------- def status(self): payload = self._transaction( 0x04, bytes([0x00]), expected=(0x02, 0x00), ) return payload[0] def play(self): payload = self._transaction( 0x04, bytes([0x01]), expected=(0x02, 0x00), ) return payload[0] def pause(self): payload = self._transaction( 0x04, bytes([0x02]), expected=(0x02, 0x00), ) return payload[0] def stop(self): payload = self._transaction( 0x04, bytes([0x03]), expected=(0x02, 0x00), ) return payload[0] def next(self): payload = self._transaction( 0x04, bytes([0x05]), expected=(0x03, 0x0E), timeout_ms=600, ) return self._u16(payload) def previous(self): payload = self._transaction( 0x04, bytes([0x04]), expected=(0x03, 0x0E), timeout_ms=600, ) return self._u16(payload) def play_index(self, index): payload = self._transaction( 0x04, bytes([ 0x06, (index >> 8) & 0xFF, index & 0xFF ]), expected=(0x03, 0x0E), ) return self._u16(payload) def play_name(self, filename): payload = self._transaction( 0x04, bytes([0x07]) + filename.encode("ascii"), expected=(0x03, 0x0E), ) return payload[0] # --------------------------------------------------------- # Volume # --------------------------------------------------------- def get_volume(self): payload = self._transaction( 0x06, bytes([0x00]), expected=(0x02, 0x00), timeout_ms=600, ) return payload[0] def set_volume(self, volume): self._send_command( 0x06, bytes([ 0x01, volume ]) ) time.sleep(0.1) def volume_up(self): self._send_command( 0x06, bytes([0x02]) ) time.sleep(0.1) def volume_down(self): self._send_command( 0x06, bytes([0x03]) ) time.sleep(0.1) # --------------------------------------------------------- # Device info # --------------------------------------------------------- def get_play_device(self): payload = self._transaction( 0x04, bytes([0x09]), expected=(0x02, 0x09), ) return payload[0] def get_total_tracks(self): payload = self._transaction( 0x04, bytes([0x0D]), expected=(0x03, 0x0D), ) return self._u16(payload) def get_current_track(self): payload = self._transaction( 0x04, bytes([0x0E]), expected=(0x03, 0x0E), ) return self._u16(payload) def get_current_path_count(self): payload = self._transaction( 0x04, bytes([0x18]), expected=(0x03, 0x18), ) return self._u16(payload) # --------------------------------------------------------- # Time functions # --------------------------------------------------------- def get_total_play_time(self): payload = self._transaction( 0x05, bytes([0x00]), expected=(0x04, 0x00), ) if len(payload) < 3: raise AudioPlayerError( "Invalid time response" ) return ( payload[0], payload[1], payload[2] ) def seek_current(self, minute, second): self._send_command( 0x04, bytes([ 0x0F, minute, second ]) ) time.sleep(0.1) def play_from_time( self, index, minute, second ): self._send_command( 0x04, bytes([ 0x10, (index >> 8) & 0xFF, index & 0xFF, minute, second ]) ) time.sleep(0.1) # --------------------------------------------------------- # Directories # --------------------------------------------------------- def next_directory(self): self._send_command(0x04, bytes([0x13])) time.sleep(0.1) def previous_directory(self): self._send_command(0x04, bytes([0x12])) time.sleep(0.1) # --------------------------------------------------------- # Play modes # --------------------------------------------------------- def get_play_mode(self): payload = self._transaction( 0x0B, bytes([0x00]), expected=(0x02, 0x00), ) return payload[0] def set_play_mode(self, mode): self._send_command( 0x0B, bytes([ 0x01, mode ]) ) time.sleep(0.1) # --------------------------------------------------------- # Repeat # --------------------------------------------------------- def repeat_between( self, start_min, start_sec, end_min, end_sec ): self._send_command( 0x08, bytes([ 0x00, start_min, start_sec, end_min, end_sec ]) ) time.sleep(0.1) def stop_repeat(self): self._send_command( 0x08, bytes([0x01]) ) time.sleep(0.1) # --------------------------------------------------------- # Sleep # --------------------------------------------------------- def sleep(self): frame = bytes([ 0x0D, 0xF3, 0x01, 0x01, 0x02 ]) self.uart.write(frame)