Improve typing and drop deprecated APIs

tests/sdp/test_types.py

@@ -0,0 +1,97 @@
+"""Tests for SDP field types (`voip.sdp.types`)."""
+
+import pytest
+from voip.sdp.types import RTPPayloadFormat, StaticPayloadType
+
+
+class TestRTPPayloadFormatFromPt:
+    def test_from_pt__mono_static_pt_applies_defaults(self):
+        """A mono static payload type fills sample rate and encoding name."""
+        fmt = RTPPayloadFormat.from_pt(0)
+        assert fmt.payload_type == 0
+        assert fmt.encoding_name == "PCMU"
+        assert fmt.sample_rate == 8000
+        assert fmt.channels == 1
+
+    def test_from_pt__stereo_static_pt_applies_channel_count(self):
+        """A multi-channel static payload type applies its channel count."""
+        fmt = RTPPayloadFormat.from_pt(10)
+        assert fmt.payload_type == 10
+        assert fmt.encoding_name == "L16"
+        assert fmt.sample_rate == 44100
+        assert fmt.channels == 2
+
+    def test_from_pt__dynamic_pt_keeps_mono_default(self):
+        """A dynamic payload type (no static entry) defaults to one channel."""
+        fmt = RTPPayloadFormat.from_pt(96)
+        assert fmt.payload_type == 96
+        assert fmt.encoding_name is None
+        assert fmt.sample_rate is None
+        assert fmt.channels == 1
+
+
+class TestRTPPayloadFormatBytes:
+    def test_bytes__mono_omits_channel_suffix(self):
+        """Mono formats serialise without a trailing channel count."""
+        fmt = RTPPayloadFormat(
+            payload_type=0, encoding_name="PCMU", sample_rate=8000, channels=1
+        )
+        assert bytes(fmt) == b"0 PCMU/8000"
+
+    def test_bytes__stereo_includes_channel_suffix(self):
+        """Multi-channel formats serialise with a trailing channel count."""
+        fmt = RTPPayloadFormat(
+            payload_type=10, encoding_name="L16", sample_rate=44100, channels=2
+        )
+        assert bytes(fmt) == b"10 L16/44100/2"
+
+
+class TestRTPPayloadFormatParse:
+    def test_parse__stereo_rtpmap(self):
+        """Parse a multi-channel rtpmap value."""
+        fmt = RTPPayloadFormat.parse("10 L16/44100/2")
+        assert fmt.payload_type == 10
+        assert fmt.encoding_name == "L16"
+        assert fmt.sample_rate == 44100
+        assert fmt.channels == 2
+
+    def test_parse__mono_rtpmap_defaults_to_one_channel(self):
+        """Parse a mono rtpmap value without an explicit channel count."""
+        fmt = RTPPayloadFormat.parse("0 PCMU/8000")
+        assert fmt.payload_type == 0
+        assert fmt.encoding_name == "PCMU"
+        assert fmt.sample_rate == 8000
+        assert fmt.channels == 1
+
+    def test_parse__invalid_value_raises(self):
+        """An rtpmap value without a clock rate raises ValueError."""
+        with pytest.raises(ValueError):
+            RTPPayloadFormat.parse("0 PCMU")
+
+
+class TestRTPPayloadFormatFrameSize:
+    def test_frame_size__static_pt_uses_spec_frame_size(self):
+        """Static payload types report the frame size from the spec table."""
+        fmt = RTPPayloadFormat.from_pt(0)
+        assert fmt.frame_size == 160
+
+    def test_frame_size__stereo_static_pt_uses_spec_frame_size(self):
+        """Stereo static payload types report the spec frame size."""
+        fmt = RTPPayloadFormat.from_pt(10)
+        assert fmt.frame_size == 882
+
+    def test_frame_size__dynamic_pt_derives_from_sample_rate(self):
+        """Dynamic payload types derive the frame size from the sample rate."""
+        fmt = RTPPayloadFormat(payload_type=96, encoding_name="opus", sample_rate=48000)
+        assert fmt.frame_size == 48000 * 20 // 1000
+
+
+class TestStaticPayloadTypeFromPt:
+    def test_from_pt__known_pt_returns_member(self):
+        """Lookup of a known static payload type returns the member."""
+        assert StaticPayloadType.from_pt(0) is StaticPayloadType.PCMU
+
+    def test_from_pt__unknown_pt_raises(self):
+        """Lookup of an unknown static payload type raises ValueError."""
+        with pytest.raises(ValueError):
+            StaticPayloadType.from_pt(96)

voip/ai.py

@@ -226,13 +226,15 @@ async def respond(self) -> None:
             model=self.llm_model,
             messages=self.messages,
         )
-        if reply := self.emoji_pattern.sub("", response.message.content or ""):
+        content = response.message.content or ""
+        reply = self.emoji_pattern.sub("", content)
+        if reply:
             self.messages.append({"role": "assistant", "content": reply})
             logger.debug("Agent reply: %r", reply)
             await self.send_speech(reply)
 
     def on_audio_speech(self) -> None:
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         if self.cancel_audio_handle is None:
             self.cancel_audio_handle = loop.call_later(
                 self.audio_interrupt_duration.total_seconds(),

voip/audio.py

@@ -364,7 +364,7 @@ def on_audio_speech(self) -> None:
 
     def on_audio_silence(self) -> None:
         if self.flush_voice_buffer_handle is None:
-            loop = asyncio.get_event_loop()
+            loop = asyncio.get_running_loop()
             self.flush_voice_buffer_handle = loop.call_later(
                 self.silence_gap.total_seconds(),
                 self.flush_voice_buffer,

voip/mcp.py

@@ -97,8 +97,10 @@ async def say(ctx: Context, target: str, prompt: str = "") -> None:
     if not hasattr(connection_pool, "sip"):
         raise RuntimeError("VoIP not connected: call run() before using tools.")
     target_uri = parse_uri(target, connection_pool.sip.aor)
+    await ctx.info(f"Dialling {target_uri}")
     dialog = Dialog(sip=connection_pool.sip)
     await dialog.dial(target_uri, session_class=SayCall, text=prompt)
+    await ctx.info("Call completed")
 
 
 @mcp.tool

voip/sdp/messages.py

@@ -142,9 +142,7 @@ def lines(self) -> Generator[str]:
             value = getattr(self, field.session_attr)
             if field.is_list:
                 yield from (f"{field.letter}={v}" for v in value)
-            elif (
-                value is not None
-                and value != ""
-                or field.session_attr in ("version", "name")
+            elif (value is not None and value != "") or (
+                field.session_attr in ("version", "name")
             ):
                 yield f"{field.letter}={value}"

voip/sdp/types.py

@@ -275,14 +275,14 @@ class RTPPayloadFormat(ByteSerializableObject):
     payload_type: int
     fmtp: str | None = None
     encoding_name: str | None = None
-    channels: int = 1
+    channels: int | None = None
     sample_rate: int | None = None
 
     def __post_init__(self):
         try:
             default = StaticPayloadType.from_pt(self.payload_type)
         except ValueError:
-            pass
+            self.channels = self.channels or 1
         else:
             self.sample_rate = self.sample_rate or default.sample_rate
             self.encoding_name = self.encoding_name or default.encoding_name

voip/sip/messages.py

@@ -175,7 +175,12 @@ def first_line(self) -> str:
 
     @classmethod
     def from_request(
-        cls, request: Request, *, headers=None, dialog: Dialog = None, **kwargs
+        cls,
+        request: Request,
+        *,
+        headers: dict[str, str] | None = None,
+        dialog: Dialog | None = None,
+        **kwargs,
     ) -> Response:
         """Create a response from a request, copying relevant headers."""
         response_headers = SIPHeaderDict(

voip/sip/transactions.py

@@ -899,6 +899,9 @@ def response_received(self, response: Response) -> None:
                 return
             case 2:  # OK
                 self.start_call(response)
+                self.ack(response)
+                self.complete()
+                return
         # SRTP offer rejected as "not acceptable" → fall back to plain RTP.
         if self.offer_srtp and response.status_code in (
             SIPStatus.NOT_ACCEPTABLE_HERE,  # 488

voip/sip/types.py

@@ -78,7 +78,7 @@ def __new__(
         user: str | None = None,
         password: str | None = None,
         port: int | None = None,
-        parameters: dict[str, str] = None,
+        parameters: dict[str, str] | None = None,
         headers: dict[str, str] | None = None,
     ) -> SipURI:
         try:
@@ -192,7 +192,7 @@ def ttl(self) -> int | None:
             return None
 
     @property
-    def transport(self):
+    def transport(self) -> str:
         return (
             self.parameters.get("transport", "UDP").upper()
             if self.scheme == "sip"