CC-1665: Scheduled stream rebroadcasting and recording

-webstreams now start and stop at the correct time.

airtime_mvc/application/models/Schedule.php

@@ -251,7 +251,7 @@ class Application_Model_Schedule
     {
         global $CC_CONFIG;
         $con = Propel::getConnection();
-        $sql = "SELECT DISTINCT
+        $templateSql = "SELECT DISTINCT
 
         showt.name AS show_name, showt.color AS show_color,
         showt.background_color AS show_background_color, showt.id AS show_id,
@@ -267,13 +267,12 @@ class Application_Model_Schedule
 
         --ft.track_title AS file_track_title, ft.artist_name AS file_artist_name,
         --ft.album_title AS file_album_title, ft.length AS file_length, ft.file_exists AS file_exists
-
+        %%columns%%
-        %%file_columns%%
 
         FROM
         ((
             --cc_schedule AS sched JOIN cc_files AS ft ON (sched.file_id = ft.id)
-            %%file_join%%
+            %%join%%
 
         --JOIN cc_webstream AS ws ON (sched.stream_id = ws.id)
         RIGHT JOIN cc_show_instances AS si ON (si.id = sched.instance_id))
@@ -287,25 +286,30 @@ class Application_Model_Schedule
         OR (si.starts <= '{$p_start}' AND si.ends >= '{$p_end}'))";
 
         if (count($p_shows) > 0) {
-            $sql .= " AND show_id IN (".implode(",", $p_shows).")";
+            $templateSql .= " AND show_id IN (".implode(",", $p_shows).")";
         }
 
-        $sql .= " ORDER BY si.starts, sched.starts";
+        $templateSql .= " ORDER BY si.starts, sched.starts";
-        $sql2 = $sql;
 
-        $sql = str_replace("%%file_columns%%", "ft.track_title AS file_track_title, ft.artist_name AS file_artist_name,
+        $filesSql = str_replace("%%columns%%", 
-            ft.album_title AS file_album_title, ft.length AS file_length, ft.file_exists AS file_exists", $sql);
+            "ft.track_title AS file_track_title, ft.artist_name AS file_artist_name,
-        $sql = str_replace("%%file_join%%", "cc_schedule AS sched JOIN cc_files AS ft ON (sched.file_id = ft.id)", $sql);
+            ft.album_title AS file_album_title, ft.length AS file_length, ft.file_exists AS file_exists", 
+            $templateSql);
+        $filesSql= str_replace("%%join%%", 
+            "cc_schedule AS sched JOIN cc_files AS ft ON (sched.file_id = ft.id)", 
+            $filesSql);
 
-        $sql2 = str_replace("%%file_columns%%", "ws.name AS file_track_title, ws.login AS file_artist_name,
+        $streamSql = str_replace("%%columns%%", 
-            ws.description AS file_album_title, ws.length AS file_length, 't'::BOOL AS file_exists", $sql2);
+            "ws.name AS file_track_title, ws.login AS file_artist_name,
-        $sql2 = str_replace("%%file_join%%", "cc_schedule AS sched JOIN cc_webstream AS ws ON (sched.stream_id = ws.id)", $sql2);
+            ws.description AS file_album_title, ws.length AS file_length, 't'::BOOL AS file_exists", 
+            $templateSql);
+        $streamSql = str_replace("%%join%%", 
+            "cc_schedule AS sched JOIN cc_webstream AS ws ON (sched.stream_id = ws.id)", 
+            $streamSql);
 
-        $sql = "($sql) UNION ($sql2)";
+        $sql = "($filesSql) UNION ($streamSql)";
 
-        Logging::debug($sql);
         $rows = $con->query($sql)->fetchAll(PDO::FETCH_ASSOC);
-
         return $rows;
     }
 
@@ -486,13 +490,13 @@ class Application_Model_Schedule
             ." st.cue_out AS cue_out,"
             ." st.fade_in AS fade_in,"
             ." st.fade_out AS fade_out,"
-            ." st.type AS type,"
+            //." st.type AS type,"
             ." si.starts AS show_start,"
             ." si.ends AS show_end,"
-            ." f.id AS file_id"
+            ." f.id AS file_id,"
-            ." f.replay_gain AS replay_gain"
+            ." f.replay_gain AS replay_gain,"
-            ." f.file_path AS file_path"
+            ." f.filepath AS filepath,"
-            ." ws.id as stream_id"
+            ." ws.id as stream_id,"
             ." ws.url as url"
             ." FROM $CC_CONFIG[scheduleTable] AS st"
             ." LEFT JOIN $CC_CONFIG[showInstances] AS si"
@@ -588,13 +592,14 @@ class Application_Model_Schedule
             $data["media"][$kick_start]['end'] = $kick_start;
             $data["media"][$kick_start]['event_type'] = "kick_out";
             $data["media"][$kick_start]['type'] = "event";
+            $data["media"][$kick_start]['independent_event'] = true;
 
             if ($kick_time !== $switch_off_time) {
                 $switch_start = Application_Model_Schedule::AirtimeTimeToPypoTime($switch_off_time);
                 $data["media"][$switch_start]['start'] = $switch_start;
                 $data["media"][$switch_start]['end'] = $switch_start;
                 $data["media"][$switch_start]['event_type'] = "switch_off";
-                $data["media"][$switch_start]['type'] = "event";
+                $data["media"][$switch_start]['independent_event'] = true;
             }
         }
 
@@ -622,23 +627,22 @@ class Application_Model_Schedule
                 $item["end"] = $showEndDateTime->format("Y-m-d H:i:s");
             }
 
-            Logging::log($item);
+            if (!is_null($item['file_id'])) {
-            //TODO: need to know item type
+                //row is from "file"
-            //
-            //
-            if ($item['type'] == 0) {
-                //row is from cc_files
                 $media_id = $item['file_id'];
-                $uri = $item['file_path'];
+                $uri = $item['filepath'];
                 $type = "file";
-            } else if ($item['type'] == 1) {
+            } else if (!is_null($item['stream_id'])) {
+                //row is type "webstream"
                 $media_id = $item['stream_id'];
                 $uri = $item['url'];
                 $type = "stream";
             }
 
+            
 
             $start = Application_Model_Schedule::AirtimeTimeToPypoTime($item["start"]);
+            $end = Application_Model_Schedule::AirtimeTimeToPypoTime($item["end"]);
 
             $data["media"][$start] = array(
                 'id' => $media_id,
@@ -650,10 +654,24 @@ class Application_Model_Schedule
                 'cue_in' => Application_Common_DateHelper::CalculateLengthInSeconds($item["cue_in"]),
                 'cue_out' => Application_Common_DateHelper::CalculateLengthInSeconds($item["cue_out"]),
                 'start' => $start,
-                'end' => Application_Model_Schedule::AirtimeTimeToPypoTime($item["end"]),
+                'end' => $end,
                 'show_name' => $showName,
-                'replay_gain' => is_null($item["replay_gain"]) ? "0": $item["replay_gain"]
+                'replay_gain' => is_null($item["replay_gain"]) ? "0": $item["replay_gain"],
+                'independent_event' => false
             );
+
+            if ($type == "stream") {
+                //since a stream never ends we have to insert an additional "kick stream" event. The "start"
+                //time of this event is the "end" time of the stream.
+                $data["media"][$end] = array(
+                    'start' => $end,
+                    'end' => $end,
+                    'uri' => $uri,
+                    'type' => 'stream_end',
+                    'independent_event' => true
+                    );
+
+            }
         }
 
         return $data;

python_apps/pypo/liquidsoap_scripts/ls_script.liq

@@ -23,7 +23,6 @@ set("harbor.bind_addr", "0.0.0.0")
 web_stream = input.harbor("test-harbor",port=8999,password="hackme")
 
 pypo_data = ref '0'
-web_stream_enabled = ref false
 stream_metadata_type = ref 0
 default_dj_fade = ref 0.
 station_name = ref ''
@@ -43,12 +42,11 @@ queue = on_metadata(notify, queue)
 queue = map_metadata(append_title, queue)
 # the crossfade function controls fade in/out
 queue = crossfade(queue)
-ignore(output.dummy(queue, fallible=true))
 
 queue = fallback([web_stream, queue])
+ignore(output.dummy(queue, fallible=true))
 
 server.register(namespace="vars", "pypo_data", fun (s) -> begin pypo_data := s "Done" end)
-server.register(namespace="vars", "web_stream_enabled", fun (s) -> begin web_stream_enabled := (s == "true") string_of(!web_stream_enabled) end)
 server.register(namespace="vars", "stream_metadata_type", fun (s) -> begin stream_metadata_type := int_of_string(s) s end)
 server.register(namespace="vars", "show_name", fun (s) -> begin show_name := s s end)
 server.register(namespace="vars", "station_name", fun (s) -> begin station_name := s s end)
@@ -179,20 +177,6 @@ s = append_dj_inputs(master_live_stream_port, master_live_stream_mp, dj_live_str
 
 # Attach a skip command to the source s:
 
-#web_stream_source = input.http(id="web_stream", autostart = false, buffer=0.5, max=20., "")
-
-#once the stream is started, give it a sink so that liquidsoap doesn't
-#create buffer overflow warnings in the log file.
-#output.dummy(fallible=true, web_stream_source)
-
-#s = switch(track_sensitive = false,
-#    transitions=[to_live,to_live],
-#    [
-#      ({ !web_stream_enabled }, web_stream_source),
-#      ({ true }, s)
-#    ]
-#)
-
 add_skip_command(s)
 
 server.register(namespace="streams",

python_apps/pypo/pypofetch.py

@@ -425,6 +425,11 @@ class PypoFetch(Thread):
 
             for key in media:
                 media_item = media[key]
+                """
+                {u'end': u'2012-07-26-04-05-00', u'fade_out': 500, u'show_name': u'Untitled Show', u'uri': u'http://', 
+ u'cue_in': 0, u'start': u'2012-07-26-04-00-00', u'replay_gain': u'0', u'row_id': 16, u'cue_out': 300, u'type': 
+ u'stream', u'id': 1, u'fade_in': 500}
+                """
                 if(media_item['type'] == 'file'):
                     fileExt = os.path.splitext(media_item['uri'])[1]
                     dst = os.path.join(download_dir, media_item['id'] + fileExt)
@@ -455,15 +460,19 @@ class PypoFetch(Thread):
 
         for mkey in media:
             media_item = media[mkey]
-            fileExt = os.path.splitext(media_item['uri'])[1]
+            if media_item['type'] == 'file':
-            scheduled_file_set.add(media_item["id"] + fileExt)
+                fileExt = os.path.splitext(media_item['uri'])[1]
+                scheduled_file_set.add(media_item["id"] + fileExt)
 
-        unneeded_files = cached_file_set - scheduled_file_set
+        expired_files = cached_file_set - scheduled_file_set
 
-        self.logger.debug("Files to remove " + str(unneeded_files))
+        self.logger.debug("Files to remove " + str(expired_files))
-        for f in unneeded_files:
+        for f in expired_files:
-            self.logger.debug("Removing %s" % os.path.join(self.cache_dir, f))
+            try:
-            os.remove(os.path.join(self.cache_dir, f))
+                self.logger.debug("Removing %s" % os.path.join(self.cache_dir, f))
+                os.remove(os.path.join(self.cache_dir, f))
+            except Exception, e:
+                self.logger.error(e)
 
     def main(self):
         # Bootstrap: since we are just starting up, we need to grab the

python_apps/pypo/pypopush.py

@@ -241,7 +241,7 @@ class PypoPush(Thread):
 
         for mkey in sorted_keys:
             media_item = media_schedule[mkey]
-            if media_item['type'] == "event":
+            if media_item['independent_event']:
                 chains.append([media_item])
             elif len(current_chain) == 0:
                 current_chain.append(media_item)
@@ -353,9 +353,55 @@ class PypoPush(Thread):
                         PypoFetch.disconnect_source(self.logger, self.telnet_lock, "live_dj")
                     elif media_item['event_type'] == "switch_off":
                         PypoFetch.switch_source(self.logger, self.telnet_lock, "live_dj", "off")
+                elif media_item['type'] == "stream":
+                    """
+                    Source is a stream that we need to being downloading to a file. Then we may simply
+                    point Liquidsoap to play this file.
+                    """
+                    self.start_web_stream(media_item)
+                elif media_item['type'] == "stream_end":
+                    self.stop_web_stream(media_item)
         except Exception, e:
             self.logger.error('Pypo Push Exception: %s', e)
 
+    def start_web_stream(self, media_item):
+        try:
+            self.telnet_lock.acquire()
+            tn = telnetlib.Telnet(LS_HOST, LS_PORT)
+            #dynamic_source.start http://87.230.101.24:80/top100station.mp3
+
+            msg = 'streams.scheduled_play_start\n'
+            tn.write(msg)
+            msg = 'dynamic_source.start %s\n' % media_item['uri'].encode('latin-1')
+            self.logger.debug(msg)
+            tn.write(msg)
+
+            tn.write("exit\n")
+            self.logger.debug(tn.read_all())
+
+        except Exception, e:
+            self.logger.error(str(e))
+        finally:
+            self.telnet_lock.release()
+
+    def stop_web_stream(self, media_item):
+        try:
+            self.telnet_lock.acquire()
+            tn = telnetlib.Telnet(LS_HOST, LS_PORT)
+            #dynamic_source.stop http://87.230.101.24:80/top100station.mp3
+
+            msg = 'dynamic_source.stop %s\n' % media_item['uri'].encode('latin-1')
+            self.logger.debug(msg)
+            tn.write(msg)
+
+            tn.write("exit\n")
+            self.logger.debug(tn.read_all())
+
+        except Exception, e:
+            self.logger.error(str(e))
+        finally:
+            self.telnet_lock.release()
+
     def clear_liquidsoap_queue(self):
         self.logger.debug("Clearing Liquidsoap queue")
         try: