Add PCR recovery from TS sources and add PCR drift compensation to TS output

transports.c

@@ -158,7 +158,7 @@ transport_start(th_transport_t *t, unsigned int weight)
 
   t->tht_monitor_suspend = 10;
   t->tht_dts_start = AV_NOPTS_VALUE;
-
+  t->tht_pcr_drift = 0;
   LIST_FOREACH(st, &t->tht_streams, st_link) {
   
     st->st_startcond = 0xffffffff;
@@ -169,6 +169,10 @@ transport_start(th_transport_t *t, unsigned int weight)
     st->st_last_dts = AV_NOPTS_VALUE;
     st->st_dts_epoch = 0; 
  
+    st->st_pcr_real_last = AV_NOPTS_VALUE;
+    st->st_pcr_last      = AV_NOPTS_VALUE;
+    st->st_pcr_drift     = 0;
+    st->st_pcr_recovery_fails = 0;
     /* Open ffmpeg context and parser */
 
     switch(st->st_type) {

tsdemux.c

@@ -143,43 +143,48 @@ ts_recv_packet0(th_transport_t *t, th_stream_t *st, uint8_t *tsb)
 
 
 /**
- * Process transport stream packets, extract PCR and optionally descramble
+ * Recover PCR
+ *
+ * st->st_pcr_drift will increase if our (system clock) runs faster
+ * than the stream PCR
  */
 static void
 ts_extract_pcr(th_transport_t *t, th_stream_t *st, uint8_t *tsb)
 {
   int64_t real = getclock_hires();
-  int64_t pcr;
+  int64_t pcr, d;
 
   pcr  = tsb[6] << 25;
   pcr |= tsb[7] << 17;
   pcr |= tsb[8] << 9;
   pcr |= tsb[9] << 1;
   pcr |= (tsb[10] >> 7) & 0x01;
-
+  
   pcr = av_rescale_q(pcr, st->st_tb, AV_TIME_BASE_Q);
 
-
-  {
-    static int64_t pcr0;
-    static int64_t real0, dv;
-
-    int64_t pcr_d, real_d, dx;
-
-    pcr_d = pcr - pcr0;
-    real_d = real - real0;
-
-    dx = pcr_d - real_d;
-
-    if(real0 && dx > -1000000LL && dx < 1000000LL)
-      dv += dx;
-#if 0
-    printf("B: realtime = %-12lld pcr = %-12lld delta = %lld\n",
-	   real_d, pcr_d, dv / 1000);
-#endif
-    pcr0 = pcr;
-    real0 = real;
+  if(st->st_pcr_real_last != AV_NOPTS_VALUE) {
+    d = (real - st->st_pcr_real_last) - (pcr - st->st_pcr_last);
+    
+    if(d < -1000000LL || d > 1000000LL) {
+      st->st_pcr_recovery_fails++;
+      if(st->st_pcr_recovery_fails > 10) {
+	st->st_pcr_recovery_fails = 0;
+	st->st_pcr_real_last = AV_NOPTS_VALUE;
+      }
+      return;
+    }
+    st->st_pcr_recovery_fails = 0;
+    st->st_pcr_drift += d;
+    
+    if(t->tht_pcr_pid == st->st_pid) {
+      /* This is the registered PCR PID, adjust transport PCR drift
+	 via an IIR filter */
+      
+      t->tht_pcr_drift = (t->tht_pcr_drift * 255 + st->st_pcr_drift) / 256;
+    }
   }
+  st->st_pcr_last = pcr;
+  st->st_pcr_real_last = real;
 }
 
 /**

tsmux.c

@@ -61,6 +61,7 @@ ts_muxer_send_packet(ts_muxer_t *ts)
   int i;
   int64_t t, tlow, pcr;
   uint8_t *d;
+  th_transport_t *tr;
 
   if(ts->ts_block == 0)
     return;
@@ -73,8 +74,9 @@ ts_muxer_send_packet(ts_muxer_t *ts)
     for(i = 0; i < ts->ts_block; i++) {
       d = ts->ts_packet + i * 188;
       if((d[3] & 0xf0) == 0x30 && d[4] >= 7 && d[5] & 0x10) {
+	tr = ts->ts_muxer->tm_subscription->ths_transport;
 
-	pcr = getclock_hires() - ts->ts_pcr_ref;
+	pcr = getclock_hires() - ts->ts_pcr_ref - tr->tht_pcr_drift;
 	t = av_rescale_q(pcr, AV_TIME_BASE_Q, mpeg_tc_27M);
 	tlow = t % 300LL;
 	t =    t / 300LL;
@@ -273,10 +275,11 @@ ts_deliver(void *opaque, int64_t now)
 {
   th_muxstream_t *tms = opaque;
   th_muxer_t *tm = tms->tms_muxer;
+  th_transport_t *t = tm->tm_subscription->ths_transport;
   ts_muxer_t *ts = tm->tm_opaque;
   th_muxpkt_t *f;
   th_muxfifo_t *tmf = &tms->tms_delivery_fifo;
-  int64_t pcr = now - ts->ts_pcr_ref;
+  int64_t pcr = now - ts->ts_pcr_ref - t->tht_pcr_drift;
   int64_t dl, next, delta;
 
   f = tmf_deq(tmf);
@@ -295,7 +298,7 @@ ts_deliver(void *opaque, int64_t now)
       return;
   }
 
-  next = f->tm_deadline + ts->ts_pcr_ref;
+  next = f->tm_deadline + ts->ts_pcr_ref - t->tht_pcr_drift;
   if(next < now + 100)
     next = now + 100;
 
@@ -316,6 +319,7 @@ ts_check_deliver(ts_muxer_t *ts, th_muxstream_t *tms)
   int64_t now;
   th_muxpkt_t *f;
   th_muxfifo_t *tmf = &tms->tms_delivery_fifo;
+  th_transport_t *t = ts->ts_muxer->tm_subscription->ths_transport;
   int64_t next;
 
   if(dtimer_isarmed(&tms->tms_mux_timer))
@@ -330,11 +334,11 @@ ts_check_deliver(ts_muxer_t *ts, th_muxstream_t *tms)
     if(ts->ts_pcr_start == AV_NOPTS_VALUE)
       return; /* dont know anything yet */
 
-    ts->ts_pcr_ref = now - ts->ts_pcr_start;
+    ts->ts_pcr_ref = now - ts->ts_pcr_start + t->tht_pcr_drift;
   }
 
   f = TAILQ_FIRST(&tmf->tmf_queue);  /* next packet we are going to send */
-  next = f->tm_deadline + ts->ts_pcr_ref;
+  next = f->tm_deadline + ts->ts_pcr_ref - t->tht_pcr_drift;
 
   if(next < now + 100)
     next = now + 100;

tvhead.h

@@ -289,6 +289,13 @@ typedef struct th_stream {
   pid_section_callback_t *st_got_section;
   void *st_got_section_opaque;
 
+  /* PCR recovery */
+
+  int st_pcr_recovery_fails;
+  int64_t st_pcr_real_last;     /* realtime clock when we saw last PCR */
+  int64_t st_pcr_last;          /* PCR clock when we saw last PCR */
+  int64_t st_pcr_drift;
+
   /* For transport stream packet reassembly */
 
   uint8_t *st_buffer;
@@ -383,7 +390,8 @@ typedef struct th_transport {
   struct th_stream_list tht_streams;
   th_stream_t *tht_video;
   th_stream_t *tht_audio;
-
+  
+  int64_t  tht_pcr_drift;
   uint16_t tht_pcr_pid;
   uint16_t tht_dvb_transport_id;
   uint16_t tht_dvb_service_id;