Subscription for real-time updates

This script sets up a subscription to receive real-time updates from a Sennheiser device.

The script uses HTTP Basic Authentication to authenticate with the device and requests an event stream from the device's API.

Once the stream is established, it adds a subscription for a specific request and listens for incoming data. The incoming data is printed to the console as a string representation. If the stream fails to establish the reason for the failure is printed to the console.

To set up a subscription for real-time updates:
  1. Replace the ip variable with the IP address of your Sennheiser device.
  2. Replace the request variable with the API request you want to subscribe to.
  3. Replace the password variable with the password for your Sennheiser Smart Control device. 

Sennheiser electronic GmbH & Co. KG 
Version: 1.0.0 
Date: 2023-09-18 
""" 

# Import necessary libraries 
from requests.auth import HTTPBasicAuth 
import requests 

# Set up variables for authentication and API requests 
ip = '192.168.42.100'  # Replace with the IP address of your Sennheiser device 
request = '/api/audio/inputs/microphone/beam/direction'  

# Replace with the API request you want to subscribe to 
password = 'the_future_of_audio'  # Replace with the password for your Sennheiser Smart Control device 

# Set up HTTP Basic Authentication and request an event stream from the device's API 
auth = HTTPBasicAuth( 
'api', 
password 
) 
stream = requests.get( 
f'https://{ip}:443/api/ssc/state/subscriptions', 
auth=auth, stream=True, 
headers={'Accept': 'text/event-stream'}, 
verify=False 
) 

# If the stream is established, add a subscription for the specified request and listen for incoming data 
if stream.status_code == 200: 
stream.uuid = stream.headers['Content-Location'].split('/')[-1] 
requests.put( 
f'https://{ip}:443/api/ssc/state/subscriptions/{stream.uuid}/add', 
json=[request], auth=auth, 
verify=False 
) 

while True: 
data = stream.raw._fp.fp.read1(1024) 
data = data.decode('utf-8') 
print(repr(data)) 
# If the stream fails to establish, print the reason for the failure to the console 

else: 
print(stream.reason)