nonstop/management/commands/stamina.py

   1 import asyncio
   2 import datetime
   3 import json
   4 import logging
   5 import random
   6 import signal
   7 import sys
   8 import time
   9 import urllib.parse
  10
  11 import django.db
  12 import requests
  13 from django.conf import settings
  14 from django.core.management.base import BaseCommand
  15 from emissions.models import Nonstop
  16
  17 from nonstop.app_settings import app_settings
  18 from nonstop.models import (
  19     Jingle,
  20     NonstopZoneSettings,
  21     RecurringRandomDirectoryOccurence,
  22     RecurringStreamOccurence,
  23     ScheduledDiffusion,
  24     SomaLogLine,
  25     Track,
  26 )
  27 from nonstop.utils import Tracklist
  28
  29 logger = logging.getLogger('stamina')
  30
  31
  32 class Command(BaseCommand):
  33     requires_system_checks = False
  34
  35     last_jingle_datetime = None
  36     quit = False
  37
  38     def handle(self, verbosity, **kwargs):
  39         alert_index = 0
  40         latest_alert_timestamp = 0
  41         latest_exception_timestamp = 0
  42
  43         def exception_alert_thresholds():
  44             yield 0
  45             duration = 3
  46             while duration < 3600:
  47                 yield duration
  48                 duration *= 5
  49             duration = 3600
  50             while True:
  51                 yield duration
  52                 duration += 3600
  53
  54         while True:
  55             try:
  56                 asyncio.run(self.main(), debug=settings.DEBUG)
  57             except KeyboardInterrupt:
  58                 break
  59             except Exception as e:
  60                 timestamp = time.time()
  61                 if (timestamp - latest_exception_timestamp) > 300:
  62                     # if latest exception was a "long" time ago, assume
  63                     # things went smooth for a while and reset things
  64                     alert_index = 0
  65                     latest_alert_timestamp = 0
  66                     latest_exception_timestamp = 0
  67
  68                 alert_threshold = 0
  69                 for i, threshold in enumerate(exception_alert_thresholds()):
  70                     if i == alert_index:
  71                         alert_threshold = threshold
  72                         break
  73
  74                 if (timestamp - latest_alert_timestamp) > alert_threshold:
  75                     logger.exception('General exception (alert index: %s)', alert_index)
  76                     latest_alert_timestamp = timestamp
  77                     alert_index += 1
  78
  79                 if alert_index and isinstance(e, django.db.InterfaceError):
  80                     # most likely "connection already closed", because postgresql
  81                     # is been restarted; log then get out to be restarted.
  82                     logger.error('Aborting on repeated database error')
  83                     break
  84
  85                 time.sleep(2)  # retry after a bit
  86                 latest_exception_timestamp = timestamp
  87                 continue
  88             break
  89
  90     def get_playlist(self, zone, start_datetime, end_datetime):
  91         current_datetime = start_datetime() if callable(start_datetime) else start_datetime
  92         if self.last_jingle_datetime is None:
  93             self.last_jingle_datetime = current_datetime
  94         # Define a max duration (1 hour), if it is reached, and far enough
  95         # from end_datetime (30 minutes), return the playlist as is, not aligned
  96         # on end time, so a new playlist gets computed once it's over.
  97         # This avoids misalignments due to track durations not matching exactly
  98         # or additional delays caused by the player program.
  99         max_duration = datetime.timedelta(hours=1)
 100         max_duration_leftover = datetime.timedelta(minutes=30)
 101         playlist = []
 102         adjustment_counter = 0
 103         try:
 104             zone_settings = zone.nonstopzonesettings_set.first()
 105             jingles = list(zone_settings.jingles.all())
 106         except AttributeError:
 107             zone_settings = NonstopZoneSettings()
 108             jingles = []
 109
 110         all_clocked_jingles = Jingle.objects.exclude(clock_time__isnull=True)
 111
 112         zone_ids = [zone.id]
 113         zone_ids.extend([x.id for x in zone_settings.extra_zones.all()])
 114         extra_zones = app_settings.EXTRA_ZONES.get(zone.slug)
 115         if extra_zones:
 116             zone_ids.extend([x.id for x in Nonstop.objects.filter(slug__in=extra_zones)])
 117
 118         recent_tracks_id = [
 119             x.track_id
 120             for x in SomaLogLine.objects.exclude(on_air=False).filter(
 121                 track__isnull=False,
 122                 play_timestamp__gt=datetime.datetime.now()
 123                 - datetime.timedelta(days=app_settings.NO_REPEAT_DELAY),
 124             )
 125         ]
 126
 127         tracklist = Tracklist(zone_settings, zone_ids, recent_tracks_id)
 128         random_tracks_iterator = tracklist.get_random_tracks()
 129         t0 = datetime.datetime.now()
 130         allow_overflow = False
 131         if callable(start_datetime):
 132             # compute start_datetime (e.g. now()) at the last moment, to get
 133             # computed playlist timestamps as close as possible as future real
 134             # ones.
 135             start_datetime = start_datetime()
 136         while current_datetime < end_datetime:
 137             if (current_datetime - start_datetime) > max_duration and (
 138                 (end_datetime - current_datetime) > max_duration_leftover
 139             ):
 140                 break
 141
 142             if zone_settings.intro_jingle and (current_datetime.hour, current_datetime.minute) == (
 143                 zone.start.hour,
 144                 zone.start.minute,
 145             ):
 146                 tracklist.playlist.append(zone_settings.intro_jingle)
 147                 self.last_jingle_datetime = current_datetime
 148                 current_datetime = start_datetime + tracklist.get_duration()
 149             elif jingles and current_datetime - self.last_jingle_datetime > datetime.timedelta(minutes=20):
 150                 # jingle time, every ~20 minutes
 151                 # maybe there's a dedicated jingle for this time of day?
 152                 current_minute = current_datetime.time().replace(second=0, microsecond=0)
 153                 next_minute = (
 154                     (current_datetime + datetime.timedelta(minutes=1)).time().replace(second=0, microsecond=0)
 155                 )
 156                 clocked_jingles = [
 157                     x
 158                     for x in all_clocked_jingles
 159                     if x.clock_time >= current_minute and x.clock_time < next_minute
 160                 ]
 161                 if clocked_jingles:
 162                     clocked_jingle = random.choice(clocked_jingles)
 163                     clocked_jingle.label = '⏰ %s' % clocked_jingle.label
 164                     tracklist.playlist.append(clocked_jingle)
 165                 else:
 166                     tracklist.playlist.append(random.choice(jingles))
 167                 self.last_jingle_datetime = current_datetime
 168                 current_datetime = start_datetime + tracklist.get_duration()
 169             remaining_time = end_datetime - current_datetime
 170
 171             track = next(random_tracks_iterator)
 172             tracklist.append(track)
 173             current_datetime = start_datetime + tracklist.get_duration()
 174             if current_datetime > end_datetime and not allow_overflow:
 175                 # last track overshot
 176                 # 1st strategy: remove last track and try to get a track with
 177                 # exact remaining time
 178                 logger.debug('Overshoot %s, %s', adjustment_counter, current_datetime)
 179                 tracklist.pop()
 180                 try:
 181                     track = next(
 182                         tracklist.get_random_tracks(
 183                             k=1,
 184                             extra_filters={
 185                                 'duration__gte': remaining_time,
 186                                 'duration__lt': remaining_time + datetime.timedelta(seconds=1),
 187                             },
 188                         )
 189                     )
 190                 except StopIteration:  # nothing
 191                     track = None
 192                 if track:
 193                     # found a track
 194                     tracklist.append(track)
 195                 else:
 196                     # fallback strategy: didn't find track of expected duration,
 197                     # reduce playlist further
 198                     adjustment_counter += 1
 199                     if tracklist.pop() is None or adjustment_counter > 5:
 200                         # a dedicated sound that ended a bit too early,
 201                         # or too many failures to get an appropriate file,
 202                         # allow whatever comes.
 203                         allow_overflow = True
 204                         logger.debug('Allowing overflows')
 205
 206                 current_datetime = start_datetime + tracklist.get_duration()
 207
 208         logger.info(
 209             'Computed playlist for "%s" (computation time: %ss)', zone, (datetime.datetime.now() - t0)
 210         )
 211         current_datetime = start_datetime
 212         for track in tracklist.playlist:
 213             logger.debug('- track: %s %s %s', current_datetime, track.duration, track.title)
 214             current_datetime += track.duration
 215         logger.debug('- end: %s', current_datetime)
 216         return tracklist.playlist
 217
 218     def is_nonstop_on_air(self):
 219         # check if nonstop system is currently on air
 220         if app_settings.ON_AIR_SWITCH_URL is None:
 221             return None
 222         switch_response = requests.get(app_settings.ON_AIR_SWITCH_URL, timeout=5)
 223         if not switch_response.ok:
 224             return None
 225         try:
 226             status = switch_response.json()
 227         except ValueError:
 228             return None
 229         if status.get('active') == 0:
 230             return True
 231         elif status.get('active') == 1 and status.get('nonstop-via-stud1') == 0:
 232             # TODO: replace this hardware check that no longer works by a
 233             # logical check on programs: if there's nothing scheduled at the
 234             # moment, consider nonstop is broadcasted, even if studio1 is on.
 235             return True
 236         elif status.get('active') == 2 and status.get('nonstop-via-stud2') == 1:
 237             return True
 238         return False
 239
 240     async def record_nonstop_line(self, track, now):
 241         log_line = SomaLogLine()
 242         log_line.play_timestamp = now
 243         log_line.track = track
 244         log_line.filepath = track.nonstopfile_set.first()
 245         log_line.on_air = self.is_nonstop_on_air()
 246         log_line.save()
 247
 248     async def player_process(self, item, timeout=None):
 249         if app_settings.PLAYER_IPC_PATH:
 250             return await self.player_process_ipc(item, timeout=timeout)
 251         cmd = [app_settings.PLAYER_COMMAND] + app_settings.PLAYER_ARGS
 252         if hasattr(item, 'is_stream') and item.is_stream():
 253             if urllib.parse.urlparse(item.stream.url).scheme == 'mumble':
 254                 cmd = [app_settings.MUMBLE_PLAYER_COMMAND]
 255             cmd.append(item.stream.url)
 256             logger.info('Play stream: %s', item.stream.url)
 257         else:
 258             cmd.append(item.file_path())
 259             logger.info('Play file: %s', item.file_path())
 260         logger.debug('cmd %r', cmd)
 261         self.player = await asyncio.create_subprocess_exec(
 262             *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
 263         )
 264         if timeout is None:
 265             await self.player.communicate()
 266         else:
 267             try:
 268                 await asyncio.wait_for(self.player.communicate(), timeout=timeout)
 269             except asyncio.TimeoutError:
 270                 self.player.kill()
 271         self.player = None
 272
 273     async def player_process_ipc(self, item, timeout=None):
 274         starting = False
 275         while True:
 276             try:
 277                 reader, writer = await asyncio.open_unix_connection(app_settings.PLAYER_IPC_PATH)
 278                 break
 279             except (FileNotFoundError, ConnectionRefusedError):
 280                 if not starting:
 281                     cmd = [app_settings.PLAYER_COMMAND] + app_settings.PLAYER_ARGS
 282                     cmd += ['--input-ipc-server=%s' % app_settings.PLAYER_IPC_PATH, '--idle']
 283                     self.player = await asyncio.create_subprocess_exec(
 284                         *cmd, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL
 285                     )
 286                     starting = True
 287                 await asyncio.sleep(0.1)
 288
 289         if hasattr(item, 'is_stream') and item.is_stream():
 290             file_path = item.stream.url
 291             logger.info('Play stream: %s', item.stream.url)
 292         else:
 293             file_path = item.file_path()
 294             logger.info('Play file: %s', item.file_path())
 295
 296         writer.write(json.dumps({'command': ['loadfile', file_path]}).encode() + b'\n')
 297         try:
 298             await writer.drain()
 299         except ConnectionResetError:  # connection lost
 300             return
 301         try:
 302             await asyncio.wait_for(self.player_ipc_idle(reader, writer), timeout=timeout)
 303         except asyncio.TimeoutError:
 304             pass
 305         writer.close()
 306         await writer.wait_closed()
 307
 308     async def player_ipc_idle(self, reader, writer):
 309         while True:
 310             data = await reader.readline()
 311             if not data:
 312                 break
 313             if json.loads(data) == {'event': 'idle'}:
 314                 break
 315
 316     async def play(self, slot):
 317         now = datetime.datetime.now()
 318         if isinstance(slot, Nonstop):
 319             self.playlist = self.get_playlist(slot, datetime.datetime.now, slot.end_datetime)
 320             self.playhead = 0
 321             self.softstop = False
 322             while not self.quit:
 323                 now = datetime.datetime.now()
 324                 try:
 325                     track = self.playlist[self.playhead]
 326                 except IndexError:
 327                     break
 328                 self.current_track_start_datetime = now
 329                 if isinstance(track, Jingle):
 330                     logger.info('Jingle: %s (id: %s) (%s)', track.title, track.id, track.duration)
 331                 else:
 332                     logger.info('Track: %s (id: %s) (%s)', track.title, track.id, track.duration)
 333                 record_task = None
 334                 if isinstance(track, Track):  # not jingles
 335                     record_task = asyncio.create_task(
 336                         self.record_nonstop_line(track, datetime.datetime.now())
 337                     )
 338                 await self.player_process(track)
 339                 if record_task:
 340                     await record_task
 341                 if self.softstop:
 342                     # track was left to finish, but now the playlist should stop.
 343                     break
 344                 self.playhead += 1
 345         elif slot.is_stream():
 346             logger.info('Stream: %s', slot.stream)
 347             if slot.jingle_id:
 348                 await self.player_process(slot.jingle, timeout=60)
 349             logger.debug('Stream timeout: %s', (slot.end_datetime - now).total_seconds())
 350             short_interruption_counter = 0
 351             has_played = False
 352             while True:
 353                 player_start_time = datetime.datetime.now()
 354                 await self.player_process(
 355                     slot, timeout=(slot.end_datetime - player_start_time).total_seconds()
 356                 )
 357                 now = datetime.datetime.now()
 358                 if (slot.end_datetime - now).total_seconds() < 2:
 359                     # it went well, stop
 360                     break
 361                 # stream got interrupted
 362                 if (datetime.datetime.now() - player_start_time).total_seconds() < 15:
 363                     # and was up for less than 15 seconds.
 364                     if not has_played:
 365                         # never up before, probably not even started
 366                         if isinstance(slot, RecurringStreamOccurence):
 367                             # no mercy for recurring stream, remove occurence
 368                             logger.info('Missing stream for %s, removing', slot)
 369                             slot.delete()
 370                         elif slot.auto_delayed is True:
 371                             # was already delayed and is still not up, remove.
 372                             logger.info('Still missing stream for %s, removing', slot)
 373                             slot.delete()
 374                         else:
 375                             # push back start datetime for 5 minutes, and get
 376                             # back to nonstop music in the meantime
 377                             logger.info('Pushing starting time of %s', slot.diffusion.episode)
 378                             slot.diffusion.datetime = slot.diffusion.datetime + datetime.timedelta(
 379                                 seconds=300
 380                             )
 381                             slot.diffusion.episode.duration = slot.diffusion.episode.get_duration() - 5
 382                             if slot.diffusion.episode.duration <= 5:
 383                                 slot.diffusion.episode.duration = 0
 384                             slot.auto_delayed = True
 385                             slot.diffusion.save()
 386                             slot.diffusion.episode.save()
 387                             slot.save()
 388                         break
 389                     short_interruption_counter += 1
 390                     # wait a bit
 391                     await asyncio.sleep(short_interruption_counter)
 392                 else:
 393                     # mark stream as ok at least one, and reset short
 394                     # interruption counter
 395                     has_played = True
 396                     short_interruption_counter = 0
 397                     logger.debug('Stream error for %s', slot)
 398
 399                 if short_interruption_counter > 5:
 400                     # many short interruptions
 401                     logger.info('Too many stream errors for %s, removing', slot)
 402                     slot.delete()
 403                     break
 404         else:
 405             if hasattr(slot, 'episode'):
 406                 logger.info('Episode: %s (id: %s)', slot.episode, slot.episode.id)
 407             else:
 408                 logger.info('Random: %s', slot)
 409             if slot.jingle_id:
 410                 await self.player_process(slot.jingle, timeout=60)
 411             await self.player_process(slot)
 412
 413     def recompute_playlist(self):
 414         current_track = self.playlist[self.playhead]
 415         logger.debug(
 416             'Recomputing playlist at %s, from %s to %s',
 417             current_track.title,
 418             self.current_track_start_datetime + current_track.duration,
 419             self.slot.end_datetime,
 420         )
 421         playlist = self.get_playlist(
 422             self.slot, self.current_track_start_datetime + current_track.duration, self.slot.end_datetime
 423         )
 424         if playlist:
 425             self.playlist[self.playhead + 1 :] = playlist
 426
 427     def get_current_diffusion(self):
 428         now = datetime.datetime.now()
 429         diffusion = (
 430             ScheduledDiffusion.objects.filter(
 431                 diffusion__datetime__gt=now - datetime.timedelta(days=1), diffusion__datetime__lt=now
 432             )
 433             .order_by('diffusion__datetime')
 434             .last()
 435         )
 436         occurence = (
 437             RecurringStreamOccurence.objects.filter(
 438                 datetime__gt=now - datetime.timedelta(days=1),
 439                 datetime__lt=now,
 440                 diffusion__is_active=True,
 441             )
 442             .order_by('datetime')
 443             .last()
 444         )
 445         directory_occurence = (
 446             RecurringRandomDirectoryOccurence.objects.filter(
 447                 datetime__gt=now - datetime.timedelta(days=1), datetime__lt=now
 448             )
 449             .order_by('datetime')
 450             .last()
 451         )
 452
 453         # factor in some tolerance for diffusions a bit shorter than known, so
 454         # they are not replayed from the start if they finished too early.
 455         tolerance = datetime.timedelta(seconds=60)
 456         if diffusion and hasattr(diffusion, 'is_stream') and diffusion.is_stream():
 457             # unless it's a stream
 458             tolerance = datetime.timedelta(seconds=0)
 459
 460         # note it shouldn't be possible to have both diffusion and occurences
 461         # running at the moment.
 462         if occurence and occurence.end_datetime > now:
 463             return occurence
 464         if diffusion and (diffusion.end_datetime - tolerance) > now:
 465             return diffusion
 466         if directory_occurence and directory_occurence.end_datetime > now:
 467             return directory_occurence
 468         return None
 469
 470     def get_next_diffusion(self, before_datetime):
 471         now = datetime.datetime.now()
 472         diffusion = (
 473             ScheduledDiffusion.objects.filter(
 474                 diffusion__datetime__gt=now,
 475                 diffusion__datetime__lt=before_datetime,
 476             )
 477             .order_by('diffusion__datetime')
 478             .first()
 479         )
 480         occurence = (
 481             RecurringStreamOccurence.objects.filter(
 482                 datetime__gt=now,
 483                 datetime__lt=before_datetime,
 484                 diffusion__is_active=True,
 485             )
 486             .order_by('datetime')
 487             .first()
 488         )
 489         directory_occurence = (
 490             RecurringRandomDirectoryOccurence.objects.filter(
 491                 datetime__gt=now,
 492                 datetime__lt=before_datetime,
 493             )
 494             .order_by('datetime')
 495             .first()
 496         )
 497         if diffusion and occurence:
 498             return diffusion if diffusion.diffusion.datetime < occurence.datetime else occurence
 499         if diffusion:
 500             return diffusion
 501         if occurence:
 502             return occurence
 503         if directory_occurence:
 504             return directory_occurence
 505         return None
 506
 507     def recompute_slots(self):
 508         now = datetime.datetime.now()
 509         diffusion = self.get_current_diffusion()
 510         if diffusion:
 511             self.slot = diffusion
 512         else:
 513             nonstops = list(Nonstop.objects.all().order_by('start'))
 514             nonstops = [x for x in nonstops if x.start != x.end]  # disabled zones
 515             try:
 516                 self.slot = [x for x in nonstops if x.start < now.time()][-1]
 517             except IndexError:
 518                 # no slots starting at midnight, and time is midnight, get latest zone,
 519                 # as it will span midnight.
 520                 self.slot = nonstops[-1]
 521             try:
 522                 next_slot = nonstops[nonstops.index(self.slot) + 1]
 523             except IndexError:
 524                 next_slot = nonstops[0]
 525             self.slot.datetime = now.replace(hour=self.slot.start.hour, minute=self.slot.start.minute)
 526             self.slot.end_datetime = now.replace(
 527                 hour=next_slot.start.hour, minute=next_slot.start.minute, second=0, microsecond=0
 528             )
 529             if self.slot.end_datetime < self.slot.datetime:
 530                 self.slot.end_datetime += datetime.timedelta(days=1)
 531
 532             diffusion = self.get_next_diffusion(before_datetime=self.slot.end_datetime)
 533             if diffusion:
 534                 self.slot.end_datetime = diffusion.datetime
 535
 536     async def recompute_slots_loop(self):
 537         now = datetime.datetime.now()
 538         sleep = (60 - now.second) % 10  # adjust to awake at :00
 539         i = 0
 540         while not self.quit:
 541             await asyncio.sleep(sleep)
 542             sleep = 10  # next cycles every 10 seconds
 543             current_slot = self.slot
 544             self.recompute_slots()
 545             expected_slot = self.slot
 546             if current_slot != expected_slot:
 547                 now = datetime.datetime.now()
 548                 logger.info('Unexpected change, %s vs %s', current_slot, expected_slot)
 549                 if isinstance(current_slot, Nonstop) and isinstance(expected_slot, Nonstop):
 550                     # ask for a softstop, i.e. finish the track then switch.
 551                     self.softstop = True
 552                 elif isinstance(current_slot, Nonstop):
 553                     # interrupt nonstop
 554                     logger.info('Interrupting nonstop')
 555                     self.play_task.cancel()
 556                 elif current_slot.is_stream():
 557                     # it should have been stopped by timeout set on player but
 558                     # maybe the episode duration has been shortened after its
 559                     # start.
 560                     logger.info('Interrupting stream')
 561                     self.play_task.cancel()
 562             elif current_slot.end_datetime > expected_slot.end_datetime:
 563                 now = datetime.datetime.now()
 564                 logger.debug(
 565                     'Change in end time, from %s to %s', current_slot.end_datetime, expected_slot.end_datetime
 566                 )
 567                 if isinstance(current_slot, Nonstop) and (
 568                     expected_slot.end_datetime - datetime.datetime.now() > datetime.timedelta(minutes=5)
 569                 ):
 570                     # more than 5 minutes left, recompute playlist
 571                     self.recompute_playlist()
 572
 573             i += 1
 574             if i == 10:
 575                 # realign clock every ten cycles
 576                 now = datetime.datetime.now()
 577                 # adjust to awake at :00
 578                 sleep = ((60 - now.second) % 10) or 10
 579                 i = 0
 580
 581     async def handle_connection(self, reader, writer):
 582         writer.write(b'Watusi!\n')
 583         writer.write(b'Known commands: status, softquit, hardquit\n')
 584         writer.write(b'(dot on empty line to stop connection)\n')
 585         await writer.drain()
 586         end = False
 587         while not end:
 588             data = await reader.read(100)
 589             try:
 590                 message = data.decode().strip()
 591             except UnicodeDecodeError:
 592                 logger.debug('Server, invalid message %r', message)
 593                 if not data:
 594                     end = True
 595                 continue
 596             logger.debug('Server, message %r', message)
 597             if message == 'status':
 598                 response = {'slot': str(self.slot)}
 599                 if isinstance(self.slot, Nonstop):
 600                     try:
 601                         track = self.playlist[self.playhead]
 602                     except IndexError:
 603                         pass
 604                     else:
 605                         response['track'] = {}
 606                         response['track']['start_datetime'] = self.current_track_start_datetime.strftime(
 607                             '%Y-%m-%d %H:%M:%S'
 608                         )
 609                         response['track']['title'] = track.title
 610                         response['track']['artist'] = track.artist.name if track.artist_id else ''
 611                         response['track']['duration'] = track.duration.total_seconds()
 612                         response['track']['elapsed'] = (
 613                             datetime.datetime.now() - self.current_track_start_datetime
 614                         ).total_seconds()
 615                         response['track']['remaining'] = (
 616                             track.duration - datetime.timedelta(seconds=response['track']['elapsed'])
 617                         ).total_seconds()
 618                 next_diffusion = self.get_next_diffusion(
 619                     before_datetime=datetime.datetime.now() + datetime.timedelta(hours=5)
 620                 )
 621                 if next_diffusion:
 622                     response['next_diffusion'] = {
 623                         'label': str(next_diffusion),
 624                         'start_datetime': next_diffusion.datetime.strftime('%Y-%m-%d %H:%M:%S'),
 625                     }
 626                     if isinstance(next_diffusion, ScheduledDiffusion):
 627                         response['next_diffusion'][
 628                             'emission'
 629                         ] = next_diffusion.diffusion.episode.emission.title
 630                         response['next_diffusion']['episode'] = next_diffusion.diffusion.episode.title
 631             elif message == '.':
 632                 end = True
 633                 response = {'ack': True}
 634             elif message == 'softquit':
 635                 self.quit = True
 636                 end = True
 637                 response = {'ack': True}
 638             elif message == 'hardquit':
 639                 self.quit = True
 640                 end = True
 641                 response = {'ack': True}
 642                 if self.player and self.player.returncode is None:  # not finished
 643                     self.player.kill()
 644             else:
 645                 response = {'err': 1, 'msg': 'unknown command: %r' % message}
 646             writer.write(json.dumps(response).encode('utf-8') + b'\n')
 647             try:
 648                 await writer.drain()
 649             except ConnectionResetError:
 650                 break
 651         writer.close()
 652
 653     def sigterm_handler(self):
 654         logger.info('Got SIGTERM')
 655         self.quit = True
 656         self.play_task.cancel()
 657
 658     async def main(self):
 659         self.player = None
 660         loop = asyncio.get_running_loop()
 661         loop.add_signal_handler(signal.SIGTERM, self.sigterm_handler)
 662         self.recompute_slots()
 663         server = await asyncio.start_server(
 664             self.handle_connection, app_settings.SERVER_BIND_IFACE, app_settings.SERVER_BIND_PORT
 665         )
 666         async with server:
 667             asyncio.create_task(server.serve_forever())
 668
 669             self.recompute_slots_task = asyncio.create_task(self.recompute_slots_loop())
 670             while not self.quit:
 671                 now = datetime.datetime.now()
 672                 duration = (self.slot.end_datetime - now).seconds
 673                 logger.debug('Next sure shot %s (in %s)', self.slot.end_datetime, duration)
 674                 if duration < 2:
 675                     # next slot is very close, wait for it
 676                     await asyncio.sleep(duration)
 677                     self.recompute_slots()
 678                 self.play_task = asyncio.create_task(self.play(self.slot))
 679                 try:
 680                     await self.play_task
 681                     self.recompute_slots()
 682                 except asyncio.CancelledError:
 683                     logger.debug('Player cancelled exception')
 684                     if self.player and self.player.returncode is None:  # not finished
 685                         self.player.kill()
 686                 except KeyboardInterrupt:
 687                     self.quit = True
 688                     break