Commit: tip - python-audiosocket (hg)

High performance Asterisk AudioSocket server implementation in asynchronous Python.

Commit MetaInfo

• Revision: cbeb3946c013a77a14a2f80152594e496628c9b6 (tree)
• Time: 2023-03-10 03:12:14
• Author: Ssor
• Commiter: Ssor

Log Message

Initial commit.

Change Summary

• add: audiosocket.py
• add: examples/google_stt_realtime_transcription.py
• add: examples/muffin_telephone.wav
• add: examples/requirements.txt
• add: examples/wave_playback.py
• add: readme.md

Incremental Difference

diff -r 000000000000 -r cbeb3946c013 audiosocket.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/audiosocket.py	Thu Mar 09 12:12:14 2023 -0600
@@ -0,0 +1,236 @@
+import asyncio
+
+
+__all__ = [
+  "HANGUP_CALL_MESSAGE",
+  "start_server",
+  "AsteriskFrameForwardError",
+  "AsteriskMemoryAllocError",
+  "AudioSocketReadError"
+]
+
+__version__ = 1.0
+
+
+HANGUP_CALL_MESSAGE = b""
+"""Can be returned from the callback function passed to `start_server` to
+manually hangup the call represented by the value of the `uuid` parameter.
+See documentation for `start_server()` for more information.
+"""
+
+
+class AsteriskFrameForwardError(Exception):
+
+  """Raised when the AudioSocket Asterisk module is unable forward a an audio
+  frame.
+  """
+
+  pass
+
+
+class AsteriskMemoryAllocError(Exception):
+
+  """Raised when the AudioSocket Asterisk module is unable to allocate memory.
+  """
+
+  pass
+
+
+class AudioSocketReadError(Exception):
+
+  """Raised when an unexpected header or payload size is received from
+  Asterisk.
+  """
+
+
+# AudioSocket message kinds (silence and hangup are never sent by Asterisk):
+_KIND_HANGUP = 0x00
+_KIND_UUID = 0x01
+_KIND_SILENCE = 0x02
+_KIND_AUDIO = 0x10
+_KIND_AUDIO_AS_BYTES = b"\x10"
+_KIND_ERROR = 0xff
+
+
+# AudioSocket errors that can occur on Asterisk's end (technically call hangup
+# is another, but we don't want to treat it as such on our end):
+_ERROR_FRAME_FORWARD = 0x02
+_ERROR_MEMEORY_ALLOC = 0x04
+
+
+async def start_server(on_audio_callback,
+                       on_exception_callback,
+                       host = None,
+                       port = None,
+                       **kwargs):
+
+  """Returns an `asyncio.Server` instance using the AudioSocket protocol has
+  its protocol factory. This function behaves just like
+  `asyncio.start_server()`, meaning the `host`, `port` and keyword arguments
+  accepted by that function have the same behavior with this one. Usually,
+  `await server_forever()` is called on the returned server instance to start
+  accepting connections.
+
+  The two callback arguments are expected to have the following function
+  signatures.
+
+  `on_audio_callback`, called any time audio data from the connected peer is
+  received. It is expected to be a function that accepts the following
+  arguments:
+
+    - `uuid`:      The universally unique ID that identifies this specific
+                   call's audio, provided as a hexdecimal string.
+
+    - `peer_name`: A tuple consisting of the IP address and port number of the
+                   remote host the audio is being sent from.
+
+    - `audio`:     A `bytearray` instance containing the received audio data.
+                   An empty `bytearray` instance (`len(audio) == 0`)
+                   indicates the call hung up and no more audio will
+                   be received. Audio is either encoded in 8KHz, 16-bit
+                   mono PCM (when using the standalone dialplan applcation), or
+                   whatever audio codec was decided upon during call setup
+                   (when using Dial() application). If this argument is empty
+                   (has a length of 0), the call has been hung up and will not
+                   generate any more audio.
+
+    - Any extra keyword arguments are passed along to
+      `asyncio.loop.create_server()`.
+
+  To send audio back to Asterisk, a bytes-like object must be returned by
+  this callback. Audio must be sent back in chunks of 65,536 bytes or less
+  (the size must be able to fit into a 16-bit unsigned integer). This audio must
+  always be encoded as 8KHz, 16-bit mono PCM, regardless of the codec in use
+  for the call.
+
+  Returning `audiosocket.HANGUP_CALL_MESSAGE` (or an empty `bytearray` instance)
+  will request that the call represented by the value of the `uuid` parameter
+  be hungup.
+
+  `on_exception_callback`, called any time an exception relating to the
+  connected peer is raised. It is expected to be a function that accepts the
+  following arguments:
+
+    - `uuid`:      The universally unique ID that identifies the specific
+                   call which caused the exception.
+
+    - `peer_name`: A tuple consisting of the IP address and port number of the
+                   remote host the exception-causing call came from.
+
+    - `error`:     An instance of the exception that occurred.
+  """
+
+  def factory():
+    protocol = _AudioSocketProtocol(on_audio_callback, on_exception_callback)
+    return protocol
+
+  loop = asyncio.get_running_loop()
+  return await loop.create_server(factory, host, port, **kwargs)
+
+
+class _AudioSocketProtocol(asyncio.BufferedProtocol):
+
+  def __init__(self, on_audio, on_exception):
+    super().__init__()
+
+    self._on_audio = on_audio
+    self._on_exception = on_exception
+    self._transport = None
+    self._peer_name = None
+    self._uuid = ""
+    self._active_kind = None
+    self._paused = False
+    self._buffer = None
+    self._write_spillover_buffer = []
+    self._next_read_size = 3
+
+
+  def _convey_or_raise_exception(self, exception):
+    if callable(self._on_exception):
+      self._on_exception(self._uuid, self._peer_name, exception)
+    else:
+      raise exception
+
+
+  def connection_made(self, transport):
+    self._peer_name = transport.get_extra_info("peername")
+    self._transport = transport
+
+
+  def connection_lost(self, exception):
+    self._on_audio(self._uuid, self._peer_name, bytearray(0))
+    if exception:
+      self._convey_or_raise_exception(exception)
+
+
+  def pause_writing(self):
+    self._paused = True
+
+
+  def resume_writing(self):
+    self._paused = False
+
+
+  def get_buffer(self, size_hint):
+    self._buffer = bytearray(self._next_read_size)
+    return self._buffer
+
+
+  def buffer_updated(self, byte_count):
+    if self._active_kind == None:
+
+      if byte_count != 3:
+        self._convey_or_raise_exception(AudioSocketReadError(msg))
+        return
+
+      self._active_kind = self._buffer[0]
+      # TODO: Convey error on unknown message kind?
+      self._next_read_size = int.from_bytes(self._buffer[1:3],
+                                            byteorder = "big")
+      return
+
+    if byte_count != self._next_read_size:
+      msg = "Expected {} byte payload, but received {}.".format(
+        self._next_read_size,
+        byte_count
+      )
+      self._convey_or_raise_exception(AudioSocketReadError(msg))
+      self._active_kind = None
+
+
+    if self._active_kind == _KIND_AUDIO:
+      to_send = self._on_audio(self._uuid, self._peer_name, self._buffer)
+
+      if to_send != None:
+        if len(to_send) == 0:
+          self._transport.write(b"\x00\x00\x00")
+          self._transport.close()
+          return
+
+        if self._paused:
+          self._write_spillover_buffer.insert(0, to_send)
+          return
+
+        if self._write_spillover_buffer:
+          self._write_spillover_buffer.insert(0, to_send)
+          to_send = self._write_spillover_buffer.pop()
+
+        audio_size = len(to_send).to_bytes(length = 2, byteorder = "big")
+        to_send = _KIND_AUDIO_AS_BYTES + audio_size + to_send
+        self._transport.write(to_send)
+
+    elif self._active_kind == _KIND_UUID:
+      self._uuid = self._buffer[:byte_count].hex()
+
+    elif self._active_kind == _KIND_ERROR:
+      if self._buffer[0] == _ERROR_FRAME_FORWARD:
+        raise AsteriskFrameForwardError()
+      elif self._buffer[0] == _ERROR_MEMEORY_ALLOC:
+        raise AsteriskMemoryAllocError()
+
+    self._active_kind = None
+    self._next_read_size = 3
+
+
+  def eof_received(self):
+    return False

diff -r 000000000000 -r cbeb3946c013 examples/google_stt_realtime_transcription.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/examples/google_stt_realtime_transcription.py	Thu Mar 09 12:12:14 2023 -0600
@@ -0,0 +1,187 @@
+import asyncio
+
+from threading import Thread
+from time import time
+from queue import SimpleQueue
+from socket import (
+  gethostname,
+  gethostbyname
+)
+
+from google.cloud import speech
+
+import audiosocket
+
+
+STREAM_TIME_LIMIT = 30.0  # 30 seconds.
+
+STREAM_CODEC = speech.RecognitionConfig.AudioEncoding.MULAW
+# `OGG_OPUS`, `WEBM_OPUS`, `SPEEX_WITH_HEADER_BYTE` and `LINEAR16` are other
+# available codecs that Asterisk can encode calls in.
+
+BIND_ADDRESS = (gethostbyname(gethostname()), 55150)
+
+STT_CONFIG = speech.RecognitionConfig(
+  encoding            = STREAM_CODEC,
+  sample_rate_hertz   = 8000,
+  audio_channel_count = 1,
+  model               = "phone_call",
+  language_code       = "en-US"
+)
+
+STT_STREAMING_CONFIG = speech.StreamingRecognitionConfig(
+  config           = STT_CONFIG,
+  single_utterance = False,
+  interim_results  = True
+)
+
+
+class AudioSocketStream():
+
+  def __init__(self):
+    super().__init__()
+    self._input_audio_queue = SimpleQueue()
+    self._active_uuids = []
+    self._hang_up = False
+
+
+  def audio_generator(self):
+    while True:
+      audio = self._input_audio_queue.get()
+      if len(audio) == 0:
+        return
+      else:
+        yield audio
+
+
+  def hang_up(self):
+    self._hang_up = True
+
+
+  def on_exception(self, uuid, peer_name, exception):
+    print(f"Call with UUID {uuid} from {peer_name} caused exception:")
+    print(exception)
+
+
+  def on_audio(self, uuid, peer_name, audio):
+
+    #print(f"UUID: {uuid}\nPeer name: {peer_name}\nAudio length: {len(audio)}")
+
+    if len(audio) == 0:
+      if uuid in self._active_uuids:
+        print("Call " + uuid + " is over.")
+        self._input_audio_queue.put(b"")
+        self._active_uuids.remove(uuid)
+        self._hang_up = False
+      return
+
+    if self._hang_up:
+      return audiosocket.HANGUP_CALL_MESSAGE
+
+    # To keep the example simple, only one call is allowed to be transcribed at
+    # a time. All other incoming calls are simply hung up if there is
+    # already an active transcription being performed.
+
+    if uuid not in self._active_uuids:
+      if len(self._active_uuids) == 1:
+        print("Received another call while already transcribing one, " \
+          + "hanging up...")
+        return audiosocket.HANGUP_CALL_MESSAGE
+      else:
+        print(f"Received call {uuid} from peer {peer_name}, " \
+          + "beginning transcription....")
+        self._active_uuids.append(uuid)
+
+    self._input_audio_queue.put(bytes(audio))
+    # Google cloud speech API expects a `bytes` object
+    # (the AudioSocket callback returns a `bytearray`).
+
+
+def transcribe(as_stream):
+  stt_client = speech.SpeechClient()
+
+  while True:
+
+    print("Waiting for call to start... (say \"stop\", or simply hang up " \
+      + "the call to end the transcription. There is a hard time limit of " \
+      + str(STREAM_TIME_LIMIT) + " seconds, this can be changed with the " \
+      + "`STREAM_TIME_LIMIT` variable.)"
+    )
+
+    for audio in as_stream.audio_generator():
+      # Wait until the first packet arrives, then start sending data to
+      # Google's API.
+      break
+
+    start_time = time()
+    print("Call started.")
+
+    requests = (
+      speech.StreamingRecognizeRequest(audio_content = content)
+      for content in as_stream.audio_generator()
+    )
+
+    responses = stt_client.streaming_recognize(STT_STREAMING_CONFIG, requests)
+
+    print("Interim result:")
+    for response in responses:
+
+      print(response)
+      # Responses typically look like the following:
+      #
+      # results {
+      #   alternatives {
+      #     transcript: "hello hello 123"
+      #     confidence: 0.58563143
+      #   }
+      #   is_final: true
+      #   result_end_time {
+      #     seconds: 4
+      #     nanos: 960000000
+      #   }
+      #   language_code: "en-us"
+      # }
+      # total_billed_time {
+      #   seconds: 5
+      # }
+      # request_id: 00000000000000000
+
+      if time() - start_time > STREAM_TIME_LIMIT:
+        print("Hanging up because stream duration limit of " \
+          + str(STREAM_TIME_LIMIT) + " seconds reached."
+        )
+        as_stream.hang_up()
+        break
+
+      elif "stop" in response.results[0].alternatives[0].transcript:
+        print("Hanging up because the \"Stop\" keyword was detected.")
+        as_stream.hang_up()
+        break
+
+
+async def main():
+  as_stream = AudioSocketStream()
+
+  server = await audiosocket.start_server(
+    as_stream.on_audio,
+    as_stream.on_exception,
+    host = BIND_ADDRESS[0],
+    port = BIND_ADDRESS[1]
+  )
+
+  transcription_thread = Thread(
+    target = transcribe,
+    args   = (as_stream,)
+  )
+
+  transcription_thread.start()
+
+  print(f"Audiosocket server listening at {BIND_ADDRESS}, expecting" \
+    + " calls to be encoded in MU-LAW, this can be changed with the " \
+    + "`STREAM_CODEC` variable. Only one call can be transcribed at a time.\n")
+
+  await server.serve_forever()
+
+
+if __name__ == "__main__":
+  asyncio.run(main())

diff -r 000000000000 -r cbeb3946c013 examples/muffin_telephone.wav
Binary file examples/muffin_telephone.wav has changed

diff -r 000000000000 -r cbeb3946c013 examples/requirements.txt
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/examples/requirements.txt	Thu Mar 09 12:12:14 2023 -0600
@@ -0,0 +1,1 @@
+google-cloud-speech

diff -r 000000000000 -r cbeb3946c013 examples/wave_playback.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/examples/wave_playback.py	Thu Mar 09 12:12:14 2023 -0600
@@ -0,0 +1,69 @@
+import asyncio
+import wave
+from socket import gethostname, gethostbyname
+
+import audiosocket
+
+
+BIND_ADDRESS = (gethostbyname(gethostname()), 55150)
+
+SAMPLE_SONG = wave.open("muffin_telephone.wav", mode = "r")
+SAMPLE_SONG = SAMPLE_SONG.readframes(SAMPLE_SONG.getnframes())
+SAMPLE_SONG_LENGTH = len(SAMPLE_SONG)
+
+
+class CallState:
+
+  def __init__(self):
+    self.call_frame_count     = 0
+    self.playback_slice_start = 0
+    self.playback_slice_end   = 320
+
+
+call_states = {}
+
+
+def on_audio(uuid, peer_name, audio):
+  if len(audio) == 0:
+    print("Call with UUID " + uuid + " was hungup.")
+    call_states.pop(uuid)
+    return
+
+  if uuid in call_states.keys():
+    state = call_states[uuid]
+  else:
+    print("Received new call with UUID of " + uuid)
+    state = CallState()
+    call_states.update({uuid: state})
+
+  slice = SAMPLE_SONG[state.playback_slice_start:state.playback_slice_end]
+
+  state.call_frame_count     += 1
+  state.playback_slice_start += 320
+  state.playback_slice_end   += 320
+
+  if state.playback_slice_start >= SAMPLE_SONG_LENGTH:
+    return audiosocket.HANGUP_CALL_MESSAGE
+  else:
+    return slice
+
+
+def on_exception(uuid, peer_name, exception):
+  print(f"Call with UUID {uuid} from {peer_name} caused exception:")
+  print(exception)
+
+
+async def main():
+  print(f"Server listening at {BIND_ADDRESS}")
+
+  server = await audiosocket.start_server(
+    on_audio,
+    on_exception,
+    host = BIND_ADDRESS[0],
+    port = BIND_ADDRESS[1]
+  )
+
+  await server.serve_forever()
+
+
+asyncio.run(main())

diff -r 000000000000 -r cbeb3946c013 readme.md
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/readme.md	Thu Mar 09 12:12:14 2023 -0600
@@ -0,0 +1,120 @@
+# Python AudioSocket Server
+
+A high performance Asterisk AudioSocket server written in asynchronous Python.
+
+## AudioSocket Introduction
+
+[AudioSocket](https://github.com/CyCoreSystems/audiosocket) is a relatively new addition to the [Asterisk IP-PBX project](https://www.asterisk.org/)
+that allows external applications to easily access the raw audio (both read/write) of
+active calls, as opposed to audio only being available via RTP.
+
+Such external applications act as a TCP server that Asterisk connects to
+after encountering an appropriate line of code in the dialplan. After, Asterisk
+will begin sending audio to the specified address/port (the encoding of this
+audio varies, as discussed in the tips section). This module is an
+implementation of such a TCP server, in Python.
+
+## Module Usage
+
+*Note: Formal documentation is provided below and in the module file itself, for
+real-world examples, see the the `examples` directory in the repository.*
+
+This Python module provides a clean and high performance interface for dealing
+with AudioSocket connections. To take advantage of the conventions established
+by the Python standard library, and to offer the best performance, this module
+is modeled around Python's concept of [Protocols](https://docs.python.org/3.11/library/asyncio-protocol.html).
+
+Underneath, this means that all connections are handled within a single thread,
+and I/O notification primitives provided by the underlying OS are used to
+signal when there is activity on any given connection, preventing the need to
+have an full OS thread per-connection. As for users of this module, this means
+all interaction with it occurs via two callbacks, and requires `asyncio` to
+be imported as well.
+
+The entire module is implemented within a single file, `audiosocket.py`. To
+start using it, simply place the file somewhere your project can import it
+from. After importing it (`import audiosocket`) and `asyncio`, the server can
+be setup with two function calls  - just like `asyncio`'s own `start_server` -
+by invoking:
+```python
+server = await audiosocket.start_server(<on_audio_callback>, <on_exception_callback>, <address>, <port>)`
+await server.serve_forever()
+```
+
+`start_server()` behaves just like `asyncio.start_server()`, meaning the
+`host`, `port` and keyword arguments accepted by that function have the same
+behavior with this one.
+
+The function callbacks should accept the following arguments:
+
+`on_audio_callback`, called any time audio data from the connected peer is
+received. It is expected to be a function that accepts the following
+arguments:
+    - `uuid`:      The universally unique ID that identifies this specific
+                   call's audio, provided as a hexdecimal string.
+
+    - `peer_name`: A tuple consisting of the IP address and port number of the
+                   remote host the audio is being sent from.
+
+    - `audio`:     A `bytearray` instance containing the received audio data.
+                   An empty `bytearray` instance (`len(audio) == 0`)
+                   indicates the call hung up and no more audio will
+                   be received. Audio is either encoded in 8KHz, 16-bit
+                   mono PCM (when using the standalone dialplan applcation), or
+                   whatever audio codec was decided upon during call setup
+                   (when using Dial() application). If this argument is empty
+                   (has a length of 0), the call has been hung up and will not
+                   generate any more audio.
+
+    - Any extra keyword arguments are passed along to
+      `asyncio.loop.create_server()`.
+
+To send audio back to Asterisk, a bytes-like object must be returned by
+this callback. Audio must be sent back in chunks of 65,536 bytes or less
+(the size must be able to fit into a 16-bit unsigned integer). This audio must
+always be encoded as 8KHz, 16-bit mono PCM, regardless of the codec in use
+for the call.
+
+Returning `audiosocket.HANGUP_CALL_MESSAGE` (or an empty `bytearray` instance)
+will request that the call represented by the value of the `uuid` parameter
+be hungup.
+
+`on_exception_callback`, called any time an exception relating to the
+connected peer is raised. It is expected to be a function that accepts the
+following arguments:
+    - `uuid`:      The universally unique ID that identifies the specific
+                   call which caused the exception.
+
+    - `peer_name`: A tuple consisting of the IP address and port number of the
+                   remote host the exception-causing call came from.
+
+    - `error`:     An instance of the exception that occurred.
+
+## Tips
+
+- There are two ways to begin an AudioSocket connection within Asterisk, via a
+standalone dialplan application (`AudioSocket(<uuid>,<address:port>)`) or as a
+channel driver to `Dial()` (`Dial(AudioSocket/<address:port>/<uuid>)`). Using
+the standalone application will cause Asterisk to send the server 8KHz, 16-bit
+mono PCM (see Issues section). Using the `Dial()` application will cause
+Asterisk to send audio encoded in whatever codec was agreed upon during call
+setup (most commonly ULAW).
+
+- Since the programming model with this module is callback-based (which is not
+all that common), as opposed to making state that must persist between function
+calls global, a recommended approach is to make the callback functions methods
+in a class. That way instance variables can be used to reduce the scope of such
+variables, while still allowing the callback paradigm to work as intended.
+
+## Issues (as of my testing with Asterisk 18)
+
+- If the standalone dialplan application is used to initiate an AudioSocket
+connection, delivering PCM audio to th server, the CPU core/thread the
+connection within Asterisk is running on will reach 100% utilization (likely
+a runaway loop, which I will try to identify and fix/report to them).
+
+
+- If there is a wireless medium at any point in the network link between
+Asterisk and the AudioSocket server, the Asterisk module encounters strange
+read/write errors. This is a problem with the Asterisk module/Asterisk itself.
+