Snapshot 2007-03-20

8 files changed, 679 insertions, 0 deletions

diff --git a/remux/extern.c b/remux/extern.c
new file mode 100644
index 0000000..51c353c
--- /dev/null
+++ b/remux/extern.c
@@ -0,0 +1,157 @@
+#include "remux/extern.h"
+#include "server/streamer.h"
+#include <vdr/tools.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <signal.h>
+#include <unistd.h>
+
+const char *g_ExternRemux = "/root/externremux.sh";
+
+class cTSExt: public cThread {
+private:
+	cRingBufferLinear *m_ResultBuffer;
+	bool               m_Active;
+	int                m_Process;
+	int                m_Inpipe, m_Outpipe;
+
+protected:
+	virtual void Action(void);
+
+public:
+	cTSExt(cRingBufferLinear *ResultBuffer);
+	virtual ~cTSExt();
+
+	void Put(const uchar *Data, int Count);
+};
+
+cTSExt::cTSExt(cRingBufferLinear *ResultBuffer):
+		m_ResultBuffer(ResultBuffer),
+		m_Active(false),
+		m_Process(0),
+		m_Inpipe(0),
+		m_Outpipe(0)
+{
+	int inpipe[2];
+	int outpipe[2];
+
+	if (pipe(inpipe) == -1) {
+		LOG_ERROR_STR("pipe failed");
+		return;
+	}
+
+	if (pipe(outpipe) == -1) {
+		LOG_ERROR_STR("pipe failed");
+		close(inpipe[0]);
+		close(inpipe[1]);
+		return;
+	}
+
+	if ((m_Process = fork()) == -1) {
+		LOG_ERROR_STR("fork failed");
+		close(inpipe[0]);
+		close(inpipe[1]);
+		close(outpipe[0]);
+		close(outpipe[1]);
+		return;
+	}
+
+	if (m_Process == 0) {
+		// child process
+		dup2(inpipe[0], STDIN_FILENO);
+		close(inpipe[1]);
+		dup2(outpipe[1], STDOUT_FILENO);
+		close(outpipe[0]);
+
+		int MaxPossibleFileDescriptors = getdtablesize();
+		for (int i = STDERR_FILENO + 1; i < MaxPossibleFileDescriptors; i++)
+			close(i); //close all dup'ed filedescriptors
+
+		//printf("starting externremux.sh\n");
+		execl("/bin/sh", "sh", "-c", g_ExternRemux, NULL);
+		//printf("failed externremux.sh\n");
+		_exit(-1);
+	}
+
+	close(inpipe[0]);
+	close(outpipe[1]);
+	m_Inpipe = inpipe[1];
+	m_Outpipe = outpipe[0];
+	Start();
+}
+
+cTSExt::~cTSExt()
+{
+	m_Active = false;
+	Cancel(3);
+	if (m_Process > 0) {
+		close(m_Outpipe);
+		close(m_Inpipe);
+		kill(m_Process, SIGTERM);
+		for (int i = 0; waitpid(m_Process, NULL, WNOHANG) == 0; i++) {
+			if (i == 20) {
+				esyslog("streamdev-server: externremux process won't stop - killing it");
+				kill(m_Process, SIGKILL);
+			}
+			cCondWait::SleepMs(100);
+		}
+	}
+}
+
+void cTSExt::Action(void)
+{
+	m_Active = true;
+	while (m_Active) {
+		fd_set rfds;
+		struct timeval tv;
+
+		FD_ZERO(&rfds);
+		FD_SET(m_Outpipe, &rfds);
+
+		while (FD_ISSET(m_Outpipe, &rfds)) {
+			tv.tv_sec = 2;
+			tv.tv_usec = 0;
+			if (select(m_Outpipe + 1, &rfds, NULL, NULL, &tv) == -1) {
+				LOG_ERROR_STR("poll failed");
+				break;;
+			}
+
+			if (FD_ISSET(m_Outpipe, &rfds)) {
+				int result;
+				if ((result = m_ResultBuffer->Read(m_Outpipe)) == -1) {
+					LOG_ERROR_STR("read failed");
+					break;
+				}
+			}
+		}
+	}
+	m_Active = false;
+}
+
+
+void cTSExt::Put(const uchar *Data, int Count)
+{
+	if (safe_write(m_Inpipe, Data, Count) == -1) {
+		LOG_ERROR_STR("write failed");
+		return;
+	}
+}
+
+cExternRemux::cExternRemux(int VPid, const int *APids, const int *Dpids, const int *SPids):
+		m_ResultBuffer(new cRingBufferLinear(WRITERBUFSIZE, TS_SIZE * 2)),
+		m_Remux(new cTSExt(m_ResultBuffer))
+{
+	m_ResultBuffer->SetTimeouts(0, 100);
+}
+
+cExternRemux::~cExternRemux()
+{
+	delete m_Remux;
+	delete m_ResultBuffer;
+}
+
+int cExternRemux::Put(const uchar *Data, int Count) 
+{
+	m_Remux->Put(Data, Count);
+	return Count;
+}
diff --git a/remux/extern.h b/remux/extern.h
new file mode 100644
index 0000000..ae055ac
--- /dev/null
+++ b/remux/extern.h
@@ -0,0 +1,25 @@
+#ifndef VDR_STREAMDEV_EXTERNREMUX_H
+#define VDR_STREAMDEV_EXTERNREMUX_H
+
+#include "remux/tsremux.h"
+#include <vdr/ringbuffer.h>
+
+extern const char *g_ExternRemux;
+
+class cTSExt;
+
+class cExternRemux: public cTSRemux {
+private:
+	cRingBufferLinear *m_ResultBuffer;
+	cTSExt            *m_Remux;
+
+public:
+	cExternRemux(int VPid, const int *APids, const int *Dpids, const int *SPids);
+	virtual ~cExternRemux();
+	
+	int Put(const uchar *Data, int Count);
+	uchar *Get(int &Count) { return m_ResultBuffer->Get(Count); }
+	void Del(int Count) { m_ResultBuffer->Del(Count); }
+};
+
+#endif // VDR_STREAMDEV_EXTERNREMUX_H
diff --git a/remux/ts2es.c b/remux/ts2es.c
new file mode 100644
index 0000000..3476e24
--- /dev/null
+++ b/remux/ts2es.c
@@ -0,0 +1,140 @@
+#include "remux/ts2es.h"
+#include "server/streamer.h"
+#include "libdvbmpeg/transform.h"
+#include "common.h"
+#include <vdr/device.h>
+
+// from VDR's remux.c
+#define MAXNONUSEFULDATA (10*1024*1024)
+
+class cTS2ES: public ipack {
+	friend void PutES(uint8_t *Buffer, int Size, void *Data);
+
+private:
+	cRingBufferLinear *m_ResultBuffer;
+
+public:
+	cTS2ES(cRingBufferLinear *ResultBuffer);
+	~cTS2ES();
+
+	void PutTSPacket(const uint8_t *Buffer);
+};
+
+void PutES(uint8_t *Buffer, int Size, void *Data) 
+{
+	cTS2ES *This = (cTS2ES*)Data;
+	uint8_t payl = Buffer[8] + 9 + This->start - 1;
+	int count = Size - payl;
+
+	int n = This->m_ResultBuffer->Put(Buffer + payl, count);
+	if (n != count)
+		esyslog("ERROR: result buffer overflow, dropped %d out of %d byte", count - n, count);
+	This->start = 1;
+}
+
+cTS2ES::cTS2ES(cRingBufferLinear *ResultBuffer) 
+{
+	m_ResultBuffer = ResultBuffer;
+
+	init_ipack(this, IPACKS, PutES, 0);
+	data = (void*)this;
+}
+
+cTS2ES::~cTS2ES() 
+{
+	free_ipack(this);
+}
+
+void cTS2ES::PutTSPacket(const uint8_t *Buffer) {
+  if (!Buffer)
+     return;
+
+  if (Buffer[1] & 0x80) { // ts error
+		// TODO
+	}
+
+  if (Buffer[1] & 0x40) { // payload start
+		if (plength == MMAX_PLENGTH - 6) {
+    	plength = found - 6;
+      found = 0;
+      send_ipack(this);
+      reset_ipack(this);
+    }
+  }
+
+	uint8_t off = 0;
+
+  if (Buffer[3] & 0x20) {  // adaptation field?
+		off = Buffer[4] + 1;
+    if (off + 4 > TS_SIZE - 1)
+      return;
+  }
+
+  instant_repack((uint8_t*)(Buffer + 4 + off), TS_SIZE - 4 - off, this);
+}
+
+cTS2ESRemux::cTS2ESRemux(int Pid):
+		m_Pid(Pid),
+		m_ResultBuffer(new cRingBufferLinear(WRITERBUFSIZE, IPACKS)),
+		m_Remux(new cTS2ES(m_ResultBuffer))
+{
+	m_ResultBuffer->SetTimeouts(0, 100);
+}
+
+cTS2ESRemux::~cTS2ESRemux() 
+{
+	delete m_Remux;
+	delete m_ResultBuffer;
+}
+
+int cTS2ESRemux::Put(const uchar *Data, int Count) 
+{
+	int used = 0;
+
+	// Make sure we are looking at a TS packet:
+
+	while (Count > TS_SIZE) {
+		if (Data[0] == TS_SYNC_BYTE && Data[TS_SIZE] == TS_SYNC_BYTE)
+			break;
+		Data++;
+		Count--;
+		used++;
+	}
+
+	if (used)
+		esyslog("ERROR: skipped %d byte to sync on TS packet", used);
+
+	// Convert incoming TS data into ES:
+
+	for (int i = 0; i < Count; i += TS_SIZE) {
+		if (Count - i < TS_SIZE)
+			break;
+		if (Data[i] != TS_SYNC_BYTE)
+			break;
+		if (m_ResultBuffer->Free() < 2 * IPACKS)
+			break; // A cTS2ES might write one full packet and also a small rest
+		int pid = cTSRemux::GetPid(Data + i + 1);
+		if (Data[i + 3] & 0x10) { // got payload
+			if (m_Pid == pid)
+				m_Remux->PutTSPacket(Data + i);
+		}
+		used += TS_SIZE;
+	}
+
+/*
+  // Check if we're getting anywhere here:
+  if (!synced && skipped >= 0) {
+     if (skipped > MAXNONUSEFULDATA) {
+        esyslog("ERROR: no useful data seen within %d byte of video stream", skipped);
+        skipped = -1;
+        if (exitOnFailure)
+           cThread::EmergencyExit(true);
+        }
+     else
+        skipped += used;
+     }
+*/
+
+	return used;
+}
+
diff --git a/remux/ts2es.h b/remux/ts2es.h
new file mode 100644
index 0000000..551df1d
--- /dev/null
+++ b/remux/ts2es.h
@@ -0,0 +1,25 @@
+#ifndef VDR_STREAMDEV_TS2ESREMUX_H
+#define VDR_STREAMDEV_TS2ESREMUX_H
+
+#include "remux/tsremux.h"
+#include <vdr/ringbuffer.h>
+
+class cTS2ES;
+class cRingBufferLinear;
+
+class cTS2ESRemux: public cTSRemux {
+private:
+	int                m_Pid;
+	cRingBufferLinear *m_ResultBuffer;
+	cTS2ES            *m_Remux;
+
+public:
+	cTS2ESRemux(int Pid);
+	virtual ~cTS2ESRemux();
+	
+	int Put(const uchar *Data, int Count);
+	uchar *Get(int &Count) { return m_ResultBuffer->Get(Count); }
+	void Del(int Count) { m_ResultBuffer->Del(Count); }
+};
+
+#endif // VDR_STREAMDEV_TS2ESREMUX_H
diff --git a/remux/ts2ps.c b/remux/ts2ps.c
new file mode 100644
index 0000000..d0d08cf
--- /dev/null
+++ b/remux/ts2ps.c
@@ -0,0 +1,211 @@
+#include "remux/ts2ps.h"
+#include "server/streamer.h"
+#include <vdr/channels.h>
+#include <vdr/device.h>
+
+class cTS2PS {
+	friend void PutPES(uint8_t *Buffer, int Size, void *Data);
+
+private:
+	ipack m_Ipack;
+	int m_Pid;
+	cRingBufferLinear *m_ResultBuffer;
+
+public:
+	cTS2PS(cRingBufferLinear *ResultBuffer, int Pid, uint8_t AudioCid = 0x00);
+	~cTS2PS();
+
+	void PutTSPacket(const uint8_t *Buffer);
+
+	int Pid(void) const { return m_Pid; }
+};
+
+void PutPES(uint8_t *Buffer, int Size, void *Data) 
+{
+	cTS2PS *This = (cTS2PS*)Data;
+	int n = This->m_ResultBuffer->Put(Buffer, Size);
+	if (n != Size)
+		esyslog("ERROR: result buffer overflow, dropped %d out of %d byte", Size - n, Size);
+}
+
+cTS2PS::cTS2PS(cRingBufferLinear *ResultBuffer, int Pid, uint8_t AudioCid)
+{
+	m_ResultBuffer = ResultBuffer;
+	m_Pid = Pid;
+
+	init_ipack(&m_Ipack, IPACKS, PutPES, false);
+	m_Ipack.cid = AudioCid;
+	m_Ipack.data = (void*)this;
+}
+
+cTS2PS::~cTS2PS() 
+{
+	free_ipack(&m_Ipack);
+}
+
+void cTS2PS::PutTSPacket(const uint8_t *Buffer) 
+{
+  if (!Buffer)
+     return;
+
+  if (Buffer[1] & 0x80) { // ts error
+		// TODO
+	}
+
+  if (Buffer[1] & 0x40) { // payload start
+		if (m_Ipack.plength == MMAX_PLENGTH - 6 && m_Ipack.found > 6) {
+    	m_Ipack.plength = m_Ipack.found - 6;
+      m_Ipack.found = 0;
+      send_ipack(&m_Ipack);
+      reset_ipack(&m_Ipack);
+    }
+  }
+
+	uint8_t off = 0;
+
+  if (Buffer[3] & 0x20) {  // adaptation field?
+		off = Buffer[4] + 1;
+    if (off + 4 > TS_SIZE - 1)
+      return;
+  }
+
+  instant_repack((uint8_t*)(Buffer + 4 + off), TS_SIZE - 4 - off, &m_Ipack);
+}
+
+cTS2PSRemux::cTS2PSRemux(int VPid, const int *APids, const int *DPids, const int *SPids):
+		m_NumTracks(0),
+		m_ResultBuffer(new cRingBufferLinear(WRITERBUFSIZE, IPACKS)),
+		m_ResultSkipped(0),
+		m_Skipped(0),
+		m_Synced(false),
+		m_IsRadio(VPid == 0 || VPid == 1 || VPid == 0x1FFF)
+{
+	m_ResultBuffer->SetTimeouts(0, 100);
+
+	if (VPid)
+		m_Remux[m_NumTracks++] = new cTS2PS(m_ResultBuffer, VPid);
+	if (APids) {
+		int n = 0;
+		while (*APids && m_NumTracks < MAXTRACKS && n < MAXAPIDS)
+			m_Remux[m_NumTracks++] = new cTS2PS(m_ResultBuffer, *APids++, 0xC0 + n++);
+	}
+	if (DPids) {
+		int n = 0;
+		while (*DPids && m_NumTracks < MAXTRACKS && n < MAXDPIDS)
+			m_Remux[m_NumTracks++] = new cTS2PS(m_ResultBuffer, *DPids++, 0x80 + n++);
+	}
+}
+
+cTS2PSRemux::~cTS2PSRemux() {
+	for (int i = 0; i < m_NumTracks; ++i)
+		delete m_Remux[i];
+	delete m_ResultBuffer;
+}
+
+int cTS2PSRemux::Put(const uchar *Data, int Count)
+{
+	int used = 0;
+
+	// Make sure we are looking at a TS packet:
+	while (Count > TS_SIZE) {
+		if (Data[0] == TS_SYNC_BYTE && Data[TS_SIZE] == TS_SYNC_BYTE)
+			break;
+		Data++;
+		Count--;
+		used++;
+	}
+	if (used)
+		esyslog("ERROR: m_Skipped %d byte to sync on TS packet", used);
+
+	// Convert incoming TS data into multiplexed PS:
+
+	for (int i = 0; i < Count; i += TS_SIZE) {
+		if (Count - i < TS_SIZE)
+			break;
+		if (Data[i] != TS_SYNC_BYTE)
+			break;
+		if (m_ResultBuffer->Free() < 2 * IPACKS)
+			break; // A cTS2PS might write one full packet and also a small rest
+		int pid = GetPid(Data + i + 1);
+		if (Data[i + 3] & 0x10) { // got payload
+			for (int t = 0; t < m_NumTracks; t++) {
+				if (m_Remux[t]->Pid() == pid) {
+					m_Remux[t]->PutTSPacket(Data + i);
+					break;
+				}
+			}
+		}
+		used += TS_SIZE;
+	}
+
+	// Check if we're getting anywhere here:
+	if (!m_Synced && m_Skipped >= 0)
+		m_Skipped += used;
+
+	return used;
+}
+
+uchar *cTS2PSRemux::Get(int &Count)
+{
+	// Remove any previously skipped data from the result buffer:
+
+	if (m_ResultSkipped > 0) {
+		m_ResultBuffer->Del(m_ResultSkipped);
+		m_ResultSkipped = 0;
+	}
+
+	// Special VPID case to enable recording radio channels:
+	if (m_IsRadio) {
+		// Force syncing of radio channels to avoid "no useful data" error
+		m_Synced = true;
+		return m_ResultBuffer->Get(Count);
+	}
+
+	// Check for frame borders:
+	Count = 0;
+	uchar *resultData = NULL;
+	int resultCount = 0;
+	uchar *data = m_ResultBuffer->Get(resultCount);
+	if (data) {
+		for (int i = 0; i < resultCount - 3; i++) {
+			if (data[i] == 0 && data[i + 1] == 0 && data[i + 2] == 1) {
+				int l = 0;
+				uchar StreamType = data[i + 3];
+				if (VIDEO_STREAM_S <= StreamType && StreamType <= VIDEO_STREAM_E) {
+					uchar pt = NO_PICTURE;
+					l = ScanVideoPacket(data, resultCount, i, pt);
+					if (l < 0)
+						return resultData;
+					if (pt != NO_PICTURE) {
+						if (pt < I_FRAME || B_FRAME < pt) {
+							esyslog("ERROR: unknown picture type '%d'", pt);
+						}
+						else if (!m_Synced) {
+							if (pt == I_FRAME) {
+								m_ResultSkipped = i; // will drop everything before this position
+								SetBrokenLink(data + i, l);
+								m_Synced = true;
+							}
+						}
+						else if (Count)
+							return resultData;
+					}
+				} else {
+					l = GetPacketLength(data, resultCount, i);
+					if (l < 0)
+						return resultData;
+				}
+				if (m_Synced) {
+					if (!Count)
+						resultData = data + i;
+					Count += l;
+				} else
+					m_ResultSkipped = i + l;
+				if (l > 0)
+					i += l - 1; // the loop increments, too
+			}
+		}
+	}
+	return resultData;
+}
+
diff --git a/remux/ts2ps.h b/remux/ts2ps.h
new file mode 100644
index 0000000..334215a
--- /dev/null
+++ b/remux/ts2ps.h
@@ -0,0 +1,29 @@
+#ifndef VDR_STREAMDEV_TS2PESREMUX_H
+#define VDR_STREAMDEV_TS2PESREMUX_H
+
+#include "remux/tsremux.h"
+#include <vdr/remux.h>
+#include <vdr/ringbuffer.h>
+
+class cTS2PS;
+
+class cTS2PSRemux: public cTSRemux {
+private:
+	int                m_NumTracks;
+	cTS2PS            *m_Remux[MAXTRACKS];
+	cRingBufferLinear *m_ResultBuffer;
+	int                m_ResultSkipped;
+	int                m_Skipped;
+	bool               m_Synced;
+	bool               m_IsRadio;
+	
+public:
+	cTS2PSRemux(int VPid, const int *Apids, const int *Dpids, const int *Spids);
+	virtual ~cTS2PSRemux();
+
+	int Put(const uchar *Data, int Count);
+	uchar *Get(int &Count);
+	void Del(int Count) { m_ResultBuffer->Del(Count); }
+};
+
+#endif // VDR_STREAMDEV_TS2PESREMUX_H
diff --git a/remux/tsremux.c b/remux/tsremux.c
new file mode 100644
index 0000000..6be5245
--- /dev/null
+++ b/remux/tsremux.c
@@ -0,0 +1,59 @@
+#include "remux/tsremux.h"
+
+#define SC_PICTURE 0x00  // "picture header"
+
+void cTSRemux::SetBrokenLink(uchar *Data, int Length)
+{
+  if (Length > 9 && Data[0] == 0 && Data[1] == 0 && Data[2] == 1 && (Data[3] & 0xF0) == VIDEO_STREAM_S) {
+     for (int i = Data[8] + 9; i < Length - 7; i++) { // +9 to skip video packet header
+         if (Data[i] == 0 && Data[i + 1] == 0 && Data[i + 2] == 1 && Data[i + 3] == 0xB8) {
+            if (!(Data[i + 7] & 0x40)) // set flag only if GOP is not closed
+               Data[i + 7] |= 0x20;
+            return;
+            }
+         }
+     dsyslog("SetBrokenLink: no GOP header found in video packet");
+     }
+  else
+     dsyslog("SetBrokenLink: no video packet in frame");
+}
+
+int cTSRemux::GetPid(const uchar *Data)
+{
+  return (((uint16_t)Data[0] & PID_MASK_HI) << 8) | (Data[1] & 0xFF);
+}
+
+int cTSRemux::GetPacketLength(const uchar *Data, int Count, int Offset)
+{
+  // Returns the length of the packet starting at Offset, or -1 if Count is
+  // too small to contain the entire packet.
+  int Length = (Offset + 5 < Count) ? (Data[Offset + 4] << 8) + Data[Offset + 5] + 6 : -1;
+  if (Length > 0 && Offset + Length <= Count)
+     return Length;
+  return -1;
+}
+
+int cTSRemux::ScanVideoPacket(const uchar *Data, int Count, int Offset, uchar &PictureType)
+{
+  // Scans the video packet starting at Offset and returns its length.
+  // If the return value is -1 the packet was not completely in the buffer.
+  int Length = GetPacketLength(Data, Count, Offset);
+  if (Length > 0) {
+     if (Length >= 8) {
+        int i = Offset + 8; // the minimum length of the video packet header
+        i += Data[i] + 1;   // possible additional header bytes
+        for (; i < Offset + Length - 5; i++) {
+            if (Data[i] == 0 && Data[i + 1] == 0 && Data[i + 2] == 1) {
+               switch (Data[i + 3]) {
+                 case SC_PICTURE: PictureType = (Data[i + 5] >> 3) & 0x07;
+                                  return Length;
+                 }
+               }
+            }
+        }
+     PictureType = NO_PICTURE;
+     return Length;
+     }
+  return -1;
+}
+
diff --git a/remux/tsremux.h b/remux/tsremux.h
new file mode 100644
index 0000000..f7e4e09
--- /dev/null
+++ b/remux/tsremux.h
@@ -0,0 +1,33 @@
+#ifndef VDR_STREAMDEV_TSREMUX_H
+#define VDR_STREAMDEV_TSREMUX_H
+
+#include "libdvbmpeg/transform.h"
+#include <vdr/remux.h>
+
+#define RESULTBUFFERSIZE KILOBYTE(256)
+
+class cTSRemux {
+protected:
+  /*uchar m_ResultBuffer[RESULTBUFFERSIZE];
+	int m_ResultCount;
+	int m_ResultDelivered;
+	int m_Synced;
+	int m_Skipped;
+	int m_Sync;
+
+
+	virtual void PutTSPacket(int Pid, const uint8_t *Data) = 0;
+
+public:
+	cTSRemux(bool Sync = true);
+	virtual ~cTSRemux();
+	
+	virtual uchar *Process(const uchar *Data, int &Count, int &Result);*/
+
+	static void SetBrokenLink(uchar *Data, int Length);
+	static int GetPid(const uchar *Data);
+ 	static int GetPacketLength(const uchar *Data, int Count, int Offset);
+	static int ScanVideoPacket(const uchar *Data, int Count, int Offset, uchar &PictureType);
+};
+
+#endif // VDR_STREAMDEV_TSREMUX_H