fip-current/fip_current.py

#!/usr/bin/python3
#
# Copyright 2016 Yann Weber
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import requests
import configparser
from argparse import ArgumentParser

API_URL = 'http://www.fipradio.fr/livemeta/7'


def main():
    """ Parse arguments and run wanted function """
    parser = ArgumentParser(description="Fetch FIP radio stream metadata")
    parser.add_argument(
        '-u', '--icecast-update',
        dest='update',
        action='store_const',
        const=True,
        default=False,
        help="Run Icecast2 metadata update")
    parser.add_argument(
        '-c', '--config',
        dest='conf',
        type=str,
        default='conf.ini',
        help="Configuration file")
    args = parser.parse_args()
    if args.update:
        conf = configparser.ConfigParser()
        conf.read(args.conf)
        host = conf.get('conf', 'host', fallback='127.0.0.1:8000')
        mount = conf.get('conf', 'mount', fallback='/example.ogg')
        login = conf.get('conf', 'login', fallback='admin')
        password = conf.get('conf', 'password', fallback='hackme')
        icecast_update(host, mount, login, password)
    else:
        print(format_current())


def get_all():
    """ Return a dict containing all FIP API metadatas """
    res = requests.get(API_URL)
    if res.status_code != 200:
        msg = "Got status code %d for %s"
        msg %= (res.status_code, API_URL)
        raise RuntimeError(msg)
    return res.json()


def get_current():
    """ Return a dict containing currently playing song metadatas """
    datas = get_all()
    position = datas['levels'][0]['position']
    item_id = datas['levels'][0]['items'][position]
    item = datas['steps'][item_id]
    expt = ['authors', 'title', 'titreAlbum', 'visual', 'lienYoutube']
    for k in expt:
        if k not in item:
            item[k] = ''

    return item


def format_current():
    """ Return a string representing formated current playing song
        metadatas """
    item = get_current()
    infos = """Title :\t\t{title}
Authors :\t{author}
Album :\t\t{album}
Visual :\t{image}
Youtube :\t{youtube}
"""
    return infos.format(
        title=item['title'].title(),
        author=item['authors'].title(),
        album=item['titreAlbum'].title(),
        image=item['visual'],
        youtube=item['lienYoutube'])


def icecast_infos():
    """ Return formated Icecast2 metadatas from FIP current song
        metadatas """
    item = get_current()
    infos = item['title']
    if len(item['authors']) > 0:
        infos += ' - '+item['authors']
    if len(item['titreAlbum']):
        infos += ' ('+item['titreAlbum']+')'
    return infos


def icecast_update(host, mount, login=None, password=None):
    """ Update metadatas to an Icecast2 mount """
    infos = icecast_infos()
    url = "http://{host}/admin/metadata"
    url = url.format(host=host)
    params = {
        'mount': mount,
        'mode': 'updinfo',
        'song': infos}
    auth = None
    if login is not None and password is not None:
        auth = (login, password)

    try:
        res = requests.get(url, auth=auth, params=params)
    except requests.exceptions.ConnectionError:
        raise RuntimeError("Connection refuse for "+res.url)

    if res.status_code != 200:
        msg = "Got status code %d for %s"
        msg %= (res.status_code, res.url)
        raise RuntimeError(msg)

if __name__ == '__main__':
    main()