create aircox_streamer as separate application

aircox_streamer/admin.py

@@ -0,0 +1,17 @@
+from django.contrib import admin
+
+from aircox.admin import StationAdmin
+from .models import Port
+
+
+__all__ = ['PortInline']
+
+
+class PortInline(admin.StackedInline):
+    model = Port
+    extra = 0
+
+
+StationAdmin.inlines = (PortInline,) + StationAdmin.inlines
+
+

aircox_streamer/apps.py

@@ -0,0 +1,5 @@
+from django.apps import AppConfig
+
+
+class AircoxStreamerConfig(AppConfig):
+    name = 'aircox_streamer'

aircox_streamer/connector.py

@@ -0,0 +1,83 @@
+import socket
+import re
+import json
+
+
+response_re = re.compile(r'(.*)\s+END\s*$')
+key_val_re = re.compile(r'(?P<key>[^=]+)="?(?P<value>([^"]|\\")+)"?')
+
+
+class Connector:
+    """
+    Connection to AF_UNIX or AF_INET, get and send data. Received
+    data can be parsed from list of `key=value` or JSON.
+    """
+    socket = None
+    """ The socket """
+    address = None
+    """
+    String to a Unix domain socket file, or a tuple (host, port) for
+    TCP/IP connection
+    """
+
+    @property
+    def is_open(self):
+        return self.socket is not None
+
+    def __init__(self, address=None):
+        if address:
+            self.address = address
+
+    def open(self):
+        if self.is_open:
+            return
+
+        family = socket.AF_UNIX if isinstance(self.address, str) else \
+            socket.AF_INET
+        try:
+            self.socket = socket.socket(family, socket.SOCK_STREAM)
+            self.socket.connect(self.address)
+        except:
+            self.close()
+            return -1
+
+    def close(self):
+        self.socket.close()
+        self.socket = None
+
+    # FIXME: return None on failed
+    def send(self, *data, try_count=1, parse=False, parse_json=False):
+        if self.open():
+            return None
+
+        data = bytes(''.join([str(d) for d in data]) + '\n', encoding='utf-8')
+        try:
+            self.socket.sendall(data)
+            data = ''
+            while not response_re.search(data):
+                data += self.socket.recv(1024).decode('utf-8')
+
+            if data:
+                data = response_re.sub(r'\1', data).strip()
+                data = self.parse(data) if parse else \
+                    self.parse_json(data) if parse_json else data
+            return data
+        except:
+            self.close()
+            if try_count > 0:
+                return self.send(data, try_count - 1)
+
+    def parse(self, value):
+        return {
+            line.groupdict()['key']: line.groupdict()['value']
+            for line in (key_val_re.search(line) for line in value.split('\n'))
+            if line
+        }
+
+    def parse_json(self, value):
+        try:
+            if value[0] == '"' and value[-1] == '"':
+                value = value[1:-1]
+            return json.loads(value) if value else None
+        except:
+            return None

aircox_streamer/liquidsoap.py

@@ -0,0 +1,358 @@
+import atexit
+import logging
+import os
+import re
+import signal
+import subprocess
+
+import psutil
+import tzlocal
+
+from django.template.loader import render_to_string
+from django.utils import timezone as tz
+
+from aircox import settings
+from aircox.models import Station, Sound
+from aircox.utils import to_seconds
+
+from .connector import Connector
+from .models import Port
+
+
+__all__ = ['BaseMetadata', 'Request', 'Streamer', 'Source',
+           'PlaylistSource', 'QueueSource']
+
+# TODO: for the moment, update in station and program names do not update the
+#       related fields.
+
+# FIXME liquidsoap does not manage timezones -- we have to convert
+#       'on_air' metadata we get from it into utc one in order to work
+#       correctly.
+
+local_tz = tzlocal.get_localzone()
+logger = logging.getLogger('aircox')
+
+
+class BaseMetadata:
+    """ Base class for handling request metadata.  """
+    controller = None
+    """ Controller """
+    rid = None
+    """ Request id """
+    uri = None
+    """ Request uri """
+    status = None
+    """ Current playing status """
+    air_time = None
+    """ Launch datetime """
+
+
+    def __init__(self, controller=None, rid=None, data=None):
+        self.controller = controller
+        self.rid = rid
+        if data is not None:
+            self.validate(data)
+
+    @property
+    def is_playing(self):
+        return self.status == 'playing'
+
+    def fetch(self):
+        data = self.controller.set('request.metadata ', self.rid, parse=True)
+        if data:
+            self.validate(data)
+
+    def validate(self, data):
+        """
+        Validate provided data and set as attribute (must already be
+        declared)
+        """
+        for key, value in data.items():
+            if hasattr(self, key) and not callable(getattr(self, key)):
+                setattr(self, key, value)
+        self.uri = data.get('initial_uri')
+
+        air_time = data.get('on_air')
+        if air_time:
+            air_time = tz.datetime.strptime(air_time, '%Y/%m/%d %H:%M:%S')
+            self.air_time = local_tz.localize(air_time)
+        else:
+            self.air_time = None
+
+
+class Request(BaseMetadata):
+    title = None
+    artist = None
+
+
+class Streamer:
+    connector = None
+    process = None
+
+    station = None
+    template_name = 'aircox_streamer/scripts/station.liq'
+    path = None
+    """ Config path """
+    sources = None
+    """ List of all monitored sources """
+    source = None
+    """ Current source being played on air """
+    # note: we disable on_air rids since we don't have use of it for the
+    # moment
+    # on_air = None
+    # """ On-air request ids (rid) """
+    inputs = None
+    """ Queryset to input ports """
+    outputs = None
+    """ Queryset to output ports """
+
+    def __init__(self, station, connector=None):
+        self.station = station
+        self.inputs = self.station.port_set.active().input()
+        self.outputs = self.station.port_set.active().output()
+
+        self.id = self.station.slug.replace('-', '_')
+        self.path = os.path.join(station.path, 'station.liq')
+        self.connector = Connector(os.path.join(station.path, 'station.sock'))
+        self.init_sources()
+
+    @property
+    def socket_path(self):
+        """ Path to Unix socket file """
+        return self.connector.address
+
+    @property
+    def is_ready(self):
+        """
+        If external program is ready to use, returns True
+        """
+        return self.send('list') != ''
+
+    @property
+    def is_running(self):
+        """ True if holds a running process """
+        if self.process is None:
+            return False
+
+        returncode = self.process.poll()
+        if returncode is None:
+            return True
+
+        self.process = None
+        logger.debug('process died with return code %s' % returncode)
+        return False
+
+    # Sources and config ###############################################
+    def send(self, *args, **kwargs):
+        return self.connector.send(*args, **kwargs) or ''
+
+    def init_sources(self):
+        streams = self.station.program_set.filter(stream__isnull=False)
+        self.dealer = QueueSource(self, 'dealer')
+        self.sources = [self.dealer] + [
+            PlaylistSource(self, program=program) for program in streams
+        ]
+
+    def make_config(self):
+        """ Make configuration files and directory (and sync sources) """
+        data = render_to_string(self.template_name, {
+            'station': self.station,
+            'streamer': self,
+            'settings': settings,
+        })
+        data = re.sub('[\t ]+\n', '\n', data)
+        data = re.sub('\n{3,}', '\n\n', data)
+
+        os.makedirs(os.path.dirname(self.path), exist_ok=True)
+        with open(self.path, 'w+') as file:
+            file.write(data)
+
+        self.sync()
+
+    def sync(self):
+        """ Sync all sources. """
+        for source in self.sources:
+            source.sync()
+
+    def fetch(self):
+        """ Fetch data from liquidsoap """
+        for source in self.sources:
+            source.fetch()
+
+        # request.on_air is not ordered: we need to do it manually
+        if self.dealer.is_playing:
+            self.source = self.dealer
+            return
+
+        self.source = next((source for source in self.sources
+                            if source.is_playing), None)
+
+    # Process ##########################################################
+    def get_process_args(self):
+        return ['liquidsoap', '-v', self.path]
+
+    def check_zombie_process(self):
+        if not os.path.exists(self.socket_path):
+            return
+
+        conns = [conn for conn in psutil.net_connections(kind='unix')
+                 if conn.laddr == self.socket_path]
+        for conn in conns:
+            if conn.pid is not None:
+                os.kill(conn.pid, signal.SIGKILL)
+
+    def run_process(self):
+        """
+        Execute the external application with corresponding informations.
+
+        This function must make sure that all needed files have been generated.
+        """
+        if self.process:
+            return
+
+        args = self.get_process_args()
+        if not args:
+            return
+
+        self.check_zombie_process()
+        self.process = subprocess.Popen(args, stderr=subprocess.STDOUT)
+        atexit.register(lambda: self.kill_process())
+
+    def kill_process(self):
+        if self.process:
+            logger.debug("kill process %s: %s", self.process.pid,
+                         ' '.join(self.get_process_args()))
+            self.process.kill()
+            self.process = None
+
+    def wait_process(self):
+        """
+        Wait for the process to terminate if there is a process
+        """
+        if self.process:
+            self.process.wait()
+            self.process = None
+
+
+class Source(BaseMetadata):
+    controller = None
+    """ parent controller """
+    id = None
+    """ source id """
+    remaining = 0.0
+    """ remaining time """
+
+    @property
+    def station(self):
+        return self.controller.station
+
+    # @property
+    # def is_on_air(self):
+    #    return self.rid is not None and self.rid in self.controller.on_air
+
+    def __init__(self, controller=None, id=None, *args, **kwargs):
+        super().__init__(controller, *args, **kwargs)
+        self.id = id
+
+    def sync(self):
+        """ Synchronize what should be synchronized """
+
+    def fetch(self):
+        data = self.controller.send(self.id, '.remaining')
+        if data:
+            self.remaining = float(data)
+
+        data = self.controller.send(self.id, '.get', parse=True)
+        if data:
+            self.validate(data if data and isinstance(data, dict) else {})
+
+    def skip(self):
+        """ Skip the current source sound """
+        self.controller.send(self.id, '.skip')
+
+    def restart(self):
+        """ Restart current sound """
+        # seek 10 hours back since there is not possibility to get current pos
+        self.seek(-216000*10)
+
+    def seek(self, n):
+        """ Seeks into the sound. """
+        self.controller.send(self.id, '.seek ', str(n))
+
+
+class PlaylistSource(Source):
+    """ Source handling playlists (program streams) """
+    path = None
+    """ Path to playlist """
+    program = None
+    """ Related program """
+    playlist = None
+    """ The playlist """
+
+    def __init__(self, controller, id=None, program=None, **kwargs):
+        id = program.slug.replace('-', '_') if id is None else id
+        self.program = program
+
+        super().__init__(controller, id=id, **kwargs)
+        self.path = os.path.join(self.station.path, self.id + '.m3u')
+
+    def get_sound_queryset(self):
+        """ Get playlist's sounds queryset """
+        return self.program.sound_set.archive()
+
+    def get_playlist(self):
+        """ Get playlist from db """
+        return self.get_sound_queryset().paths()
+
+    def write_playlist(self, playlist=[]):
+        """ Write playlist to file. """
+        os.makedirs(os.path.dirname(self.path), exist_ok=True)
+        with open(self.path, 'w') as file:
+            file.write('\n'.join(playlist or []))
+
+    def stream(self):
+        """ Return program's stream info if any (or None) as dict. """
+        # used in templates
+        # TODO: multiple streams
+        stream = self.program.stream_set.all().first()
+        if not stream or (not stream.begin and not stream.delay):
+            return
+
+        return {
+            'begin': stream.begin.strftime('%Hh%M') if stream.begin else None,
+            'end': stream.end.strftime('%Hh%M') if stream.end else None,
+            'delay': to_seconds(stream.delay) if stream.delay else 0
+        }
+
+    def sync(self):
+        playlist = self.get_playlist()
+        self.write_playlist(playlist)
+
+
+class QueueSource(Source):
+    queue = None
+    """ Source's queue (excluded on_air request) """
+    as_requests = False
+    """ If True, queue is a list of Request """
+
+    def __init__(self, *args, queue_metadata=False, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.queue_metadata = queue_metadata
+
+    def push(self, *paths):
+        """ Add the provided paths to source's play queue """
+        for path in paths:
+            self.controller.send(self.id, '_queue.push ', path)
+
+    def fetch(self):
+        super().fetch()
+        queue = self.controller.send(self.id, '_queue.queue').split(' ')
+        if not self.as_requests:
+            self.queue = queue
+            return
+
+        self.queue = [Request(self.controller, rid) for rid in queue]
+        for request in self.queue:
+            request.fetch()
+
+

aircox_streamer/management/commands/streamer.py

@@ -0,0 +1,321 @@
+"""
+Handle the audio streamer and controls it as we want it to be. It is
+used to:
+- generate config files and playlists;
+- monitor Liquidsoap, logs and scheduled programs;
+- cancels Diffusions that have an archive but could not have been played;
+- run Liquidsoap
+"""
+# TODO:
+# x controllers: remaining
+# x diffusion conflicts
+# x cancel
+# x when liquidsoap fails to start/exists: exit
+# - handle restart after failure
+# - is stream restart after live ok?
+from argparse import RawTextHelpFormatter
+import time
+
+import pytz
+from django.db.models import Q
+from django.core.management.base import BaseCommand
+from django.utils import timezone as tz
+
+from aircox.models import Station, Episode, Diffusion, Track, Sound, Log
+from aircox.utils import date_range
+
+from aircox_streamer.liquidsoap import Streamer, PlaylistSource
+
+
+# force using UTC
+tz.activate(pytz.UTC)
+
+
+class Monitor:
+    """
+    Log and launch diffusions for the given station.
+
+    Monitor should be able to be used after a crash a go back
+    where it was playing, so we heavily use logs to be able to
+    do that.
+
+    We keep trace of played items on the generated stream:
+    - sounds played on this stream;
+    - scheduled diffusions
+    - tracks for sounds of streamed programs
+    """
+    streamer = None
+    """ Streamer controller """
+    delay = None
+    """ Timedelta: minimal delay between two call of monitor. """
+    logs = None
+    """ Queryset to station's logs (ordered by -pk) """
+    cancel_timeout = 20
+    """ Timeout in minutes before cancelling a diffusion. """
+    sync_timeout = 5
+    """ Timeout in minutes between two streamer's sync. """
+    sync_next = None
+    """ Datetime of the next sync """
+
+    @property
+    def station(self):
+        return self.streamer.station
+
+    @property
+    def last_log(self):
+        """ Last log of monitored station. """
+        return self.logs.first()
+
+    @property
+    def last_diff_start(self):
+        """ Log of last triggered item (sound or diffusion). """
+        return self.logs.start().with_diff().first()
+
+    def __init__(self, streamer, delay, cancel_timeout, **kwargs):
+        self.streamer = streamer
+        # adding time ensure all calculation have a margin
+        self.delay = delay + tz.timedelta(seconds=5)
+        self.cancel_timeout = cancel_timeout
+        self.__dict__.update(kwargs)
+        self.logs = self.get_logs_queryset()
+
+    def get_logs_queryset(self):
+        """ Return queryset to assign as `self.logs` """
+        return self.station.log_set.select_related('diffusion', 'sound') \
+                           .order_by('-pk')
+
+    def monitor(self):
+        """ Run all monitoring functions once. """
+        if not self.streamer.is_ready:
+            return
+
+        self.streamer.fetch()
+
+        # Skip tracing - analyzis:
+        # Reason: multiple database request every x seconds, reducing it.
+        # We could skip this part when remaining time is higher than a minimal
+        # value (which should be derived from Command's delay). Problems:
+        # - How to trace tracks? (+ Source can change: caching log might sucks)
+        # - if liquidsoap's source/request changes: remaining time goes higher,
+        #   thus avoiding fetch
+        #
+        # Approach: something like having a mean time, such as:
+        #
+        # ```
+        # source = stream.source
+        # mean_time = source.air_time
+        #           + min(next_track.timestamp, source.remaining)
+        #           - (command.delay + 1)
+        # trace_required = \/ source' != source
+        #                  \/ source.uri' != source.uri
+        #                  \/ now < mean_time
+        # ```
+        #
+        source = self.streamer.source
+        if source and source.uri:
+            log = self.trace_sound(source)
+            if log:
+                self.trace_tracks(log)
+        else:
+            print('no source or sound for stream; source = ', source)
+
+        self.handle_diffusions()
+        self.sync()
+
+    def log(self, date=None, **kwargs):
+        """ Create a log using **kwargs, and print info """
+        kwargs.setdefault('station', self.station)
+        log = Log(date=date or tz.now(), **kwargs)
+        log.save()
+        log.print()
+        return log
+
+    def trace_sound(self, source):
+        """ Return on air sound log (create if not present). """
+        air_uri, air_time = source.uri, source.air_time
+
+        # check if there is yet a log for this sound on the source
+        log = self.logs.on_air().filter(
+            Q(sound__path=air_uri) |
+            # sound can be null when arbitrary sound file is played
+            Q(sound__isnull=True, track__isnull=True, comment=air_uri),
+            source=source.id,
+            date__range=date_range(air_time, self.delay),
+        ).first()
+        if log:
+            return log
+
+        # get sound
+        diff = None
+        sound = Sound.objects.filter(path=air_uri).first()
+        if sound and sound.episode_id is not None:
+            diff = Diffusion.objects.episode(id=sound.episode_id).on_air() \
+                                    .now(air_time).first()
+
+        # log sound on air
+        return self.log(type=Log.TYPE_ON_AIR, date=source.air_time,
+                        source=source.id, sound=sound, diffusion=diff,
+                        comment=air_uri)
+
+    def trace_tracks(self, log):
+        """
+        Log tracks for the given sound log (for streamed programs only).
+        """
+        if log.diffusion:
+            return
+
+        tracks = Track.objects \
+                      .filter(sound__id=log.sound_id, timestamp__isnull=False)\
+                      .order_by('timestamp')
+        if not tracks.exists():
+            return
+
+        # exclude already logged tracks
+        tracks = tracks.exclude(log__station=self.station, log__pk__gt=log.pk)
+        now = tz.now()
+        for track in tracks:
+            pos = log.date + tz.timedelta(seconds=track.timestamp)
+            if pos > now:
+                break
+            # log track on air
+            self.log(type=Log.TYPE_ON_AIR, date=pos, source=log.source,
+                     track=track, comment=track)
+
+    def handle_diffusions(self):
+        """
+        Handle scheduled diffusion, trigger if needed, preload playlists
+        and so on.
+        """
+        # TODO: program restart
+
+        # Diffusion conflicts are handled by the way a diffusion is defined
+        # as candidate for the next dealer's start.
+        #
+        # ```
+        # logged_diff: /\ \A diff in diffs: \E log: /\ log.type = START
+        #                                           /\ log.diff = diff
+        #                                           /\ log.date = diff.start
+        # queue_empty: /\ dealer.queue is empty
+        #              /\ \/ ~dealer.on_air
+        #                 \/ dealer.remaining < delay
+        #
+        # start_allowed: /\ diff not in logged_diff
+        #                /\ queue_empty
+        #
+        # start_canceled: /\ diff not in logged diff
+        #                 /\ ~queue_empty
+        #                 /\ diff.start < now + cancel_timeout
+        # ```
+        #
+        now = tz.now()
+        diff = Diffusion.objects.station(self.station).on_air().now(now) \
+                        .filter(episode__sound__type=Sound.TYPE_ARCHIVE) \
+                        .first()
+        # Can't use delay: diffusion may start later than its assigned start.
+        log = None if not diff else self.logs.start().filter(diffusion=diff)
+        if not diff or log:
+            return
+
+        dealer = self.streamer.dealer
+        # start
+        if not dealer.queue and dealer.rid is None or \
+                dealer.remaining < self.delay.total_seconds():
+            self.start_diff(dealer, diff)
+
+        # cancel
+        if diff.start < now - self.cancel_timeout:
+            self.cancel_diff(dealer, diff)
+
+    def start_diff(self, source, diff):
+        playlist = Sound.objects.episode(id=diff.episode_id).paths()
+        source.push(*playlist)
+        self.log(type=Log.TYPE_START, source=source.id, diffusion=diff,
+                 comment=str(diff))
+
+    def cancel_diff(self, source, diff):
+        diff.type = Diffusion.TYPE_CANCEL
+        diff.save()
+        self.log(type=Log.TYPE_CANCEL, source=source.id, diffusion=diff,
+                 comment=str(diff))
+
+    def sync(self):
+        """ Update sources' playlists. """
+        now = tz.now()
+        if self.sync_next is not None and now < self.sync_next:
+            return
+
+        self.sync_next = now + tz.timedelta(minutes=self.sync_timeout)
+
+        for source in self.streamer.sources:
+            if isinstance(source, PlaylistSource):
+                source.sync()
+
+
+class Command (BaseCommand):
+    help = __doc__
+
+    def add_arguments(self, parser):
+        parser.formatter_class = RawTextHelpFormatter
+        group = parser.add_argument_group('actions')
+        group.add_argument(
+            '-c', '--config', action='store_true',
+            help='generate configuration files for the stations'
+        )
+        group.add_argument(
+            '-m', '--monitor', action='store_true',
+            help='monitor the scheduled diffusions and log what happens'
+        )
+        group.add_argument(
+            '-r', '--run', action='store_true',
+            help='run the required applications for the stations'
+        )
+
+        group = parser.add_argument_group('options')
+        group.add_argument(
+            '-d', '--delay', type=int,
+            default=1000,
+            help='time to sleep in MILLISECONDS between two updates when we '
+                 'monitor. This influence the delay before a diffusion is '
+                 'launched.'
+        )
+        group.add_argument(
+            '-s', '--station', type=str, action='append',
+            help='name of the station to monitor instead of monitoring '
+                 'all stations'
+        )
+        group.add_argument(
+            '-t', '--timeout', type=float,
+            default=Monitor.cancel_timeout,
+            help='time to wait in MINUTES before canceling a diffusion that '
+                 'should have ran but did not. '
+        )
+        # TODO: sync-timeout, cancel-timeout
+
+    def handle(self, *args,
+               config=None, run=None, monitor=None,
+               station=[], delay=1000, timeout=600,
+               **options):
+        stations = Station.objects.filter(name__in=station) if station else \
+                   Station.objects.all()
+        streamers = [Streamer(station) for station in stations]
+
+        for streamer in streamers:
+            if config:
+                streamer.make_config()
+            if run:
+                streamer.run_process()
+
+        if monitor:
+            delay = tz.timedelta(milliseconds=delay)
+            timeout = tz.timedelta(minutes=timeout)
+            monitors = [Monitor(streamer, delay, timeout)
+                        for streamer in streamers]
+
+            while not run or streamer.is_running:
+                for monitor in monitors:
+                    monitor.monitor()
+                time.sleep(delay.total_seconds())
+
+        if run:
+            for streamer in streamers:
+                streamer.wait_process()

aircox_streamer/models.py

@@ -0,0 +1,101 @@
+from django.db import models
+from django.utils.translation import ugettext_lazy as _
+
+from aircox.models import Station
+
+
+__all__ = ['PortQuerySet', 'Port']
+
+
+class PortQuerySet(models.QuerySet):
+    def active(self, value=True):
+        """ Active ports """
+        return self.filter(active=value)
+
+    def output(self):
+        """ Filter in output ports """
+        return self.filter(direction=Port.DIRECTION_OUTPUT)
+
+    def input(self):
+        """ Fitler in input ports """
+        return self.filter(direction=Port.DIRECTION_INPUT)
+
+
+class Port(models.Model):
+    """
+    Represent an audio input/output for the audio stream
+    generation.
+
+    You might want to take a look to LiquidSoap's documentation
+    for the options available for each kind of input/output.
+
+    Some port types may be not available depending on the
+    direction of the port.
+    """
+    DIRECTION_INPUT = 0x00
+    DIRECTION_OUTPUT = 0x01
+    DIRECTION_CHOICES = ((DIRECTION_INPUT, _('input')),
+                         (DIRECTION_OUTPUT, _('output')))
+
+    TYPE_JACK = 0x00
+    TYPE_ALSA = 0x01
+    TYPE_PULSEAUDIO = 0x02
+    TYPE_ICECAST = 0x03
+    TYPE_HTTP = 0x04
+    TYPE_HTTPS = 0x05
+    TYPE_FILE = 0x06
+    TYPE_CHOICES = (
+        # display value are not translated becaused used as is in config
+        (TYPE_JACK, 'jack'), (TYPE_ALSA, 'alsa'),
+        (TYPE_PULSEAUDIO, 'pulseaudio'), (TYPE_ICECAST, 'icecast'),
+        (TYPE_HTTP, 'http'), (TYPE_HTTPS, 'https'),
+        (TYPE_FILE, 'file')
+    )
+
+    station = models.ForeignKey(
+        Station, models.CASCADE, verbose_name=_('station'), related_name='+')
+    direction = models.SmallIntegerField(
+        _('direction'), choices=DIRECTION_CHOICES)
+    type = models.SmallIntegerField(_('type'), choices=TYPE_CHOICES)
+    active = models.BooleanField(
+        _('active'), default=True,
+        help_text=_('this port is active')
+    )
+    settings = models.TextField(
+        _('port settings'),
+        help_text=_('list of comma separated params available; '
+                    'this is put in the output config file as raw code; '
+                    'plugin related'),
+        blank=True, null=True
+    )
+
+    objects = PortQuerySet.as_manager()
+
+    def __str__(self):
+        return "{direction}: {type} #{id}".format(
+            direction=self.get_direction_display(),
+            type=self.get_type_display(), id=self.pk or ''
+        )
+
+    def is_valid_type(self):
+        """
+        Return True if the type is available for the given direction.
+        """
+
+        if self.direction == self.DIRECTION_INPUT:
+            return self.type not in (
+                self.TYPE_ICECAST, self.TYPE_FILE
+            )
+
+        return self.type not in (
+            self.TYPE_HTTP, self.TYPE_HTTPS
+        )
+
+    def save(self, *args, **kwargs):
+        if not self.is_valid_type():
+            raise ValueError(
+                "port type is not allowed with the given port direction"
+            )
+
+        return super().save(*args, **kwargs)
+

aircox_streamer/serializers.py

@@ -0,0 +1,40 @@
+from rest_framework import serializers
+
+
+__all__ = ['RequestSerializer', 'StreamerSerializer', 'SourceSerializer',
+           'PlaylistSerializer', 'QueueSourceSerializer']
+# TODO: use models' serializers
+
+
+class BaseMetadataSerializer(serializers.Serializer):
+    rid = serializers.IntegerField()
+    air_time = serializers.DateTimeField()
+    uri = serializers.CharField()
+
+
+class RequestSerializer(serializers.Serializer):
+    title = serializers.CharField()
+    artist = serializers.CharField()
+
+
+class StreamerSerializer(serializers.Serializer):
+    station = serializers.CharField(source='station.title')
+
+
+class SourceSerializer(BaseMetadataSerializer):
+    id = serializers.CharField()
+    uri = serializers.CharField()
+    rid = serializers.IntegerField()
+    air_time = serializers.DateTimeField()
+    status = serializers.CharField()
+
+
+class PlaylistSerializer(SourceSerializer):
+    program = serializers.CharField(source='program.title')
+    playlist = serializers.ListField(child=serializers.CharField())
+
+
+class QueueSourceSerializer(SourceSerializer):
+    queue = serializers.ListField(child=RequestSerializer())
+
+

aircox_streamer/templates/aircox_streamer/scripts/station.liq

@@ -0,0 +1,130 @@
+{% comment %}
+Base liquidsoap station configuration.
+
+
+[stream] +--> streams ---+---> station
+                         |
+              dealer  ---'
+
+{% endcomment %}
+
+{% block functions %}
+{# Seek function #}
+def seek(source, t) =
+  t = float_of_string(default=0.,t)
+  ret = source.seek(source,t)
+  log("seek #{ret} seconds.")
+  "#{ret}"
+end
+
+{# Transition to live sources #}
+def to_live(stream, live)
+  stream = fade.final(duration=2., type='log', stream)
+  live = fade.initial(duration=2., type='log', live)
+  add(normalize=false, [stream,live])
+end
+
+{# Transition to stream sources #}
+def to_stream(live, stream)
+  source.skip(stream)
+  add(normalize=false, [live,stream])
+end
+
+
+{% comment %}
+An interactive source is a source that:
+- is skippable through the given id on external interfaces
+- is seekable through the given id and amount of seconds on e.i.
+- store metadata
+{% endcomment %}
+def interactive (id, s) =
+    server.register(namespace=id,
+                    description="Seek to a relative position",
+                    usage="seek <duration>",
+                    "seek", fun (x) ->  begin seek(s, x) end)
+    server.register(namespace=id,
+                    description="Get source's track remaining time",
+                    usage="remaining",
+                    "remaining", fun (_) ->  begin json_of(source.remaining(s)) end)
+
+    s = store_metadata(id=id, size=1, s)
+    add_skip_command(s)
+    s
+end
+
+{% comment %}
+A stream is an interactive playlist
+{% endcomment %}
+def stream (id, file) =
+    s = playlist(mode = "random", reload_mode='watch', file)
+    interactive(id, s)
+end
+{% endblock %}
+
+
+{% block config %}
+set("server.socket", true)
+set("server.socket.path", "{{ streamer.socket_path }}")
+set("log.file.path", "{{ station.path }}/liquidsoap.log")
+{% for key, value in settings.AIRCOX_LIQUIDSOAP_SET.items %}
+set("{{ key|safe }}", {{ value|safe }})
+{% endfor %}
+{% endblock %}
+
+{% block config_extras %}
+{% endblock %}
+
+
+{% block sources %}
+{% with source=streamer.dealer %}
+live = interactive('{{ source.id }}',
+    request.queue(id="{{ source.id }}_queue")
+)
+{% endwith %}
+
+
+streams = rotate(id="streams", [
+    {% for source in streamer.sources %}
+    {% if source != streamer.dealer %}
+    {% with stream=source.stream %}
+        {% if stream.delay %}
+        delay({{ stream.delay }}.,
+              stream("{{ source.id }}", "{{ source.path }}")),
+        {% elif stream.begin and stream.end %}
+        at({ {{stream.begin}}-{{stream.end}} },
+           stream("{{ source.id }}", "{{ source.path }}")),
+        {% else %}
+        stream("{{ source.id }}", "{{ source.path }}"),
+        {% endif %}
+    {% endwith %}
+    {% endif %}
+    {% endfor %}
+])
+
+{% endblock %}
+
+
+{% block station %}
+{{ streamer.id }} = interactive (
+    "{{ streamer.id }}",
+    fallback([
+        live,
+        streams,
+        blank(id="blank", duration=0.1)
+    ], track_sensitive=false, transitions=[to_live,to_stream])
+)
+{% endblock %}
+
+
+{% block outputs %}
+{% for output in streamer.outputs %}
+output.{{ output.get_type_display }}(
+    {% if output.settings %}
+    {{ output.settings|safe }},
+    {% endif %}
+    {{ streamer.id }}
+)
+{% endfor %}
+{% endblock %}
+
+

aircox_streamer/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

aircox_streamer/views.py

@@ -0,0 +1,3 @@
+from django.shortcuts import render
+
+# Create your views here.

aircox_streamer/viewsets.py

@@ -0,0 +1,158 @@
+from django.http import Http404
+from django.utils import timezone as tz
+
+from rest_framework import viewsets
+from rest_framework.decorators import action
+from rest_framework.permissions import IsAdminUser
+
+from aircox import controllers
+from aircox.models import Station
+from .serializers import *
+
+
+__all__ = ['Streamers', 'BaseControllerAPIView',
+           'RequestViewSet', 'StreamerViewSet', 'SourceViewSet',
+           'PlaylistSourceViewSet', 'QueueSourceViewSet']
+
+
+class Streamers:
+    date = None
+    """ next update datetime """
+    streamers = None
+    """ stations by station id """
+    timeout = None
+    """ timedelta to next update """
+
+    def __init__(self, timeout=None):
+        self.timeout = timeout or tz.timedelta(seconds=2)
+
+    def load(self, force=False):
+        # FIXME: cf. TODO in aircox.controllers about model updates
+        stations = Station.objects.active()
+        if self.streamers is None or force:
+            self.streamers = {station.pk: controllers.Streamer(station)
+                              for station in stations}
+            return
+
+        streamers = self.streamers
+        self.streamers = {station.pk: controllers.Streamer(station)
+                          if station.pk in streamers else streamers[station.pk]
+                          for station in stations}
+
+    def fetch(self):
+        if self.streamers is None:
+            self.load()
+
+        now = tz.now()
+        if self.date is not None and now < self.date:
+            return
+
+        for streamer in self.streamers.values():
+            streamer.fetch()
+        self.date = now + self.timeout
+
+    def get(self, key, default=None):
+        self.fetch()
+        return self.streamers.get(key, default)
+
+    def values(self):
+        self.fetch()
+        return self.streamers.values()
+
+    def __getitem__(self, key):
+        return self.streamers[key]
+
+
+streamers = Streamers()
+
+
+class BaseControllerAPIView(viewsets.ViewSet):
+    permission_classes = (IsAdminUser,)
+    serializer = None
+    streamer = None
+
+    def get_streamer(self, pk=None):
+        streamer = streamers.get(self.request.pk if pk is None else pk)
+        if not streamer:
+            raise Http404('station not found')
+        return streamer
+
+    def get_serializer(self, obj, **kwargs):
+        return self.serializer(obj, **kwargs)
+
+    def serialize(self, obj, **kwargs):
+        serializer = self.get_serializer(obj, **kwargs)
+        return serializer.data
+
+    def dispatch(self, request, *args, **kwargs):
+        self.streamer = self.get_streamer(request.station.pk)
+        return super().dispatch(request, *args, **kwargs)
+
+
+class RequestViewSet(BaseControllerAPIView):
+    serializer = RequestSerializer
+
+
+class StreamerViewSet(BaseControllerAPIView):
+    serializer = StreamerSerializer
+
+    def retrieve(self, request, pk=None):
+        return self.serialize(self.streamer)
+
+    def list(self, request):
+        return self.serialize(streamers.values(), many=True)
+
+
+class SourceViewSet(BaseControllerAPIView):
+    serializer = SourceSerializer
+    model = controllers.Source
+
+    def get_sources(self):
+        return (s for s in self.streamer.souces if isinstance(s, self.model))
+
+    def get_source(self, pk):
+        source = next((source for source in self.get_sources()
+                      if source.pk == pk), None)
+        if source is None:
+            raise Http404('source `%s` not found' % pk)
+        return source
+
+    def retrieve(self, request, pk=None):
+        source = self.get_source(pk)
+        return self.serialize(source)
+
+    def list(self, request):
+        return self.serialize(self.get_sources(), many=True)
+
+    @action(detail=True, methods=['POST'])
+    def sync(self, request, pk):
+        self.get_source(pk).sync()
+
+    @action(detail=True, methods=['POST'])
+    def skip(self, request, pk):
+        self.get_source(pk).skip()
+
+    @action(detail=True, methods=['POST'])
+    def restart(self, request, pk):
+        self.get_source(pk).restart()
+
+    @action(detail=True, methods=['POST'])
+    def seek(self, request, pk):
+        count = request.POST['seek']
+        self.get_source(pk).seek(count)
+
+
+class PlaylistSourceViewSet(SourceViewSet):
+    serializer = PlaylistSerializer
+    model = controllers.PlaylistSource
+
+
+class QueueSourceViewSet(SourceViewSet):
+    serializer = QueueSourceSerializer
+    model = controllers.QueueSource
+
+    @action(detail=True, methods=['POST'])
+    def push(self, request, pk):
+        self.get_source(pk).push()
+
+