rc/aircox
13
1
Fork 1
Code Issues 15 Pull Requests Projects Releases Wiki Activity

changes

Browse Source
This commit is contained in:
bkfox
2015-10-29 11:01:15 +01:00
parent e2696b7322
commit c3b5104f69
13 changed files with 289 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
aircox_programs/management/commands/diffusions_monitor.py
View File

@ -14,10 +14,10 @@ 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 *
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:
@ -52,7 +52,7 @@ class Actions:
if schedule.match(diffusion.date):
break
else:
print('> #{}: {}'.format(diffusion.date, str(diffusion)))
print('> #{}: {}'.format(diffusion.pk, str(diffusion)))
items.append(diffusion.id)
print('{} diffusions will be removed'.format(len(items)))

70
aircox_programs/management/commands/sounds_monitor.py
View File

@ -42,22 +42,22 @@ class Command (BaseCommand):
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'
)
def handle (self, *args, **options):
programs = Program.objects.filter()
for program in programs:
path = lambda x: os.path.join(program.path, x)
self.check_files(
program, path(settings.AIRCOX_SOUND_ARCHIVES_SUBDIR),
archive = True,
)
self.check_files(
program, path(settings.AIRCOX_SOUND_EXCERPTS_SUBDIR),
excerpt = True,
)
self.check_quality()
if options.get('scan'):
self.scan()
if options.get('quality_check'):
self.check_quality()
def get_sound_info (self, path):
"""
@ -111,12 +111,28 @@ class Command (BaseCommand):
diffusion = diffusion[0]
return diffusion.episode or None
def scan (self):
print('scan files for all programs...')
programs = Program.objects.filter()
def check_files (self, program, dir_path, **sound_kwargs):
for program in programs:
print('- program ', program.name)
path = lambda x: os.path.join(program.path, x)
self.scan_for_program(
program, path(settings.AIRCOX_SOUND_ARCHIVES_SUBDIR),
type = Sound.Type['archive'],
)
self.scan_for_program(
program, path(settings.AIRCOX_SOUND_EXCERPTS_SUBDIR),
type = Sound.Type['excerpt'],
)
def scan_for_program (self, program, dir_path, **sound_kwargs):
"""
Scan a given directory that is associated to the given program, and
update sounds information.
"""
print(' - scan files in', dir_path)
if not os.path.exists(dir_path):
return
@ -154,5 +170,29 @@ class Command (BaseCommand):
"""
Check all files where quality has been set to bad
"""
import sound_quality_check as quality_check
print('start quality check...')
files = [ sound.path
for sound in Sound.objects.filter(good_quality = False) ]
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:
sound.duration = int(stats.get('length'))
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)

47
aircox_programs/management/commands/sounds_quality_check.py
View File

@ -66,10 +66,14 @@ class Sound:
def __init__ (self, path, sample_length = None):
self.path = path
self.sample_length = sample_length or self.sample_length
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')
print('- complete file analysis')
self.stats = [ Stats(self.path) ]
position = 0
length = self.stats[0].get('length')
@ -77,7 +81,7 @@ class Sound:
if not self.sample_length:
return
print('- Samples analysis: ', end=' ')
print('- samples analysis: ', end=' ')
while position < length:
print(len(self.stats), end=' ')
stats = Stats(self.path, at = position, length = self.sample_length)
@ -85,18 +89,6 @@ class Sound:
position += self.sample_length
print()
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 self.good
]
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')
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 ]
@ -104,6 +96,17 @@ class Sound:
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__
@ -117,7 +120,7 @@ class Command (BaseCommand):
help='file(s) to analyse'
)
parser.add_argument(
'-s', '--sample_length', type=int,
'-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',
)
@ -163,11 +166,11 @@ class Command (BaseCommand):
# resume
if options.get('resume'):
if good:
print('Files that did not failed the test:\033[92m\n ',
'\n '.join(good), '\033[0m')
if bad:
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(bad),'\033[0m')
print('files that failed the test:\033[91m\n ',
'\n '.join([sound.path for sound in self.bad]),'\033[0m')

112
aircox_programs/models.py
View File

@ -74,47 +74,60 @@ class Track (Nameable):
class Sound (Nameable):
"""
A Sound is the representation of a sound, that can be:
- An episode podcast/complete record
- An episode partial podcast
- An episode is a part of the episode but not usable for direct podcast
A Sound is the representation of a sound file that can be either an excerpt
or a complete archive of the related episode.
We can manage this using the "public" and "fragment" fields. If a Sound is
public, then we can podcast it. If a Sound is a fragment, then it is not
usable for diffusion.
Each sound can be associated to a filesystem's file or an embedded
code (for external podcasts).
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 = '*(' + '|'.join(settings.AIRCOX_SOUNDFILE_EXT) + ')$',
match = '*(' + '|'.join(settings.AIRCOX_SOUND_FILE_EXT) + ')$',
recursive = True,
blank = True, null = True,
)
embed = models.TextField(
_('embed HTML code from external website'),
_('embed HTML code'),
blank = True, null = True,
help_text = _('if set, consider the sound podcastable'),
help_text = _('HTML code used to embed a sound from external plateform'),
)
duration = models.TimeField(
_('duration'),
blank = True, null = True,
)
date = models.DateTimeField(
_('date'),
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 = _("the element is public"),
)
fragment = models.BooleanField(
_('incomplete sound'),
default = False,
help_text = _("the file is a cut"),
)
removed = models.BooleanField(
default = False,
help_text = _('this sound has been removed from filesystem'),
help_text = _('sound\'s is accessible through the website')
)
def get_mtime (self):
@ -125,10 +138,34 @@ class Sound (Nameable):
mtime = tz.datetime.fromtimestamp(mtime)
return tz.make_aware(mtime, timezone.get_current_timezone())
def save (self, *args, **kwargs):
if not self.pk:
self.date = self.get_mtime()
def file_exists (self):
return os.path.exists(self.path)
def check_on_file (self):
"""
Check sound file info again'st self, and update informations if
needed. Return True if there was changes.
"""
if not self.file_exists():
if self.removed:
return
self.removed = True
return True
mtime = self.get_mtime()
if self.date != mtime:
self.date = mtime
self.good_quality = False
return True
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):
@ -194,12 +231,13 @@ class Schedule (models.Model):
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 = tz.datetime.date(date.year, date.month, 1)
first_of_month = date.replace(day = 1)
week = date.isocalendar()[1] - first_of_month.isocalendar()[1]
# weeks of month
@ -359,8 +397,7 @@ class Stream (models.Model):
name = models.CharField(
_('name'),
max_length = 32,
blank = True,
null = True,
blank = True, null = True,
)
public = models.BooleanField(
_('public'),
@ -376,13 +413,22 @@ class Stream (models.Model):
default = 0,
help_text = _('priority of the stream')
)
time_start = models.TimeField(
_('start'),
blank = True, null = True,
help_text = _('if random, used to define a time range this stream is'
'played')
)
time_end = models.TimeField(
_('end'),
blank = True, null = True,
help_text = _('if random, used to define a time range this stream is'
'played')
)
# get info for:
# - random lists
# - scheduled lists
# link between Streams and Programs:
# - hours range (non-stop)
# - stream/pgm
def __str__ (self):
return '#{} {}'.format(self.priority, self.name)
@ -391,7 +437,7 @@ class Stream (models.Model):
class Program (Nameable):
stream = models.ForeignKey(
Stream,
verbose_name = _('stream'),
verbose_name = _('streams'),
)
active = models.BooleanField(
_('inactive'),

21
aircox_programs/settings.py
View File

@ -10,15 +10,28 @@ def ensure (key, default):
ensure('AIRCOX_PROGRAMS_DIR',
os.path.join(settings.MEDIA_ROOT, 'programs'))
# Default directory for the sounds
ensure('AIRCOX_SOUNDFILE_DEFAULT_DIR',
os.path.join(AIRCOX_PROGRAMS_DIR + 'default'))
# 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_SOUNDFILE_EXT',
ensure('AIRCOX_SOUND_FILE_EXT',
('.ogg','.flac','.wav','.mp3','.opus'))
# Stream for the scheduled diffusions
ensure('AIRCOX_SCHEDULED_STREAM', 0)

82
aircox_programs/tests.py
View File

@ -1,3 +1,81 @@
from django.test import TestCase
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)
#Stream.objects.create(name = 'bns', type = Stream.Type['random'], priority = 1)
#Stream.objects.create(name = 'jingle', type = Stream.Type['random'] priority = 2)
#Stream.objects.create(name = 'loves', type = Stream.Type['random'], priority = 3)
pass
def test_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
)
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.check_schedule(schedule, dates)
return schedule
def check_schedule (self, schedule, dates):
dates = [ tz.make_aware(date) for date in dates ]
dates.sort()
# match date and weeks
for date in dates:
#self.assertTrue(schedule.match(date, check_time = False))
#self.assertTrue(schedule.match_week(date))
# 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)
# Create your tests here.

74
aircox_programs/views.py
View File

@ -12,79 +12,5 @@ import aircox_programs.settings
import aircox_programs.utils
class ListQueries:
@staticmethod
def search (qs, q):
qs = qs.filter(tags__slug__in = re.compile(r'(\s|\+)+').split(q)) | \
qs.filter(title__icontains = q) | \
qs.filter(subtitle__icontains = q) | \
qs.filter(content__icontains = q)
qs.distinct()
return qs
@staticmethod
def thread (qs, q):
return qs.filter(parent = q)
@staticmethod
def next (qs, q):
qs = qs.filter(date__gte = timezone.now())
if q:
qs = qs.filter(parent = q)
return qs
@staticmethod
def prev (qs, q):
qs = qs.filter(date__lte = timezone.now())
if q:
qs = qs.filter(parent = q)
return qs
@staticmethod
def date (qs, q):
if not q:
q = timezone.datetime.today()
if type(q) is str:
q = timezone.datetime.strptime(q, '%Y/%m/%d').date()
return qs.filter(date__startswith = q)
class Diffusion:
@staticmethod
def episode (qs, q):
return qs.filter(episode = q)
@staticmethod
def program (qs, q):
return qs.filter(program = q)
class ListQuery:
model = None
qs = None
def __init__ (self, model, *kwargs):
self.model = model
self.__dict__.update(kwargs)
def get_queryset (self, by, q):
qs = model.objects.all()
if model._meta.get_field_by_name('public'):
qs = qs.filter(public = True)
# run query set
queries = Queries.__dict__.get(self.model) or Queries
filter = queries.__dict__.get(by)
if filter:
qs = filter(qs, q)
# order
if self.sort == 'asc':
qs = qs.order_by('date', 'id')
else:
qs = qs.order_by('-date', '-id')
# exclude
qs = qs.exclude(id = exclude)
return qs