Communication with Axis PA Systems

[HemanthPurview]

Hello, I have a few basic questions regarding the PA systems from Axis.
How can we connect to the system via code and integrate it into our own Python script?
How does the communication from the console microphone to the speakers work exactly?

[github-actions[bot]]

This automatically generated reply acts as a friendly reminder.

Answers to your questions will most often come from the community, from developers like yourself. You will, from time to time, find that Axis employees answers some of the questions, but this is not a guarantee. Think of the discussion forum as a complement to other support channels, not a replacement to any of them.

If your question remains unanswered for a period of time, please revisit it to see whether it can be improved by following the guidelines listed in Axis support guidelines.

[vivekatoffice]

Hi @HemanthEfundzz ,
Not sure what is the scope of the integration you are looking for but do have look to the REST APIs here:
Audio systems

SIP protocol is also a good option based on scope.

You can explore some open-source library for SIP in Python if SIP is not already supported in your Python application 😉 :
PySIP : An open-source Python library for easy VoIP call integration using SIP and SDP protocols. Simplify real-time communication and audio handling in your applications.

Which Axis Audio network speakers model you have for target integration? Network speakers

AXIS C6110 Network Paging Console supports SIP and VAPIX:

Image Reference

[HemanthPurview]

Is it better to use the Media Clip API for pre-recorded audio, rather than the transmit.cgi used in code 870? And for live transmission, how exactly should we use transmit.cgi? Also, could you please explain more about the storage limit part and speakers?

[HemanthPurview]

Also, I might be getting Axis Audio Manager Pro and the Network Paging Console. I am still not sure about Axis Audio Manager Pro. So, I wanted to know with and without it.

[rocky-ji]

if transmit.cgi?audiotransmitmode=live works, that will be easier than media clip. But i have not tried that. For storage in speaker, I don't have that data, but it is very limited for audio clip. if you want to put more, just insert SD card.

There is no API/information about how Axis Audio Manager Pro implement that, maybe wireshark can help to find it out.

[rocky-ji]

Attched with sample code for ransmit.cgi?audiotransmitmode=live
axis_audio.py.zip

[HemanthPurview]

Thank you so much for the code, as far as I can see we are streaming a 3 seconds audio looping again and again, by live transmission I was referring to transmitting audio from the microphone to the speaker(s) in real-time.

[vivekatoffice]

Hi @HemanthPurview ,
Adding my working Python snippet here with AXIS C1410 Mk II Network Mini Speaker (AXIS OS version 12.5.73):

import socket
import ssl
import threading
from datetime import datetime
import time
import hashlib
import os
import re
import requests
import pyaudio
import audioop

class AxisAudioStreamer:
    def __init__(self, axis_ip, port=11027, username=None, password=None, chunk_size=160):
        self.host = axis_ip
        self.port = port
        self.should_stream = False
        self.username = username
        self.password = password
        self.chunk_size = chunk_size  # 160 samples ≈ 20 ms @ 8kHz

        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        # Setup microphone stream (Windows compatible)
        self.p = pyaudio.PyAudio()
        self.stream = self.p.open(
            format=pyaudio.paInt16,   # 16-bit PCM
            channels=1,              # mono
            rate=8000,               # Axis expects 8kHz
            input=True,
            frames_per_buffer=self.chunk_size
        )

    def getAuthChallenge(self):
        url = f"http://{self.host}:{self.port}/axis-cgi/audio/transmit.cgi?audiotransmitmode=live"
        response = requests.get(url)
        if response.status_code == 401:
            auth_header = response.headers.get('WWW-Authenticate')
            if auth_header and 'Digest' in auth_header:
                return auth_header
        print(f"no digest auth header found in response: {response.headers}")
        return None

    def generateDigestResponse(self, auth_header, method='POST',
                               uri='/axis-cgi/audio/transmit.cgi?audiotransmitmode=live'):
        realm = re.search(r'realm="([^"]+)"', auth_header).group(1)
        nonce = re.search(r'nonce="([^"]+)"', auth_header).group(1)
        qop = re.search(r'qop="([^"]+)"', auth_header).group(1)
        algorithm = re.search(r'algorithm=([^,]+)', auth_header).group(1)

        ha1 = hashlib.md5(f"{self.username}:{realm}:{self.password}".encode()).hexdigest()
        ha2 = hashlib.md5(f"{method}:{uri}".encode()).hexdigest()
        nc = '00000001'
        cnonce = hashlib.md5(os.urandom(8)).hexdigest()
        response = hashlib.md5(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}".encode()).hexdigest()

        auth_response = (
            f'Digest username="{self.username}", realm="{realm}", nonce="{nonce}", uri="{uri}", '
            f'algorithm={algorithm}, qop={qop}, nc={nc}, cnonce="{cnonce}", response="{response}"'
        )
        return auth_response

    def _build_request_headers(self):
        auth_header = self.getAuthChallenge()
        digest_auth = self.generateDigestResponse(auth_header)
        headers = [
            f"POST /axis-cgi/audio/transmit.cgi?audiotransmitmode=live HTTP/1.1",
            f"Host: {self.host}",
            f"Authorization: {digest_auth}",
            "Content-Type: audio/basic",
            "", ""  # terminate headers
        ]
        return "\r\n".join(headers)

    def _stream_audio(self):
        try:
            self.socket.connect((self.host, self.port))
            self.listen_thread = threading.Thread(target=self._listen, daemon=True)
            self.listen_thread.start()

            # Send headers + CRLFCRLF
            headers = self._build_request_headers().encode()
            self.socket.sendall(headers)

            # Immediately send one frame to avoid timeout
            pcm_data = self.stream.read(self.chunk_size, exception_on_overflow=False)
            ulaw_data = audioop.lin2ulaw(pcm_data, 2)
            self.socket.sendall(ulaw_data)

            while self.should_stream:
                pcm_data = self.stream.read(self.chunk_size, exception_on_overflow=False)
                ulaw_data = audioop.lin2ulaw(pcm_data, 2)
                self.socket.sendall(ulaw_data)

        except Exception as e:
            print(f"Streaming error: {e}")
        finally:
            self.stream.stop_stream()
            self.stream.close()
            self.p.terminate()
            self.socket.close()
            self.should_stream = False




    def _listen(self):
        while self.should_stream:
            try:
                data = self.socket.recv(1024)
                if not data:
                    print("Connection closed by server")
                    break
                timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                print(f"[{timestamp}] Received: {data.decode(errors='ignore').strip()}")
            except socket.error as e:
                if self.should_stream:
                    print(f"Error receiving data: {e}")
                break
        self.stop_streaming()

    def start_streaming(self):
        if not self.should_stream:
            self.should_stream = True
            threading.Thread(target=self._stream_audio).start()

    def stop_streaming(self):
        self.should_stream = False


# ---- Example usage ----
streamer = AxisAudioStreamer(
    axis_ip="195.60.68.14",
    username="VLTuser",
    password="XXXXXXX",
    chunk_size=160  # ~20 ms per chunk
)

streamer.start_streaming()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    streamer.stop_streaming()

MIC settings:

Using FFmpeg:

ffmpeg -list_devices true -f dshow -i dummy

ffmpeg -f dshow -i audio="Stereo Mix (Synaptics HD Audio)" -probesize 32 -analyzeduration 32 -c:a pcm_mulaw -ab 128k -ac 1 -ar 16000 -f wav -chunked_post 0 -content_type audio/axis-mulaw-128 http://root:[email protected]/axis-cgi/audio/transmit.cgi

I have used the snippet provided by @rocky-ji for reference.

[vivekatoffice]

Adding my working Python snippet here with AXIS C1410 Mk II Network Mini Speaker (AXIS OS version 12.5.73) and Google Text-to-Speech:

import socket
import time
import re
import hashlib
import os
import io
from gtts import gTTS
from pydub import AudioSegment
import audioop
import requests


class AxisAudioStreamerTTS:
    def __init__(self, axis_ip, username=None, password=None,
                 http_port=80, chunk_size=800):
        self.host = axis_ip
        self.http_port = http_port   # only one port for HTTP control + audio
        self.username = username
        self.password = password
        self.chunk_size = chunk_size
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    def getAuthChallenge(self):
        url = f"http://{self.host}:{self.http_port}/axis-cgi/audio/transmit.cgi?audiotransmitmode=live"
        response = requests.get(url)
        if response.status_code == 401:
            auth_header = response.headers.get("WWW-Authenticate")
            if auth_header and "Digest" in auth_header:
                return auth_header
        print(f"no digest auth header found in response: {response.headers}")
        return None

    def generateDigestResponse(self, auth_header, method="POST",
                               uri="/axis-cgi/audio/transmit.cgi?audiotransmitmode=live"):
        realm = re.search(r'realm="([^"]+)"', auth_header).group(1)
        nonce = re.search(r'nonce="([^"]+)"', auth_header).group(1)
        qop = re.search(r'qop="([^"]+)"', auth_header).group(1)
        algorithm = re.search(r"algorithm=([^,]+)", auth_header).group(1)

        ha1 = hashlib.md5(f"{self.username}:{realm}:{self.password}".encode()).hexdigest()
        ha2 = hashlib.md5(f"{method}:{uri}".encode()).hexdigest()
        nc = "00000001"
        cnonce = hashlib.md5(os.urandom(8)).hexdigest()
        response = hashlib.md5(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}".encode()).hexdigest()

        return (
            f'Digest username="{self.username}", realm="{realm}", nonce="{nonce}", '
            f'uri="{uri}", algorithm={algorithm}, qop={qop}, nc={nc}, '
            f'cnonce="{cnonce}", response="{response}"'
        )

    def _build_request_headers(self):
        auth_header = self.getAuthChallenge()
        digest_auth = self.generateDigestResponse(auth_header)
        headers = [
            "POST /axis-cgi/audio/transmit.cgi?audiotransmitmode=live HTTP/1.1",
            f"Host: {self.host}:{self.http_port}",
            f"Authorization: {digest_auth}",
            "Content-Type: audio/basic",
            "",
            "",
        ]
        return "\r\n".join(headers)

    def _stream_audio(self, data, sample_rate=8000):
        try:
            self.socket.connect((self.host, self.http_port))  # connect to HTTP port
            self.socket.send(self._build_request_headers().encode())

            position = 0
            while position < len(data):
                chunk = data[position: position + self.chunk_size]
                position += self.chunk_size
                print(f"Sending chunk of {len(chunk)} bytes")
                self.socket.send(chunk)

                # pacing to avoid buffer overrun
                time.sleep(len(chunk) / sample_rate)
        except Exception as e:
            print(f"Streaming error: {str(e)}")
        finally:
            self.socket.close()

    def speak_and_stream(self, text):
        # Generate TTS MP3 in memory
        with io.BytesIO() as buf:
            tts = gTTS(text=text, lang="en")
            tts.write_to_fp(buf)
            buf.seek(0)
            audio = AudioSegment.from_file(buf, format="mp3")

        pcm_audio = audio.raw_data
        if audio.frame_rate != 8000:
            print(f"Resampling from {audio.frame_rate} Hz to 8000 Hz")
            pcm_audio, _ = audioop.ratecv(audio.raw_data, 2, 1,
                                          audio.frame_rate, 8000, None)

        # Convert to μ-law
        mulaw = audioop.lin2ulaw(pcm_audio, 2)

        # Stream to Axis
        self._stream_audio(mulaw, 8000)


# ---- Usage Example ----
if __name__ == "__main__":
    streamer = AxisAudioStreamerTTS(
        axis_ip="195.60.68.14",   # change to your Axis device IP
        username="VLTuser",
        password="XXXXXXX",
        http_port=11027,             # Axis HTTP(S) port (80 or 443 typically)
        #chunk_size=800            # 100ms chunks at 8kHz μ-law
        chunk_size=160            # 20ms chunks at 8kHz μ-law
    )

    streamer.speak_and_stream(
        "Hello Vivek! This is a test Google Text-to-Speech message sent to the Axis speaker."
    )

Output recorded using another Axis camera in the same room as the speaker:

vlc-axis-media_media.amp.mp4

[HemanthPurview]

Thank you so much! Both pieces of code are really helpful. I also have a question regarding the pre-recorded audio: how can we stream these audio files based on their IDs? Where should we upload and store the audio files in the speaker system, and how can we access and stream them programmatically using their IDs? Thanks again!

[vivekatoffice]

Hi @HemanthPurview ,
Glad to know the samples were helpful 😄

For Play a media clip, related discussions

1029

It is more like the following steps:

1. Uploading an audio file

2. Getting the ID

3. Play the audio clip using the Clip ID

[vivekatoffice]

Feel free to re-open it OR share the solution/alternative/comment if already resolved👁️‍🗨️