nonstop/management/commands/stamina.py

   1 import asyncio
   2 import datetime
   3 import random
   4 import signal
   5
   6 from django.core.management.base import BaseCommand
   7
   8 from emissions.models import Nonstop
   9 from nonstop.models import Track, ScheduledDiffusion, RecurringStreamOccurence
  10
  11
  12 class Command(BaseCommand):
  13     last_jingle_datetime = None
  14     quit = False
  15
  16     def handle(self, verbosity, **kwargs):
  17         try:
  18             asyncio.run(self.main(), debug=True)
  19         except KeyboardInterrupt:
  20             pass
  21
  22     def get_playlist(self, zone, start_datetime, end_datetime):
  23         current_datetime = start_datetime
  24         if self.last_jingle_datetime is None:
  25             self.last_jingle_datetime = current_datetime
  26         playlist = []
  27         adjustment_counter = 0
  28         try:
  29             jingles = list(zone.nonstopzonesettings_set.first().jingles.all())
  30         except AttributeError:
  31             jingles = []
  32
  33         while current_datetime < end_datetime and adjustment_counter < 5:
  34
  35             if jingles and current_datetime - self.last_jingle_datetime > datetime.timedelta(minutes=20):
  36                 # jingle time, every ~20 minutes
  37                 playlist.append(random.choice(jingles))
  38                 self.last_jingle_datetime = current_datetime
  39                 current_datetime = start_datetime + sum(
  40                         [x.duration for x in playlist], datetime.timedelta(seconds=0))
  41
  42             remaining_time = (end_datetime - current_datetime)
  43             track = Track.objects.filter(
  44                     nonstop_zones=zone,
  45                     duration__isnull=False).exclude(
  46                             id__in=[x.id for x in playlist if isinstance(x, Track)]
  47                     ).order_by('?').first()
  48             playlist.append(track)
  49             current_datetime = start_datetime + sum(
  50                     [x.duration for x in playlist], datetime.timedelta(seconds=0))
  51             if current_datetime > end_datetime:
  52                 # last track overshot
  53                 # 1st strategy: remove last track and try to get a track with
  54                 # exact remaining time
  55                 playlist = playlist[:-1]
  56                 track = Track.objects.filter(
  57                         nonstop_zones=zone,
  58                         duration__gte=remaining_time,
  59                         duration__lt=remaining_time + datetime.timedelta(seconds=1)
  60                         ).exclude(
  61                             id__in=[x.id for x in playlist if isinstance(x, Track)]
  62                         ).order_by('?').first()
  63                 if track:
  64                     # found a track
  65                     playlist.append(track)
  66                 else:
  67                     # fallback strategy: didn't find track of expected duration,
  68                     # reduce playlist further
  69                     adjustment_counter += 1
  70                     playlist = playlist[:-1]
  71
  72                 current_datetime = start_datetime + sum(
  73                         [x.duration for x in playlist], datetime.timedelta(seconds=0))
  74
  75         print('computed playlist:')
  76         current_datetime = start_datetime
  77         for track in playlist:
  78             print('   ', current_datetime, track.duration, track.title)
  79             current_datetime += track.duration
  80         print('   ', current_datetime, '---')
  81         print('   adjustment_counter:', adjustment_counter)
  82
  83         return playlist
  84
  85     async def player_process(self, cmd, timeout=None):
  86         self.player = await asyncio.create_subprocess_shell(
  87                 cmd,
  88                 stdout=asyncio.subprocess.PIPE,
  89                 stderr=asyncio.subprocess.PIPE)
  90         if timeout is None:
  91             await self.player.communicate()
  92         else:
  93             try:
  94                 await asyncio.wait_for(self.player.communicate(), timeout=timeout)
  95             except asyncio.TimeoutError:
  96                 pass
  97         self.player = None
  98
  99     async def play(self, slot):
 100         now = datetime.datetime.now()
 101         if isinstance(slot, Nonstop):
 102             self.playlist = self.get_playlist(slot, now, slot.end_datetime)
 103             self.playhead = 0
 104             while True:
 105                 now = datetime.datetime.now()
 106                 try:
 107                     track = self.playlist[self.playhead]
 108                 except IndexError:
 109                     break
 110                 self.current_track_start_datetime = now
 111                 print(now, track.title, track.duration,
 112                         '- future tracks:', [x.title for x in self.playlist[self.playhead + 1:self.playhead + 3]])
 113                 cmd = 'sleep %s # %s' % (track.duration.seconds, track.title)
 114                 await self.player_process(cmd)
 115                 # TODO: detect "soft spot", to switch to another nonstop
 116                 # tranche
 117                 self.playhead += 1
 118         elif slot.is_stream():
 119             print(now, 'playing stream', slot.stream)
 120             # TODO: jingle
 121             cmd = 'sleep 86400 # stream'  # will never stop by itself
 122             print('timeout at', (slot.end_datetime - now).total_seconds())
 123             await self.player_process(cmd, timeout=(slot.end_datetime - now).total_seconds())
 124         else:
 125             print(now, 'playing sound', slot.episode)
 126             # TODO: jingle
 127             cmd = 'sleep %s # %s' % (slot.soundfile.duration, slot.episode)
 128             await self.player_process(cmd)
 129
 130     def recompute_playlist(self):
 131         current_track = self.playlist[self.playhead]
 132         print('recompute_playlist, from', current_track.title, self.current_track_start_datetime + current_track.duration, 'to', self.slot.end_datetime)
 133         playlist = self.get_playlist(self.slot,
 134                 self.current_track_start_datetime + current_track.duration, self.slot.end_datetime)
 135         if playlist:
 136             self.playlist[self.playhead + 1:] = playlist
 137
 138     def get_current_diffusion(self):
 139         now = datetime.datetime.now()
 140         diffusion = ScheduledDiffusion.objects.filter(
 141                 diffusion__datetime__gt=now - datetime.timedelta(days=1),
 142                 diffusion__datetime__lt=now).order_by('diffusion__datetime').last()
 143         occurence = RecurringStreamOccurence.objects.filter(
 144                 datetime__gt=now - datetime.timedelta(days=1),
 145                 datetime__lt=now).order_by('datetime').last()
 146         # note it shouldn't be possible to have both diffusion and occurence
 147         # running at the moment.
 148         if occurence and occurence.end_datetime > now:
 149             return occurence
 150         if diffusion and diffusion.end_datetime > now:
 151             return diffusion
 152         return None
 153
 154     def get_next_diffusion(self, before_datetime):
 155         now = datetime.datetime.now()
 156         diffusion = ScheduledDiffusion.objects.filter(
 157                 diffusion__datetime__gt=now,
 158                 diffusion__datetime__lt=before_datetime,
 159                 ).order_by('diffusion__datetime').first()
 160         occurence = RecurringStreamOccurence.objects.filter(
 161                 datetime__gt=now,
 162                 datetime__lt=before_datetime,
 163                 ).order_by('datetime').first()
 164         if diffusion and occurence:
 165             return diffusion if diffusion.diffusion__datetime < occurence.datetime else occurence
 166         if diffusion:
 167             return diffusion
 168         if occurence:
 169             return occurence
 170         return None
 171
 172     def recompute_slots(self):
 173         now = datetime.datetime.now()
 174         # print(now, 'recompute_slots')
 175         diffusion = self.get_current_diffusion()
 176         if diffusion:
 177             self.slot = diffusion
 178         else:
 179             nonstops = list(Nonstop.objects.all().order_by('start'))
 180             nonstops = [x for x in nonstops if x.start != x.end]  # disabled zones
 181             try:
 182                 self.slot = [x for x in nonstops if x.start < now.time()][-1]
 183             except IndexError:
 184                 self.slot = nonstops[0]
 185             try:
 186                 next_slot = nonstops[nonstops.index(self.slot) + 1]
 187             except IndexError:
 188                 next_slot = nonstops[0]
 189             self.slot.datetime = now.replace(
 190                     hour=self.slot.start.hour,
 191                     minute=self.slot.start.minute)
 192             self.slot.end_datetime = now.replace(
 193                     hour=next_slot.start.hour,
 194                     minute=next_slot.start.minute,
 195                     second=0,
 196                     microsecond=0)
 197             if self.slot.end_datetime < self.slot.datetime:
 198                 self.slot.end_datetime += datetime.timedelta(days=1)
 199
 200             diffusion = self.get_next_diffusion(before_datetime=self.slot.end_datetime)
 201             if diffusion:
 202                 self.slot.end_datetime = diffusion.datetime
 203
 204     async def recompute_slots_loop(self):
 205         now = datetime.datetime.now()
 206         print(now, 'recompute_slots_loop')
 207         sleep = (60 - now.second) % 10  # adjust to awake at :00
 208         while not self.quit:
 209             await asyncio.sleep(sleep)
 210             sleep = 10  # next cycles every 10 seconds
 211             current_slot = self.slot
 212             self.recompute_slots()
 213             expected_slot = self.slot
 214             if current_slot != expected_slot:
 215                 print(now, 'unexpected change', current_slot, 'vs', expected_slot)
 216                 if isinstance(current_slot, Nonstop) and not isinstance(expected_slot, Nonstop):
 217                     # interrupt nonstop
 218                     print('interrupting nonstop')
 219                     self.play_task.cancel()
 220             elif current_slot.end_datetime > expected_slot.end_datetime:
 221                 print('change in end time, from %s to %s' %
 222                         (current_slot.end_datetime, expected_slot.end_datetime))
 223                 if expected_slot.end_datetime - datetime.datetime.now() > datetime.timedelta(minutes=5):
 224                     # more than 5 minutes left, recompute playlist
 225                     self.recompute_playlist()
 226
 227     async def handle_connection(self, reader, writer):
 228         data = await reader.read(100)
 229         message = data.decode().strip()
 230         response = 'err'
 231         if message == 'playing?':
 232             response = '%s' % self.slot
 233         writer.write(response.encode('utf-8'))
 234         await writer.drain()
 235         writer.close()
 236
 237     def sigterm_handler(self):
 238         print('got signal')
 239         self.quit = True
 240         self.play_task.cancel()
 241
 242     async def main(self):
 243         loop = asyncio.get_running_loop()
 244         loop.add_signal_handler(
 245                 signal.SIGTERM,
 246                 self.sigterm_handler)
 247         now = datetime.datetime.now()
 248         self.recompute_slots()
 249         server = await asyncio.start_server(self.handle_connection, '127.0.0.1', 8888)
 250         async with server:
 251             asyncio.create_task(server.serve_forever())
 252
 253             self.recompute_slots_task = asyncio.create_task(self.recompute_slots_loop())
 254             while not self.quit:
 255                 duration = (self.slot.end_datetime - now).seconds
 256                 print('next sure slot', duration, self.slot.end_datetime)
 257                 if duration < 2:
 258                     # next slot is very close, wait for it
 259                     await asyncio.sleep(duration)
 260                     self.recompute_slots()
 261                 self.play_task = asyncio.create_task(self.play(self.slot))
 262                 try:
 263                     await self.play_task
 264                     self.recompute_slots()
 265                 except asyncio.CancelledError as exc:
 266                     print('exc:', exc)
 267                     if self.player and self.player.returncode is None:  # not finished
 268                         self.player.kill()
 269                 except KeyboardInterrupt:
 270                     self.quit = True
 271                     break