pstruebi
8ea7aeb412
- update multicast_server - modify config models - add dockerfile and docker compose Reviewed-on: https://gitea.pstruebi.xyz/auracaster/bumble-auracast/pulls/4
88 lines
3.6 KiB
Python
88 lines
3.6 KiB
Python
import aiohttp
|
|
import asyncio
|
|
import base64
|
|
|
|
from typing import Optional, Dict
|
|
from auracast import auracast_config
|
|
from auracast.utils.read_lc3_file import read_lc3_file
|
|
from auracast.auracast_config import AuracastConfigGroup
|
|
|
|
BASE_URL = "http://127.0.0.1:5000" # Default base URL
|
|
|
|
|
|
async def init(request_data : AuracastConfigGroup, base_url: Optional[str] = None):
|
|
url = base_url if base_url is not None else BASE_URL
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(f"{url}/init", json=request_data.model_dump()) as response:
|
|
if response.status != 200:
|
|
raise Exception(f"Error: {response.status}, {await response.text()}")
|
|
return await response.json()
|
|
|
|
|
|
async def send_audio(data_dict: Dict[str, str], base_url: Optional[str] = None):
|
|
# TODO: use base64 encoding
|
|
for language, lc3_data in data_dict.items():
|
|
data_dict[language] =lc3_data.decode('latin-1')
|
|
|
|
url = base_url if base_url is not None else BASE_URL
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(f"{url}/stream_lc3", json=data_dict) as response:
|
|
if response.status != 200:
|
|
raise Exception(f"Error: {response.status}, {await response.text()}")
|
|
return await response.json()
|
|
|
|
|
|
async def shutdown(base_url: Optional[str] = None):
|
|
url = base_url if base_url is not None else BASE_URL
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(f"{url}/shutdown") as response:
|
|
if response.status != 200:
|
|
raise Exception(f"Error: {response.status}, {await response.text()}")
|
|
return await response.json()
|
|
|
|
|
|
async def stop_audio(base_url: Optional[str] = None):
|
|
url = base_url if base_url is not None else BASE_URL
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(f"{url}/stop_audio") as response:
|
|
if response.status != 200:
|
|
raise Exception(f"Error: {response.status}, {await response.text()}")
|
|
return await response.json()
|
|
|
|
|
|
async def get_status(base_url: Optional[str] = None):
|
|
url = base_url if base_url is not None else BASE_URL
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(f"{url}/status") as response:
|
|
if response.status != 200:
|
|
raise Exception(f"Error: {response.status}, {await response.text()}")
|
|
return await response.json()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import asyncio
|
|
|
|
async def main():
|
|
config = AuracastConfigGroup(
|
|
auracast_config.AuracastBigConfigDeu()
|
|
)
|
|
config.transport = 'serial:/dev/serial/by-id/usb-ZEPHYR_Zephyr_HCI_UART_sample_81BD14B8D71B5662-if00,115200,rtscts'
|
|
|
|
# Initialize language-based configurations
|
|
for conf in config.bigs:
|
|
conf.loop = False
|
|
|
|
audio_data = { # TODO: investigate further whats the best way to actually transfer the data
|
|
"deu": read_lc3_file('src/auracast/testdata/announcement_de_10_16_32.lc3').decode('latin-1'),
|
|
#"eng": read_lc3_file('src/auracast/testdata/announcement_en_10_16_32.lc3').decode('latin-1'),
|
|
#"fra": read_lc3_file('src/auracast/testdata/announcement_fr_10_16_32.lc3').decode('latin-1'),
|
|
}
|
|
|
|
print("Getting status:", await get_status())
|
|
print("Initializing server:", await init(config))
|
|
print("Getting status:", await get_status())
|
|
print("Sending audio:", await send_audio(audio_data))
|
|
print("Getting status:", await get_status())
|
|
|
|
asyncio.run(main())
|