I thought I'd share a script I whipped up to make it a lot easier to play certain radio streams from the command line. These streams come from StreamTheWorld, the company who is now responsible for all of the online feeds of various CBS Radio stations throughout the United States. Because my home stereo is connected to a headless box, it wasn't too practical to use their web-based Flash player to listen to the streams. I also discovered the feed URL was not something you could grab once and continue using; instead you need to use an API to discover it.
A few tcpdump captures later, using Wireshark as a trusty decoder, and I was in business. The procedure for using the API is rather simple:
- Form a URL with the API version, "mount" (e.g. radio callsign), and language
- Get an XML response, which if the above parameters were correct, will contain server IPs, ports, and a "mountpoint" (e.g. URL to hit)
- Hit the URL and you have audio
I've whipped up a handy script to do all of the above, and you can fetch it directly. I've also inlined it in the page below. It should be in good shape to be used as a module, but I simply invoke it directly and it handles the launching of mplayer with the correct URL for me.
#!/usr/bin/env python
from random import choice
import os
import sys
import urllib2
import xml.dom.minidom as minidom
def validate_callsign(cs):
'''
Normal callsign format is 'WWWWFFAAA', where 'WWWW' is the radio station
callsign, 'FF' is either 'AM' or 'FM', and 'AAA' is always 'AAC'.
For this function, we expect the 'WWWWFF' part as input.
'''
if not cs or not isinstance(cs, str):
raise ValueError('callsign \'%s\' is not a string.' % cs)
if len(cs) < 6:
raise ValueError('callsign \'%s\' is too short.' % cs)
if not cs.endswith('AAC'):
cs = cs + 'AAC'
band = cs[-5:-3]
if band != 'AM' and band != 'FM':
raise ValueError('callsign \'%s\' is missing \'FM\' or \'AM\'.' % cs)
return cs
def make_request(callsign):
host = 'playerservices.streamtheworld.com'
req = urllib2.Request(
'http://%s/api/livestream?version=1.2&mount=%s&lang=en' %
(host, callsign))
req.add_header('User-Agent', 'Mozilla/5.0')
return req
## Example XML document we are parsing follows, as the minidom code is so beautiful to follow
#
#<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
#<live_stream_config version="1" xmlns="http://provisioning.streamtheworld.com/player/livestream-1.2">
# <mountpoints>
# <mountpoint>
# <status>
# <status-code>200</status-code>
# <status-message>OK</status-message>
# </status>
# <servers>
# <server sid="5203">
# <ip>77.67.109.167</ip>
# <ports>
# <port>80</port>
# <port>443</port>
# <port>3690</port>
# </ports>
# </server>
# <!-- multiple server elements usually present -->
# </servers>
# <mount>WXYTFMAAC</mount>
# <format>FLV</format>
# <bitrate>64000</bitrate>
# <authentication>0</authentication>
# <timeout>0</timeout>
# </mountpoint>
# </mountpoints>
#</live_stream_config>
def t(element):
'''get the text of a DOM element'''
return element.firstChild.data
def check_status(ele):
# should only be one status element inside a mountpoint
status = ele.getElementsByTagName('status')[0]
if t(status.getElementsByTagName('status-code')[0]) != '200':
msg = t(status.getElementsByTagName('status-message')[0])
raise Exception('Error locating stream: ' + msg)
def create_stream_urls(srcfile):
doc = minidom.parse(srcfile)
mp = doc.getElementsByTagName('mountpoint')[0]
check_status(mp)
mt = t(mp.getElementsByTagName('mount')[0])
allurls = []
for s in mp.getElementsByTagName('server'):
# a thing of beauty, right?
ip = t(s.getElementsByTagName('ip')[0])
ports = [t(p) for p in s.getElementsByTagName('port')]
# yes, it is always HTTP. We see ports 80, 443, and 3690 usually
urls = ['http://%s:%s/%s' % (ip, p, mt) for p in ports]
allurls.extend(urls)
return allurls
def start_mplayer(location):
return os.system('mplayer %s' % location)
if __name__ == '__main__':
if len(sys.argv) < 2:
print 'usage: station callsign must be the first argument'
sys.exit(1)
callsign = validate_callsign(sys.argv[1])
req = make_request(callsign)
result = urllib2.urlopen(req)
urls = create_stream_urls(result)
if len(urls) > 0:
u = choice(urls)
sys.exit(start_mplayer(u))
sys.exit(1)
Simply run something like the following to use it. If I want to listen to some Detroit sports radio, the following works perfectly:
$ python streamtheworld.py WXYTFM
Given the customer list StreamTheWorld lists on their website, I would expect this to be useful for a lot more stations than I initially intended it for. If you find that is true, please let me know in the comments. On that same note, if the script needs some adjustments to allow more streams, let me know too. Enjoy!
Comments (18)