lda
Verified
038e152bbc
Checks / tests (ubuntu-latest, 3.11) (push) Canceled after 0s
Checks / tests (ubuntu-latest, 3.14) (push) Canceled after 0s
Checks / tests (windows-latest, 3.11) (push) Canceled after 0s
Checks / tests (windows-latest, 3.14) (push) Canceled after 0s
Checks / package (push) Canceled after 0s
57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
"""Samsung connection state shared by BLE and RFCOMM adapters.
|
|
|
|
The caller supplies received bytes and writes returned packets. This module
|
|
owns sequence numbers, framing, connection-info validation, and readiness;
|
|
it has no threads, Bluetooth dependencies, timers, or UI dependencies.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from controller.protocol import Decoder, Encoder, Frame
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Update:
|
|
frames: tuple[Frame, ...]
|
|
outgoing: tuple[bytes, ...]
|
|
became_ready: bool = False
|
|
|
|
|
|
class Session:
|
|
"""A fresh instance represents exactly one speaker connection."""
|
|
|
|
def __init__(self, strict_checksums=False):
|
|
self.encoder = Encoder()
|
|
self.decoder = Decoder(strict_checksums=strict_checksums)
|
|
self.ready = False
|
|
self._complete_sequence = None
|
|
|
|
def start(self):
|
|
return self.command(b"\x01")
|
|
|
|
def command(self, payload):
|
|
return self.encoder.encode(payload)
|
|
|
|
def receive(self, chunk):
|
|
frames = tuple(self.decoder.feed(chunk))
|
|
outgoing = []
|
|
became_ready = False
|
|
for frame in frames:
|
|
# Speaker notifications have their own sequence counter. Only an
|
|
# ACK for our link-complete request establishes session readiness.
|
|
if frame.kind == 0 and frame.sequence == self._complete_sequence and not self.ready:
|
|
self.ready = True
|
|
became_ready = True
|
|
if (
|
|
frame.kind == 1
|
|
and frame.payload
|
|
and frame.payload[0] == 2
|
|
and self._complete_sequence is None
|
|
):
|
|
payload = frame.payload
|
|
if len(payload) < 7 or len(payload) < 7 + payload[5]:
|
|
raise ValueError("Malformed speaker connection response")
|
|
outgoing.append(self.command(b"\x04"))
|
|
self._complete_sequence = self.encoder.sequence
|
|
return Update(frames, tuple(outgoing), became_ready)
|