quentin.dufour.io/assets/code/icy.py

import asyncio

async def readHeaders(reader):
  status_code = await reader.readline()
  assert status_code, b'HTTP/1.0 200 OK\r\n'

  headers = {}
  while True:
    data = await reader.readline()
    if data == b'\r\n':
      print("End of metadata part, all key/value headers have been read")
      return headers
    header_name, header_value = data.split(b':', 1)
    headers[header_name] = header_value 
   

async def icy():
  reader, writer = await asyncio.open_connection('streaming.radionti.com', 80)

  writer.write(b"""GET /nti-320.mp3 HTTP/1.0
Host: streaming.radionti.com
User-Agent: Icy-Test
Accept: */*
Icy-Metadata: 1

""")
  
  headers = await readHeaders(reader)
  metaint = int(headers[b'icy-metaint'])

  while True:
    audio = await reader.readexactly(metaint)
    metadata_size = int.from_bytes(await reader.readexactly(1), "big") * 16
    metadata_content = await reader.readexactly(metadata_size)
    if metadata_size > 0: print(metadata_content)
   
asyncio.run(icy())