wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace assert-based type narrowing with local variables and guards

Browse Source 
Use local variable narrowing and if-guards instead of assert statements
for type safety, since asserts are stripped with python -O.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Mark Backman
2026-02-08 16:46:45 -05:00
parent 28be775740
commit 5a6cc4d35c
3 changed files with 14 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -1150,8 +1150,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
thought = concatenate_aggregated_text(self._thought_aggregation)
if self._thought_append_to_context:
assert self._thought_llm is not None, "llm is required when append_to_context is True"
if self._thought_append_to_context and self._thought_llm is not None:
self._context.add_message(
LLMSpecificMessage(
llm=self._thought_llm,

11
src/pipecat/transports/base_output.py
View File

@@ -701,15 +701,16 @@ class BaseOutputTransport(FrameProcessor):
# Notify the bot stopped speaking upstream if necessary.
await self._bot_stopped_speaking()
async def with_mixer(vad_stop_secs: float) -> AsyncGenerator[Frame, None]:
assert self._mixer is not None
async def with_mixer(
mixer: BaseAudioMixer, vad_stop_secs: float
) -> AsyncGenerator[Frame, None]:
last_frame_time = 0
silence = b"\x00" * self._audio_chunk_size
while True:
try:
frame = self._audio_queue.get_nowait()
if isinstance(frame, OutputAudioRawFrame):
frame.audio = await self._mixer.mix(frame.audio)
frame.audio = await mixer.mix(frame.audio)
last_frame_time = time.time()
yield frame
self._audio_queue.task_done()
@@ -720,7 +721,7 @@ class BaseOutputTransport(FrameProcessor):
await self._bot_stopped_speaking()
# Generate an audio frame with only the mixer's part.
frame = OutputAudioRawFrame(
audio=await self._mixer.mix(silence),
audio=await mixer.mix(silence),
sample_rate=self._sample_rate,
num_channels=self._params.audio_out_channels,
)
@@ -732,7 +733,7 @@ class BaseOutputTransport(FrameProcessor):
await asyncio.sleep(0)
if self._mixer:
return with_mixer(BOT_VAD_STOP_SECS)
return with_mixer(self._mixer, BOT_VAD_STOP_SECS)
else:
return without_mixer(BOT_VAD_STOP_SECS)

13
src/pipecat/transports/websocket/client.py
View File

@@ -141,8 +141,8 @@ class WebsocketClientSession:
self._client_task_handler(),
f"{self._transport_name}::WebsocketClientSession::_client_task_handler",
)
assert self._websocket is not None
await self._callbacks.on_connected(self._websocket)
if self._websocket:
await self._callbacks.on_connected(self._websocket)
except TimeoutError:
logger.error(f"Timeout connecting to {self._uri}")
@@ -193,17 +193,18 @@ class WebsocketClientSession:
async def _client_task_handler(self):
"""Handle incoming messages from the WebSocket connection."""
if not self._websocket:
return
websocket = self._websocket
try:
assert self._websocket is not None
websocket = self._websocket
# Handle incoming messages
async for message in websocket:
await self._callbacks.on_message(websocket, message)
except Exception as e:
logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})")
if self._websocket:
await self._callbacks.on_disconnected(self._websocket)
await self._callbacks.on_disconnected(websocket)
def __str__(self):
"""String representation of the WebSocket client session."""