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 sys
import time
import requests
import configparser
from argparse import ArgumentParser

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


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")
    parser.add_argument(
        '-l', '--loop',
        dest='loop',
        action='store_const',
        const=True,
        default=False,
        help="Combined with -u enables loop update")
    """
    parser.add_argument(
        '-d', '--loop-delay',
        dest='delay',
        type=int,
        default=15,
        help="Loop delay in seconds")
    """
    args = parser.parse_args()
    conf = configparser.ConfigParser()
    conf.read(args.conf)
    debug = conf.get('conf', 'debug', fallback="False").lower() == 'true'
    globals()['DEBUG'] = debug
    if args.update:
        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')
        if args.loop:
            delay = conf.get('conf', 'loop-delay', fallback='15')
            try:
                delay = int(delay)
            except ValueError:
                msg = "loop-delay expected to be an integer '%s' found"
                raise ValueError( msg % delay)
            prev = None
            while True:
                try:
                    if globals()['DEBUG']:
                        print("Update start");
                    prev = icecast_update(host, mount, login, password, prev)
                    if globals()['DEBUG']:
                        print("Updated");
                    time.sleep(delay)
                except Exception as e:
                    prev = None
                    print(str(e), file=sys.stderr)
                
        else:
            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()
    if len(datas['levels'])==2:
        level = 1
    else:
        level = 0
    position = datas['levels'][level]['position']
    item_id = datas['levels'][level]['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}
"""
    res = infos.format(
        title=item['title'].title(),
        author=item['authors'].title(),
        album=item['titreAlbum'].title(),
        image=item['visual'],
        youtube=item['lienYoutube'])
    if globals()['DEBUG']:
        res = "[debug mode]\n"+res
    return res


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


def icecast_update(host, mount, login=None, password=None, prev = 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)
    
    if infos != prev:
        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)

    return infos

if __name__ == '__main__':
    main()