ctactic/aircox-radiocampus
1
0
Fork 0
forked from rc/aircox
Code Issues 10 Pull Requests Activity

move files

Browse Source
This commit is contained in:
bkfox
2015-12-22 08:37:17 +01:00
parent 0511ec5bc3
commit 6bb13904da
55 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
programs/README.md Normal file
View File

@ -0,0 +1,33 @@
# Aircox Programs
This application defines all base models and basic control of them. We have:
* **Nameable**: generic class used in any class needing to be named. Includes some utility functions;
* **Station**: a station
* **Program**: the program itself;
* **Diffusion**: occurrence of a program planified in the timetable. For rerun, informations are bound to the initial diffusion;
* **Schedule**: describes diffusions frequencies for each program;
* **Track**: track informations in a playlist of a diffusion;
* **Sound**: information about a sound that can be used for podcast or rerun;
* **Log**: logs
## Architecture
A Station is basically an object that represent a radio station. On each station, we use the Program object, that is declined in two different type:
* **Scheduled**: the diffusion is based on a timetable and planified through one Schedule or more; Diffusion object represent the occurrence of these programs;
* **Streamed**: the diffusion is based on random playlist, used to fill gaps between the programs;
Each program has a directory in **AIRCOX_PROGRAMS_DIR**; For each, subdir:
* **archives**: complete episode record, can be used for diffusions or as a podcast
* **excerpts**: excerpt of an episode, or other elements, can be used as a podcast
## manage.py's commands
* **diffusions_monitor**: update/create, check and clean diffusions; When a diffusion is created its type can be set on "unconfirmed" (this depends on the approval mode).
* **sound_monitor**: check for existing and missing sounds files in programs directories and synchronize the database. Can also check for the quality of file and synchronize the database according to them.
* **sound_quality_check**: check for the quality of the file (don't update database)
## Requirements
* Sox (and soxi): sound file monitor and quality check
* requirements.txt for python's dependecies

0
programs/__init__.py Executable file
View File

123
programs/admin.py Executable file
View File

@ -0,0 +1,123 @@
import copy
from django import forms
from django.contrib import admin
from django.db import models
from aircox.programs.models import *
#
# Inlines
#
class SoundInline (admin.TabularInline):
model = Sound
class ScheduleInline (admin.TabularInline):
model = Schedule
extra = 1
class StreamInline (admin.TabularInline):
fields = ['delay', 'begin', 'end']
model = Stream
extra = 1
# from suit.admin import SortableTabularInline, SortableModelAdmin
#class TrackInline (SortableTabularInline):
# fields = ['artist', 'name', 'tags', 'position']
# form = TrackForm
# model = Track
# sortable = 'position'
# extra = 10
class NameableAdmin (admin.ModelAdmin):
fields = [ 'name' ]
list_display = ['id', 'name']
list_filter = []
search_fields = ['name',]
@admin.register(Sound)
class SoundAdmin (NameableAdmin):
fields = None
list_display = ['id', 'name', 'duration', 'type', 'mtime', 'good_quality', 'removed', 'public']
fieldsets = [
(None, { 'fields': NameableAdmin.fields + ['path', 'type'] } ),
(None, { 'fields': ['embed', 'duration', 'mtime'] }),
(None, { 'fields': ['removed', 'good_quality', 'public' ] } )
]
readonly_fields = ('path', 'duration',)
@admin.register(Stream)
class StreamAdmin (admin.ModelAdmin):
list_display = ('id', 'program', 'delay', 'begin', 'end')
@admin.register(Station)
class StationAdmin (NameableAdmin):
fields = NameableAdmin.fields + [ 'active', 'public', 'fallback' ]
@admin.register(Program)
class ProgramAdmin (NameableAdmin):
fields = NameableAdmin.fields + [ 'station', 'active' ]
# TODO list_display
inlines = [ ScheduleInline, StreamInline ]
# SO#8074161
#def get_form (self, request, obj=None, **kwargs):
#if obj:
# if Schedule.objects.filter(program = obj).count():
# self.inlines.remove(StreamInline)
# elif Stream.objects.filter(program = obj).count():
# self.inlines.remove(ScheduleInline)
#return super().get_form(request, obj, **kwargs)
@admin.register(Diffusion)
class DiffusionAdmin (admin.ModelAdmin):
def archives (self, obj):
sounds = [ str(s) for s in obj.get_archives()]
return ', '.join(sounds) if sounds else ''
def conflicts (self, obj):
if obj.type == Diffusion.Type['unconfirmed']:
return ', '.join([ str(d) for d in obj.get_conflicts()])
return ''
list_display = ('id', 'type', 'date', 'program', 'initial', 'archives', 'conflicts')
list_filter = ('type', 'date', 'program')
list_editable = ('type', 'date')
fields = ['type', 'date', 'initial', 'program', 'sounds']
def get_form(self, request, obj=None, **kwargs):
if request.user.has_perm('aircox_program.programming'):
self.readonly_fields = []
else:
self.readonly_fields = ['program', 'date', 'duration']
if obj and obj.initial:
self.readonly_fields += ['program', 'sounds']
return super().get_form(request, obj, **kwargs)
def get_queryset(self, request):
qs = super(DiffusionAdmin, self).get_queryset(request)
if '_changelist_filters' in request.GET or \
'type__exact' in request.GET and \
str(Diffusion.Type['unconfirmed']) in request.GET['type__exact']:
return qs
return qs.exclude(type = Diffusion.Type['unconfirmed'])
@admin.register(Log)
class LogAdmin (admin.ModelAdmin):
list_display = ['id', 'date', 'source', 'comment', 'related_object']
list_filter = ['date', 'related_type']
admin.site.register(Track)
admin.site.register(Schedule)

0
programs/management/__init__.py Normal file
View File

BIN
programs/management/__pycache__/__init__.cpython-34.pyc Normal file
View File

Binary file not shown.

BIN
programs/management/__pycache__/__init__.cpython-35.pyc Normal file
View File

Binary file not shown.

0
programs/management/commands/__init__.py Normal file
View File

0
programs/management/commands/_private.py Normal file
View File

158
programs/management/commands/diffusions_monitor.py Normal file
View File

@ -0,0 +1,158 @@
"""
Manage diffusions using schedules, to update, clean up or check diffusions.
A generated diffusion can be unconfirmed, that means that the user must confirm
it by changing its type to "normal". The behaviour is controlled using
--approval.
Different actions are available:
- "update" is the process that is used to generated them using programs
schedules for the (given) month.
- "clean" will remove all diffusions that are still unconfirmed and have been
planified before the (given) month.
- "check" will remove all diffusions that are unconfirmed and have been planified
from the (given) month and later.
"""
from argparse import RawTextHelpFormatter
from django.core.management.base import BaseCommand, CommandError
from django.utils import timezone as tz
from aircox.programs.models import *
class Actions:
@staticmethod
def __check_conflicts (item, saved_items):
"""
Check for conflicts, and update conflictual
items if they have been generated during this
update.
Return the number of conflicts
"""
conflicts = item.get_conflicts()
if not conflicts:
item.type = Diffusion.Type['normal']
return 0
item.type = Diffusion.Type['unconfirmed']
for conflict in conflicts:
if conflict.pk in saved_items and \
conflict.type != Diffusion.Type['unconfirmed']:
conflict.type = Diffusion.Type['unconfirmed']
conflict.save()
return len(conflicts)
@classmethod
def update (cl, date, mode):
manual = (mode == 'manual')
if not manual:
saved_items = set()
count = [0, 0]
for schedule in Schedule.objects.filter(program__active = True) \
.order_by('initial'):
# in order to allow rerun links between diffusions, we save items
# by schedule;
items = schedule.diffusions_of_month(date, exclude_saved = True)
count[0] += len(items)
if manual:
Diffusion.objects.bulk_create(items)
else:
for item in items:
count[1] += cl.__check_conflicts(item, saved_items)
item.save()
saved_items.add(item)
print('> {} new diffusions for schedule #{} ({})'.format(
len(items), schedule.id, str(schedule)
))
print('total of {} diffusions have been created,'.format(count[0]),
'do not forget manual approval' if manual else
'{} conflicts found'.format(count[1]))
@staticmethod
def clean (date):
qs = Diffusion.objects.filter(type = Diffusion.Type['unconfirmed'],
date__lt = date)
print('{} diffusions will be removed'.format(qs.count()))
qs.delete()
@staticmethod
def check (date):
qs = Diffusion.objects.filter(type = Diffusion.Type['unconfirmed'],
date__gt = date)
items = []
for diffusion in qs:
schedules = Schedule.objects.filter(program = diffusion.program)
for schedule in schedules:
if schedule.match(diffusion.date):
break
else:
print('> #{}: {}'.format(diffusion.pk, str(diffusion)))
items.append(diffusion.id)
print('{} diffusions will be removed'.format(len(items)))
if len(items):
Diffusion.objects.filter(id__in = items).delete()
class Command (BaseCommand):
help= __doc__
def add_arguments (self, parser):
parser.formatter_class=RawTextHelpFormatter
now = tz.datetime.today()
group = parser.add_argument_group('action')
group.add_argument(
'--update', action='store_true',
help='generate (unconfirmed) diffusions for the given month. '
'These diffusions must be confirmed manually by changing '
'their type to "normal"')
group.add_argument(
'--clean', action='store_true',
help='remove unconfirmed diffusions older than the given month')
group.add_argument(
'--check', action='store_true',
help='check future unconfirmed diffusions from the given date '
'agains\'t schedules and remove it if that do not match any '
'schedule')
group = parser.add_argument_group('date')
group.add_argument(
'--year', type=int, default=now.year,
help='used by update, default is today\'s year')
group.add_argument(
'--month', type=int, default=now.month,
help='used by update, default is today\'s month')
group = parser.add_argument_group('mode')
group.add_argument(
'--approval', type=str, choices=['manual', 'auto'],
default='auto',
help='manual means that all generated diffusions are unconfirmed, '
'thus must be approved manually; auto confirmes all '
'diffusions except those that conflicts with others'
)
def handle (self, *args, **options):
date = tz.datetime(year = options.get('year'),
month = options.get('month'),
day = 1)
date = tz.make_aware(date)
if options.get('update'):
Actions.update(date, mode = options.get('mode'))
elif options.get('clean'):
Actions.clean(date)
elif options.get('check'):
Actions.check(date)
else:
raise CommandError('no action has been given')

228
programs/management/commands/sounds_monitor.py Normal file
View File

@ -0,0 +1,228 @@
"""
Monitor sound files; For each program, check for:
- new files;
- deleted files;
- differences between files and sound;
- quality of the files;
It tries to parse the file name to get the date of the diffusion of an
episode and associate the file with it; We use the following format:
yyyymmdd[_n][_][name]
Where:
'yyyy' the year of the episode's diffusion;
'mm' the month of the episode's diffusion;
'dd' the day of the episode's diffusion;
'n' the number of the episode (if multiple episodes);
'name' the title of the sound;
To check quality of files, call the command sound_quality_check using the
parameters given by the setting AIRCOX_SOUND_QUALITY. This script requires
Sox (and soxi).
"""
import os
import re
import subprocess
from argparse import RawTextHelpFormatter
from django.core.management.base import BaseCommand, CommandError
from aircox.programs.models import *
import aircox.programs.settings as settings
import aircox.programs.utils as utils
class Command (BaseCommand):
help= __doc__
def report (self, program = None, component = None, *content):
if not component:
print('{}: '.format(program), *content)
else:
print('{}, {}: '.format(program, component), *content)
def add_arguments (self, parser):
parser.formatter_class=RawTextHelpFormatter
parser.add_argument(
'-q', '--quality_check', action='store_true',
help='Enable quality check using sound_quality_check on all ' \
'sounds marqued as not good'
)
parser.add_argument(
'-s', '--scan', action='store_true',
help='Scan programs directories for changes, plus check for a '
' matching episode on sounds that have not been yet assigned'
)
def handle (self, *args, **options):
if options.get('scan'):
self.scan()
if options.get('quality_check'):
self.check_quality(check = (not options.get('scan')) )
def _get_duration (self, path):
p = subprocess.Popen(['soxi', '-D', path], stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
out, err = p.communicate()
if not err:
return utils.seconds_to_time(int(float(out)))
def get_sound_info (self, program, path):
"""
Parse file name to get info on the assumption it has the correct
format (given in Command.help)
"""
file_name = os.path.basename(path)
file_name = os.path.splitext(file_name)[0]
r = re.search('^(?P<year>[0-9]{4})'
'(?P<month>[0-9]{2})'
'(?P<day>[0-9]{2})'
'(_(?P<n>[0-9]+))?'
'_?(?P<name>.*)$',
file_name)
if not (r and r.groupdict()):
self.report(program, path, "file path is not correct, use defaults")
r = {
'name': file_name
}
else:
r = r.groupdict()
r['duration'] = self._get_duration(path)
r['name'] = r['name'].replace('_', ' ').capitalize()
r['path'] = path
return r
def find_initial (self, program, sound_info):
"""
For a given program, and sound path check if there is an initial
diffusion to associate to, using the diffusion's date.
If there is no matching episode, return None.
"""
# check on episodes
diffusion = Diffusion.objects.filter(
program = program,
date__year = int(sound_info['year']),
date__month = int(sound_info['month']),
date__day = int(sound_info['day'])
)
if not diffusion.count():
self.report(program, sound_info['path'],
'no diffusion found for the given date')
return
return diffusion[0]
@staticmethod
def check_sounds (qs):
"""
Only check for the sound existence or update
"""
# check files
for sound in qs:
if sound.check_on_file():
sound.save(check = False)
def scan (self):
"""
For all programs, scan dirs
"""
print('scan files for all programs...')
programs = Program.objects.filter()
for program in programs:
print('- program', program.name)
self.scan_for_program(
program, settings.AIRCOX_SOUND_ARCHIVES_SUBDIR,
type = Sound.Type['archive'],
)
self.scan_for_program(
program, settings.AIRCOX_SOUND_EXCERPTS_SUBDIR,
type = Sound.Type['excerpt'],
)
def scan_for_program (self, program, subdir, **sound_kwargs):
"""
Scan a given directory that is associated to the given program, and
update sounds information.
"""
print(' - scan files in', subdir)
if not program.ensure_dir(subdir):
return
subdir = os.path.join(program.path, subdir)
# new/existing sounds
for path in os.listdir(subdir):
path = os.path.join(subdir, path)
if not path.endswith(settings.AIRCOX_SOUND_FILE_EXT):
continue
sound_info = self.get_sound_info(program, path)
sound = Sound.objects.get_or_create(
path = path,
defaults = { 'name': sound_info['name'],
'duration': sound_info['duration'] or None }
)[0]
sound.__dict__.update(sound_kwargs)
sound.save(check = False)
# initial diffusion association
if 'year' in sound_info:
initial = self.find_initial(program, sound_info)
if initial:
if initial.initial:
# FIXME: allow user to overwrite rerun info?
self.report(program, path,
'the diffusion must be an initial diffusion')
else:
sound = initial.sounds.get_queryset() \
.filter(path == sound.path)
if not sound:
self.report(program, path,
'add sound to diffusion ', initial.id)
initial.sounds.add(sound)
initial.save()
self.check_sounds(Sound.objects.filter(path__startswith = subdir))
def check_quality (self, check = False):
"""
Check all files where quality has been set to bad
"""
import aircox.programs.management.commands.sounds_quality_check \
as quality_check
sounds = Sound.objects.filter(good_quality = False)
if check:
self.check_sounds(sounds)
files = [ sound.path for sound in sounds if not sound.removed ]
else:
files = [ sound.path for sound in sounds.filter(removed = False) ]
print('start quality check...')
cmd = quality_check.Command()
cmd.handle( files = files,
**settings.AIRCOX_SOUND_QUALITY )
print('- update sounds in database')
def update_stats(sound_info, sound):
stats = sound_info.get_file_stats()
if stats:
duration = int(stats.get('length'))
sound.duration = utils.seconds_to_time(duration)
for sound_info in cmd.good:
sound = Sound.objects.get(path = sound_info.path)
sound.good_quality = True
update_stats(sound_info, sound)
sound.save(check = False)
for sound_info in cmd.bad:
sound = Sound.objects.get(path = sound_info.path)
update_stats(sound_info, sound)
sound.save(check = False)

175
programs/management/commands/sounds_quality_check.py Normal file
View File

@ -0,0 +1,175 @@
"""
Analyse and check files using Sox, prints good and bad files.
"""
import sys
import re
import subprocess
from argparse import RawTextHelpFormatter
from django.core.management.base import BaseCommand, CommandError
class Stats:
attributes = [
'DC offset', 'Min level', 'Max level',
'Pk lev dB', 'RMS lev dB', 'RMS Pk dB',
'RMS Tr dB', 'Flat factor', 'Length s',
]
def __init__ (self, path, **kwargs):
"""
If path is given, call analyse with path and kwargs
"""
self.values = {}
if path:
self.analyse(path, **kwargs)
def get (self, attr):
return self.values.get(attr)
def parse (self, output):
for attr in Stats.attributes:
value = re.search(attr + r'\s+(?P<value>\S+)', output)
value = value and value.groupdict()
if value:
try:
value = float(value.get('value'))
except ValueError:
value = None
self.values[attr] = value
self.values['length'] = self.values['Length s']
def analyse (self, path, at = None, length = None):
"""
If at and length are given use them as excerpt to analyse.
"""
args = ['sox', path, '-n']
if at is not None and length is not None:
args += ['trim', str(at), str(length) ]
args.append('stats')
p = subprocess.Popen(args, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
# sox outputs to stderr (my god WHYYYY)
out_, out = p.communicate()
self.parse(str(out, encoding='utf-8'))
class Sound:
path = None # file path
sample_length = 120 # default sample length in seconds
stats = None # list of samples statistics
bad = None # list of bad samples
good = None # list of good samples
def __init__ (self, path, sample_length = None):
self.path = path
self.sample_length = sample_length if sample_length is not None \
else self.sample_length
def get_file_stats (self):
return self.stats and self.stats[0]
def analyse (self):
print('- complete file analysis')
self.stats = [ Stats(self.path) ]
position = 0
length = self.stats[0].get('length')
if not self.sample_length:
return
print('- samples analysis: ', end=' ')
while position < length:
print(len(self.stats), end=' ')
stats = Stats(self.path, at = position, length = self.sample_length)
self.stats.append(stats)
position += self.sample_length
print()
def check (self, name, min_val, max_val):
self.good = [ index for index, stats in enumerate(self.stats)
if min_val <= stats.get(name) <= max_val ]
self.bad = [ index for index, stats in enumerate(self.stats)
if index not in self.good ]
self.resume()
def resume (self):
view = lambda array: [
'file' if index is 0 else
'sample {} (at {} seconds)'.format(index, (index-1) * self.sample_length)
for index in array
]
if self.good:
print('- good:\033[92m', ', '.join( view(self.good) ), '\033[0m')
if self.bad:
print('- bad:\033[91m', ', '.join( view(self.bad) ), '\033[0m')
class Command (BaseCommand):
help = __doc__
sounds = None
def add_arguments (self, parser):
parser.formatter_class=RawTextHelpFormatter
parser.add_argument(
'files', metavar='FILE', type=str, nargs='+',
help='file(s) to analyse'
)
parser.add_argument(
'-s', '--sample_length', type=int, default=120,
help='size of sample to analyse in seconds. If not set (or 0), does'
' not analyse by sample',
)
parser.add_argument(
'-a', '--attribute', type=str,
help='attribute name to use to check, that can be:\n' + \
', '.join([ '"{}"'.format(attr) for attr in Stats.attributes ])
)
parser.add_argument(
'-r', '--range', type=float, nargs=2,
help='range of minimal and maximal accepted value such as: ' \
'--range min max'
)
parser.add_argument(
'-i', '--resume', action='store_true',
help='print a resume of good and bad files'
)
def handle (self, *args, **options):
# parameters
minmax = options.get('range')
if not minmax:
raise CommandError('no range specified')
attr = options.get('attribute')
if not attr:
raise CommandError('no attribute specified')
# sound analyse and checks
self.sounds = [ Sound(path, options.get('sample_length'))
for path in options.get('files') ]
self.bad = []
self.good = []
for sound in self.sounds:
print(sound.path)
sound.analyse()
sound.check(attr, minmax[0], minmax[1])
print()
if sound.bad:
self.bad.append(sound)
else:
self.good.append(sound)
# resume
if options.get('resume'):
if self.good:
print('files that did not failed the test:\033[92m\n ',
'\n '.join([sound.path for sound in self.good]), '\033[0m')
if self.bad:
# bad at the end for ergonomy
print('files that failed the test:\033[91m\n ',
'\n '.join([sound.path for sound in self.bad]),'\033[0m')

679
programs/models.py Executable file
View File

@ -0,0 +1,679 @@
import os
from django.db import models
from django.template.defaultfilters import slugify
from django.utils.translation import ugettext as _, ugettext_lazy
from django.utils import timezone as tz
from django.utils.html import strip_tags
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from taggit.managers import TaggableManager
import aircox.programs.utils as utils
import aircox.programs.settings as settings
def date_or_default (date, date_only = False):
"""
Return date or default value (now) if not defined, and remove time info
if date_only is True
"""
date = date or tz.datetime.today()
if not tz.is_aware(date):
date = tz.make_aware(date)
if date_only:
return date.replace(hour = 0, minute = 0, second = 0, microsecond = 0)
return date
class Nameable (models.Model):
name = models.CharField (
_('name'),
max_length = 128,
)
@property
def slug (self):
"""
Slug based on the name. We replace '-' by '_'
"""
return slugify(self.name).replace('-', '_')
def __str__ (self):
#if self.pk:
# return '#{} {}'.format(self.pk, self.name)
return '{}'.format(self.name)
class Meta:
abstract = True
class Track (Nameable):
"""
Track of a playlist of a diffusion. The position can either be expressed
as the position in the playlist or as the moment in seconds it started.
"""
# There are no nice solution for M2M relations ship (even without
# through) in django-admin. So we unfortunately need to make one-
# to-one relations and add a position argument
diffusion = models.ForeignKey(
'Diffusion',
)
artist = models.CharField(
_('artist'),
max_length = 128,
)
# position can be used to specify a position in seconds for non-
# stop programs or a position in the playlist
position = models.SmallIntegerField(
default = 0,
help_text=_('position in the playlist'),
)
tags = TaggableManager(
verbose_name=_('tags'),
)
def __str__(self):
return ' '.join([self.artist, ':', self.name ])
class Meta:
verbose_name = _('Track')
verbose_name_plural = _('Tracks')
class Sound (Nameable):
"""
A Sound is the representation of a sound file that can be either an excerpt
or a complete archive of the related diffusion.
The podcasting and public access permissions of a Sound are managed through
the related program info.
"""
Type = {
'other': 0x00,
'archive': 0x01,
'excerpt': 0x02,
}
for key, value in Type.items():
ugettext_lazy(key)
type = models.SmallIntegerField(
verbose_name = _('type'),
choices = [ (y, x) for x,y in Type.items() ],
blank = True, null = True
)
path = models.FilePathField(
_('file'),
path = settings.AIRCOX_PROGRAMS_DIR,
match = r'(' + '|'.join(settings.AIRCOX_SOUND_FILE_EXT) \
.replace('.', r'\.') + ')$',
recursive = True,
blank = True, null = True,
)
embed = models.TextField(
_('embed HTML code'),
blank = True, null = True,
help_text = _('HTML code used to embed a sound from external plateform'),
)
duration = models.TimeField(
_('duration'),
blank = True, null = True,
help_text = _('duration of the sound'),
)
mtime = models.DateTimeField(
_('modification time'),
blank = True, null = True,
help_text = _('last modification date and time'),
)
removed = models.BooleanField(
_('removed'),
default = False,
help_text = _('this sound has been removed from filesystem'),
)
good_quality = models.BooleanField(
_('good quality'),
default = False,
help_text = _('sound\'s quality is okay')
)
public = models.BooleanField(
_('public'),
default = False,
help_text = _('sound\'s is accessible through the website')
)
def get_mtime (self):
"""
Get the last modification date from file
"""
mtime = os.stat(self.path).st_mtime
mtime = tz.datetime.fromtimestamp(mtime)
# db does not store microseconds
mtime = mtime.replace(microsecond = 0)
return tz.make_aware(mtime, tz.get_current_timezone())
def file_exists (self):
"""
Return true if the file still exists
"""
return os.path.exists(self.path)
def check_on_file (self):
"""
Check sound file info again'st self, and update informations if
needed (do not save). Return True if there was changes.
"""
if not self.file_exists():
if self.removed:
return
self.removed = True
return True
old_removed = self.removed
self.removed = False
mtime = self.get_mtime()
if self.mtime != mtime:
self.mtime = mtime
self.good_quality = False
return True
return old_removed != self.removed
def save (self, check = True, *args, **kwargs):
if check:
self.check_on_file()
if not self.name and self.path:
self.name = os.path.basename(self.path) \
.splitext() \
.replace('_', ' ')
super().save(*args, **kwargs)
def __str__ (self):
return '/'.join(self.path.split('/')[-3:])
class Meta:
verbose_name = _('Sound')
verbose_name_plural = _('Sounds')
class Stream (models.Model):
"""
When there are no program scheduled, it is possible to play sounds
in order to avoid blanks. A Stream is a Program that plays this role,
and whose linked to a Stream.
All sounds that are marked as good and that are under the related
program's archive dir are elligible for the sound's selection.
"""
program = models.ForeignKey(
'Program',
verbose_name = _('related program'),
)
delay = models.TimeField(
_('delay'),
blank = True, null = True,
help_text = _('plays this playlist at least every delay')
)
begin = models.TimeField(
_('begin'),
blank = True, null = True,
help_text = _('used to define a time range this stream is'
'played')
)
end = models.TimeField(
_('end'),
blank = True, null = True,
help_text = _('used to define a time range this stream is'
'played')
)
class Schedule (models.Model):
"""
A Schedule defines time slots of programs' diffusions. It can be an initial
run or a rerun (in such case it is linked to the related schedule).
"""
# Frequency for schedules. Basically, it is a mask of bits where each bit is
# a week. Bits > rank 5 are used for special schedules.
# Important: the first week is always the first week where the weekday of
# the schedule is present.
# For ponctual programs, there is no need for a schedule, only a diffusion
Frequency = {
'first': (0b000001, _('first week of the month')),
'second': (0b000010, _('second week of the month')),
'third': (0b000100, _('third week of the month')),
'fourth': (0b001000, _('fourth week of the month')),
'last': (0b010000, _('last week of the month')),
'first and third': (0b000101, _('first and third weeks of the month')),
'second and fourth': (0b001010, _('second and fourth weeks of the month')),
'every': (0b011111, _('once a week')),
'one on two': (0b100000, _('one week on two')),
}
VerboseFrequency = { value[0]: value[1] for key, value in Frequency.items() }
Frequency = { key: value[0] for key, value in Frequency.items() }
program = models.ForeignKey(
'Program',
verbose_name = _('related program'),
)
date = models.DateTimeField(_('date'))
duration = models.TimeField(
_('duration'),
help_text = _('regular duration'),
)
frequency = models.SmallIntegerField(
_('frequency'),
choices = VerboseFrequency.items(),
)
initial = models.ForeignKey(
'self',
verbose_name = _('initial'),
blank = True, null = True,
help_text = 'this schedule is a rerun of this one',
)
def match (self, date = None, check_time = True):
"""
Return True if the given datetime matches the schedule
"""
date = date_or_default(date)
if self.date.weekday() == date.weekday() and self.match_week(date):
return self.date.time() == date.time() if check_time else True
return False
def match_week (self, date = None):
"""
Return True if the given week number matches the schedule, False
otherwise.
If the schedule is ponctual, return None.
"""
# FIXME: does not work if first_day > date_day
date = date_or_default(date)
if self.frequency == Schedule.Frequency['one on two']:
week = date.isocalendar()[1]
return (week % 2) == (self.date.isocalendar()[1] % 2)
first_of_month = date.replace(day = 1)
week = date.isocalendar()[1] - first_of_month.isocalendar()[1]
# weeks of month
if week == 4:
# fifth week: return if for every week
return self.frequency == 0b1111
return (self.frequency & (0b0001 << week) > 0)
def normalize (self, date):
"""
Set the time of a datetime to the schedule's one
"""
return date.replace(hour = self.date.hour, minute = self.date.minute)
def dates_of_month (self, date = None):
"""
Return a list with all matching dates of date.month (=today)
"""
date = date_or_default(date, True).replace(day=1)
fwday = date.weekday()
wday = self.date.weekday()
# move date to the date weekday of the schedule
# check on SO#3284452 for the formula
date += tz.timedelta(days = (7 if fwday > wday else 0) - fwday + wday)
fwday = date.weekday()
# special frequency case
weeks = self.frequency
if self.frequency == Schedule.Frequency['last']:
date += tz.timedelta(month = 1, days = -7)
return self.normalize([date])
if weeks == Schedule.Frequency['one on two']:
# if both week are the same, then the date week of the month
# matches. Note: wday % 2 + fwday % 2 => (wday + fwday) % 2
fweek = date.isocalendar()[1]
if date.month == 1 and fweek >= 50:
# isocalendar can think we are on the last week of the
# previous year
fweek = 0
week = self.date.isocalendar()[1]
weeks = 0b010101 if not (fweek + week) % 2 else 0b001010
dates = []
for week in range(0,5):
# there can be five weeks in a month
if not weeks & (0b1 << week):
continue
wdate = date + tz.timedelta(days = week * 7)
if wdate.month == date.month:
dates.append(self.normalize(wdate))
return dates
def diffusions_of_month (self, date, exclude_saved = False):
"""
Return a list of Diffusion instances, from month of the given date, that
can be not in the database.
If exclude_saved, exclude all diffusions that are yet in the database.
"""
dates = self.dates_of_month(date)
saved = Diffusion.objects.filter(date__in = dates,
program = self.program)
diffusions = []
# existing diffusions
for item in saved:
if item.date in dates:
dates.remove(item.date)
if not exclude_saved:
diffusions.append(item)
# others
for date in dates:
first_date = date
if self.initial:
first_date -= self.date - self.initial.date
first_diffusion = Diffusion.objects.filter(date = first_date,
program = self.program)
first_diffusion = first_diffusion[0] if first_diffusion.count() \
else None
diffusions.append(Diffusion(
program = self.program,
type = Diffusion.Type['unconfirmed'],
initial = first_diffusion if self.initial else None,
date = date,
duration = self.duration,
))
return diffusions
def __str__ (self):
frequency = [ x for x,y in Schedule.Frequency.items()
if y == self.frequency ]
return self.program.name + ': ' + frequency[0] + ' (' + str(self.date) + ')'
class Meta:
verbose_name = _('Schedule')
verbose_name_plural = _('Schedules')
class Station (Nameable):
"""
A Station regroup one or more programs (stream and normal), and is the top
element used to generate streams outputs and configuration.
"""
active = models.BooleanField(
_('active'),
default = True,
help_text = _('this station is active')
)
public = models.BooleanField(
_('public'),
default = True,
help_text = _('information are available to the public'),
)
fallback = models.FilePathField(
_('fallback song'),
match = r'(' + '|'.join(settings.AIRCOX_SOUND_FILE_EXT) \
.replace('.', r'\.') + ')$',
recursive = True,
blank = True, null = True,
help_text = _('use this song file if there is a problem and nothing is '
'played')
)
class Program (Nameable):
"""
A Program can either be a Streamed or a Scheduled program.
A Streamed program is used to generate non-stop random playlists when there
is not scheduled diffusion. In such a case, a Stream is used to describe
diffusion informations.
A Scheduled program has a schedule and is the one with a normal use case.
"""
station = models.ForeignKey(
Station,
verbose_name = _('station')
)
active = models.BooleanField(
_('active'),
default = True,
help_text = _('if not set this program is no longer active')
)
public = models.BooleanField(
_('public'),
default = True,
help_text = _('information are available to the public')
)
@property
def path (self):
"""
Return the path to the programs directory
"""
return os.path.join(settings.AIRCOX_PROGRAMS_DIR,
self.slug + '_' + str(self.id) )
def ensure_dir (self, subdir = None):
"""
Make sur the program's dir exists (and optionally subdir). Return True
if the dir (or subdir) exists.
"""
path = os.path.join(self.path, subdir) if subdir else \
self.path
os.makedirs(path, exist_ok = True)
return os.path.exists(path)
def find_schedule (self, date):
"""
Return the first schedule that matches a given date.
"""
schedules = Schedule.objects.filter(program = self)
for schedule in schedules:
if schedule.match(date, check_time = False):
return schedule
class Diffusion (models.Model):
"""
A Diffusion is an occurrence of a Program that is scheduled on the
station's timetable. It can be a rerun of a previous diffusion. In such
a case, use rerun's info instead of its own.
A Diffusion without any rerun is named Episode (previously, a
Diffusion was different from an Episode, but in the end, an
episode only has a name, a linked program, and a list of sounds, so we
finally merge theme).
A Diffusion can have different types:
- default: simple diffusion that is planified / did occurred
- unconfirmed: a generated diffusion that has not been confirmed and thus
is not yet planified
- cancel: the diffusion has been canceled
- stop: the diffusion has been manually stopped
"""
Type = {
'normal': 0x00, # diffusion is planified
'unconfirmed': 0x01, # scheduled by the generator but not confirmed for diffusion
'cancel': 0x02, # diffusion canceled
}
for key, value in Type.items():
ugettext_lazy(key)
# common
program = models.ForeignKey (
'Program',
verbose_name = _('program'),
)
sounds = models.ManyToManyField(
Sound,
blank = True,
verbose_name = _('sounds'),
)
# specific
type = models.SmallIntegerField(
verbose_name = _('type'),
choices = [ (y, x) for x,y in Type.items() ],
)
initial = models.ForeignKey (
'self',
verbose_name = _('initial'),
blank = True, null = True,
help_text = _('the diffusion is a rerun of this one')
)
date = models.DateTimeField( _('start of the diffusion') )
duration = models.TimeField(
_('duration'),
help_text = _('regular duration'),
)
def archives_duration (self):
"""
Get total duration of the archives. May differ from the schedule
duration.
"""
sounds = self.initial.sounds if self.initial else self.sounds
r = [ sound.duration
for sound in sounds.filter(type = Sound.Type['archive'])
if sound.duration ]
return utils.time_sum(r) if r else self.duration
def get_archives (self):
"""
Return an ordered list of archives sounds for the given episode.
"""
sounds = self.initial.sounds if self.initial else self.sounds
r = [ sound for sound in sounds.all().order_by('path')
if sound.type == Sound.Type['archive'] ]
return r
@classmethod
def get_next (cl, station = None, date = None, **filter_args):
"""
Return a queryset with the upcoming diffusions, ordered by
+date
"""
filter_args['date__gte'] = date_or_default(date)
if station:
filter_args['program__station'] = station
return cl.objects.filter(**filter_args).order_by('date')
@classmethod
def get_prev (cl, station = None, date = None, **filter_args):
"""
Return a queryset with the previous diffusion, ordered by
-date
"""
filter_args['date__lte'] = date_or_default(date)
if station:
filter_args['program__station'] = station
return cl.objects.filter(**filter_args).order_by('-date')
def get_conflicts (self):
"""
Return a list of conflictual diffusions, based on the scheduled duration.
Note: for performance reason, check next and prev are limited to a
certain amount of diffusions.
"""
r = []
# prev
qs = self.get_prev(self.program.station, self.date)
count = 0
for diff in qs:
if diff.pk == self.pk:
continue
end = diff.date + utils.to_timedelta(diff.duration)
if end > self.date:
r.append(diff)
continue
count+=1
if count > 5: break
# next
end = self.date + utils.to_timedelta(self.duration)
qs = self.get_next(self.program.station, self.date)
count = 0
for diff in qs:
if diff.pk == self.pk:
continue
if diff.date < end:
r.append(diff)
continue
count+=1
if count > 5: break
return r
def save (self, *args, **kwargs):
if self.initial:
if self.initial.initial:
self.initial = self.initial.initial
self.program = self.initial.program
super(Diffusion, self).save(*args, **kwargs)
def __str__ (self):
return self.program.name + ', ' + \
self.date.strftime('%Y-%m-%d %H:%M') +\
'' # FIXME str(self.type_display)
class Meta:
verbose_name = _('Diffusion')
verbose_name_plural = _('Diffusions')
permissions = (
('programming', _('edit the diffusion\'s planification')),
)
class Log (models.Model):
"""
Log a played sound start and stop, or a single message
"""
source = models.CharField(
_('source'),
max_length = 64,
help_text = 'source information',
blank = True, null = True,
)
date = models.DateTimeField(
'date',
auto_now_add=True,
)
comment = models.CharField(
max_length = 512,
blank = True, null = True,
)
related_type = models.ForeignKey(
ContentType,
blank = True, null = True,
)
related_id = models.PositiveIntegerField(
blank = True, null = True,
)
related_object = GenericForeignKey(
'related_type', 'related_id',
)
@classmethod
def get_for_related_model (cl, model):
"""
Return a queryset that filter related_type to the given one.
"""
return cl.objects.filter(related_type__pk =
ContentType.objects.get_for_model(model).id)
def print (self):
print(str(self), ':', self.comment or '')
if self.related_object:
print(' - {}: #{}'.format(self.related_type,
self.related_id))
def __str__ (self):
return self.date.strftime('%Y-%m-%d %H:%M') + ', ' + self.source

35
programs/settings.py Executable file
View File

@ -0,0 +1,35 @@
import os
from django.conf import settings
def ensure (key, default):
globals()[key] = getattr(settings, key, default)
# Directory for the programs data
ensure('AIRCOX_PROGRAMS_DIR',
os.path.join(settings.MEDIA_ROOT, 'programs'))
# Default directory for the sounds that not linked to a program
ensure('AIRCOX_SOUND_DEFAULT_DIR',
os.path.join(AIRCOX_PROGRAMS_DIR, 'defaults'))
# Sub directory used for the complete episode sounds
ensure('AIRCOX_SOUND_ARCHIVES_SUBDIR', 'archives')
# Sub directory used for the excerpts of the episode
ensure('AIRCOX_SOUND_EXCERPTS_SUBDIR', 'excerpts')
# Quality attributes passed to sound_quality_check from sounds_monitor
ensure('AIRCOX_SOUND_QUALITY', {
'attribute': 'RMS lev dB',
'range': (-18.0, -8.0),
'sample_length': 120,
}
)
# Extension of sound files
ensure('AIRCOX_SOUND_FILE_EXT',
('.ogg','.flac','.wav','.mp3','.opus'))
# Stream for the scheduled diffusions
ensure('AIRCOX_SCHEDULED_STREAM', 0)

79
programs/tests.py Executable file
View File

@ -0,0 +1,79 @@
import datetime
from django.test import TestCase
from django.utils import timezone as tz
from aircox.programs.models import *
class Programs (TestCase):
def setUp (self):
stream = Stream.objects.get_or_create(
name = 'diffusions',
defaults = { 'type': Stream.Type['schedule'] }
)[0]
Program.objects.create(name = 'source', stream = stream)
Program.objects.create(name = 'microouvert', stream = stream)
self.schedules = {}
self.programs = {}
def test_create_programs_schedules (self):
program = Program.objects.get(name = 'source')
sched_0 = self.create_schedule(program, 'one on two', [
tz.datetime(2015, 10, 2, 18),
tz.datetime(2015, 10, 16, 18),
tz.datetime(2015, 10, 30, 18),
]
)
sched_1 = self.create_schedule(program, 'one on two', [
tz.datetime(2015, 10, 5, 18),
tz.datetime(2015, 10, 19, 18),
],
rerun = sched_0
)
self.programs[program.pk] = program
program = Program.objects.get(name = 'microouvert')
# special case with november first week starting on sunday
sched_2 = self.create_schedule(program, 'first and third', [
tz.datetime(2015, 11, 6, 18),
tz.datetime(2015, 11, 20, 18),
],
date = tz.datetime(2015, 10, 23, 18),
)
def create_schedule (self, program, frequency, dates, date = None, rerun = None):
frequency = Schedule.Frequency[frequency]
schedule = Schedule(
program = program,
frequency = frequency,
date = date or dates[0],
rerun = rerun,
duration = datetime.time(1, 30)
)
print(schedule.__dict__)
schedule.save()
self.schedules[schedule.pk] = (schedule, dates)
return schedule
def test_check_schedule (self):
for schedule, dates in self.schedules:
dates = [ tz.make_aware(date) for date in dates ]
dates.sort()
# dates
dates_ = schedule.dates_of_month(dates[0])
dates_.sort()
self.assertEqual(dates_, dates)
# diffusions
dates_ = schedule.diffusions_of_month(dates[0])
dates_ = [date_.date for date_ in dates_]
dates_.sort()
self.assertEqual(dates_, dates)

30
programs/utils.py Normal file
View File

@ -0,0 +1,30 @@
import datetime
def to_timedelta (time):
"""
Transform a datetime or a time instance to a timedelta,
only using time info
"""
return datetime.timedelta(
hours = time.hour,
minutes = time.minute,
seconds = time.second
)
def seconds_to_time (seconds):
"""
Seconds to datetime.time
"""
minutes, seconds = divmod(seconds, 60)
hours, minutes = divmod(minutes, 60)
return datetime.time(hour = hours, minute = minutes, second = seconds)
def time_sum (times):
"""
Sum up a list of time elements
"""
seconds = sum([ time.hour * 3600 + time.minute * 60 + time.second
for time in times ])
return seconds_to_time(seconds)

16
programs/views.py Executable file
View File

@ -0,0 +1,16 @@
from django.db import models
from django.shortcuts import render
from django.core.serializers.json import DjangoJSONEncoder
from django.utils import timezone, dateformat
from django.views.generic import ListView
from django.views.generic import DetailView
from django.utils.translation import ugettext as _, ugettext_lazy
from aircox.programs.models import *
import aircox.programs.settings
import aircox.programs.utils