Raspberry Pi and the HD PVR.. streaming and recording

Success! sort of...

I've finished rebuilding the C app, to read from the HDPVR and offer the data via a simple HTTP interface, and embedded a little control to toggle recording on & off. With the HDPVR running around 8mbit, I'm able to stream to about 5 clients simultaneously, which is pretty good going.

The downside is dropouts are still present.. (Update: not anymore!!) I'm hoping its just the HDPVR's fault, as these units are known to glitch occasionally & require a reopen.. even so I'm not going to call it a day yet, I'll have a bash at compiling MythTV and seeing how it's recording code gets on with the device.



I've linked the source below, with a bit of luck, you can save it as hdpvr.c on the pi, and compile it using..
   gcc -lpthread -o hdpvr hdpvr.c

Then run it using..
   ./hdpvr

You should get a response of..
   waiting for a connection

Once it's running, fire up VLC (http://www.videolan.org/) and use 'open network stream' and give it the url..
   http://ip.address.of.pi:1101/video

And then you should see the video start playing =) Check my other post if you need to select inputs other than component/stereo.

Once the video is playing, you can start recording by visiting (with a browser)..
   http://ip.address.of.pi:1101/startrec

And stop recording by visiting (with a browser)..
   http://ip.address.of.pi:1101/stoprec

Video is written to the directory you ran hdpvr from.. remember that the video streams will be large!! I've had success writing the data directly to a mounted SMB share =)

Dropouts are still present, the code will close & reopen the device.. usually the VLC client just glitches a little and then resumes playing. The recordings will need streamfixing afterwards to repair the glitches, (which will eventually end up as short cuts to the recording).. I've tried VideoRedo quick streamfix with mixed results so far.. Once I've got the Myth code recording, I'll see if I get better results.

With this code, the CPU load sits around 12% or so, so there's quite a bit of time free to still do stuff yet!!

Source:

	#include <fcntl.h>
	#include <string.h>
	#include <stdlib.h>
	#include <errno.h>
	#include <stdio.h>
	#include <netinet/in.h>
	#include <resolv.h>
	#include <sys/socket.h>
	#include <arpa/inet.h>
	#include <unistd.h>
	#include <pthread.h>
	#include <time.h>
	#include <sys/types.h>
	#include <sys/poll.h>
	#include <signal.h>
	#define MEMBUF_SIZE 4096
	#define POLL_DELAY 2500
	#define PORT 1101
	//thread prototype for the connected children.
	void* SocketHandler(void*);
	void* StreamHandler(void*);
	//data we'll share between all the threads.
	struct shared {
	pthread_mutex_t sharedlock;
	int datarelayenabled;
	int outfds[128];
	char buffer[80];
	int recording;
	int outfd;
	};
	//data we'll give to each thread uniquely.
	struct data {
	int csock;
	struct shared *data; //pointer to shared data.
	};
	int main(int argv, char** argc) {
	signal(SIGPIPE, SIG_IGN);
	int host_port=PORT;
	struct sockaddr_in my_addr;
	int hsock;
	int * p_int ;
	socklen_t addr_size = 0;
	struct sockaddr_in sadr;
	pthread_t thread_id=0;
	//init the global data.
	struct shared *global = (struct shared*)malloc(sizeof(struct shared));
	global->datarelayenabled=0;
	global->recording=0;
	pthread_mutex_init(&(global->sharedlock), NULL);
	pthread_create(&thread_id,0,&StreamHandler, (void*)global );
	pthread_detach(thread_id);
	struct data *datainst;
	hsock = socket(AF_INET, SOCK_STREAM, 0);
	if(hsock == -1) {
	printf("Error initializing socket %d\n", errno);
	goto FINISH;
	}
	p_int = (int*)malloc(sizeof(int));
	*p_int = 1;
	if( (setsockopt(hsock, SOL_SOCKET, SO_REUSEADDR, (char*)p_int, sizeof(int)) == -1 )||
	(setsockopt(hsock, SOL_SOCKET, SO_KEEPALIVE, (char*)p_int, sizeof(int)) == -1 ) ) {
	printf("Error setting options %d\n", errno);
	free(p_int);
	goto FINISH;
	}
	free(p_int);
	my_addr.sin_family = AF_INET ;
	my_addr.sin_port = htons(host_port);
	memset(&(my_addr.sin_zero), 0, 8);
	my_addr.sin_addr.s_addr = INADDR_ANY ;
	if( bind( hsock, (struct sockaddr*)&my_addr, sizeof(my_addr)) == -1 ) {
	fprintf(stderr,"Error binding to socket, make sure nothing else is listening on this port %d\n",errno);
	goto FINISH;
	}
	if(listen( hsock, 10) == -1 ) {
	fprintf(stderr, "Error listening %d\n",errno);
	goto FINISH;
	}
	//Now lets do the server stuff
	addr_size = sizeof(struct sockaddr_in);
	while(1) {
	printf("waiting for a connection\n");
	//allocate the block we'll send to the thread
	datainst = (struct data *)malloc(sizeof(struct data));
	//hook up our global shared data struct..
	datainst->data = global;
	//get the socket, store into struct.
	if((datainst->csock = accept( hsock, (struct sockaddr*)&sadr, &addr_size))!= -1) {
	pthread_mutex_lock(&(global->sharedlock));
	printf("-%d--------------------\nReceived connection from %s\n",global->datarelayenabled,inet_ntoa(sadr.sin_addr));
	pthread_mutex_unlock(&(global->sharedlock));
	pthread_create(&thread_id,0,&SocketHandler, (void*)datainst );
	pthread_detach(thread_id);
	} else {
	fprintf(stderr, "Error accepting %d\n", errno);
	}
	}
	FINISH:
	free(global);
	return 0;
	}
	void* StreamHandler(void* lp) {
	struct shared *global = (struct shared *)lp;
	int devfd;
	char *ifname;
	struct pollfd fds[2];
	void *membuf;
	ssize_t nbytes;
	time_t timer;
	char buffer[80];
	struct tm* tm_info;
	int outfdcount;
	int retval;
	ifname = "/dev/video0";
	if(NULL == (membuf = malloc(MEMBUF_SIZE))) {
	printf("Not enough memory to allocate buffer\n");
	fprintf(stderr, "Not enough memory\n");
	exit(EXIT_FAILURE);
	}
	while(1) {
	int enabled=0;
	pthread_mutex_lock(&(global->sharedlock));
	enabled = global->datarelayenabled;
	pthread_mutex_unlock(&(global->sharedlock));
	while(!enabled) {
	sleep(1);
	pthread_mutex_lock(&(global->sharedlock));
	enabled = global->datarelayenabled;
	pthread_mutex_unlock(&(global->sharedlock));
	}
	//someone enabled the datarelay, we'd better setup & start reading data.
	/** open the device **/
	if(-1 == (devfd = open(ifname, O_RDWR | O_NONBLOCK))) {
	perror("Unable to open device");
	exit(EXIT_FAILURE);
	}
	usleep(5000);
	/** setup descriptors for event polling **/
	fds[0].fd = STDIN_FILENO;
	fds[0].events = POLLIN;
	fds[1].fd = devfd;
	fds[1].events = POLLIN;
	/** start the recording loop **/
	int countdown=0;
	while(enabled) {
	pthread_mutex_lock(&(global->sharedlock));
	enabled = global->datarelayenabled;
	pthread_mutex_unlock(&(global->sharedlock));
	while(countdown>0) {
	retval = poll(fds, 2, POLL_DELAY);
	if(0 == retval) {
	time(&timer);
	tm_info = localtime(&timer);
	strftime(buffer,80,"%Y-%m-%d %H:%M:%S", tm_info);
	fprintf(stderr, "%s Waiting for ready (%d)...\n", buffer, countdown);
	usleep(300);
	countdown--;
	} else {
	countdown=0;
	}
	}
	retval = poll(fds, 2, POLL_DELAY);
	if(0 == retval) {
	time(&timer);
	tm_info = localtime(&timer);
	strftime(buffer,80,"%Y-%m-%d %H:%M:%S", tm_info);
	fprintf(stderr, "%s Lost signal, restarting device...\n",buffer);
	close(devfd);
	countdown=5;
	if(-1 == (devfd = open(ifname, O_RDWR | O_NONBLOCK))) {
	perror("Unable to reopen the device");
	exit(EXIT_FAILURE);
	} else {
	fds[1].fd = devfd;
	fds[1].events = POLLIN;
	fprintf(stderr,"%s Device reaquired. Usleep for 5k for data\n",buffer);
	usleep(5000);
	continue;
	}
	} else if(-1 == retval) {
	printf("polling failed\n");
	perror("Polling failed");
	break;
	} else if(fds[0].revents & POLLIN) {
	printf("user quit\n");
	fprintf(stderr, "User quit\n");
	break;
	} else if(fds[1].revents & POLLIN) {
	nbytes = read(devfd, membuf, MEMBUF_SIZE);
	if(0 > nbytes) {
	switch(errno) {
	case EINTR:
	case EAGAIN:{
	usleep(2500);
	continue;
	}
	default:
	printf("Unknown errno response %d when reading device\n",errno);
	perror("Unknown");
	exit(EXIT_FAILURE);
	}
	} else if(MEMBUF_SIZE == nbytes) {
	pthread_mutex_lock(&(global->sharedlock));
	//if recording.. write out to outfd.
	if(global->recording){
	ssize_t written = write(global->outfd, membuf, MEMBUF_SIZE);
	if(written==-1) {
	perror("Error writing to file.");
	}
	}
	//iterate over the output fd's.. set them to -1 if they fail to write.
	for(outfdcount=0; outfdcount<(global->datarelayenabled); outfdcount++) {
	if(global->outfds[outfdcount]!=-1) {
	ssize_t written = write(global->outfds[outfdcount], membuf, MEMBUF_SIZE);
	if(written==-1) {
	global->outfds[outfdcount]=-1;
	} else {
	fsync(global->outfds[outfdcount]);
	}
	}
	}
	//iterate over the outputfd's.. collapsing the array to move the valids to the front.
	int writepos=0;
	int currentmax=global->datarelayenabled;
	for(outfdcount=0; outfdcount<currentmax; outfdcount++) {
	if(global->outfds[outfdcount] != -1) {
	if(writepos!=outfdcount) {
	//move the data back to the writepos slot, and set self to -1..
	global->outfds[writepos] = global->outfds[outfdcount];
	global->outfds[outfdcount] = -1;
	}
	writepos++;
	} else {
	global->datarelayenabled--;
	}
	}
	pthread_mutex_unlock(&(global->sharedlock));
	continue;
	} else {
	printf("Short read\n");
	perror("Short read");
	exit(EXIT_FAILURE);
	}
	} else if(fds[1].revents & POLLERR) {
	printf("Pollerr\n");
	perror("pollerr");
	exit(EXIT_FAILURE);
	break;
	}
	}
	/** clean up **/
	close(devfd);
	}
	free(membuf);
	return 0;
	}
	void* SocketHandler(void* lp) {
	struct data *datainst = (struct data *)lp;
	int csock = datainst->csock; //(int*)lp;
	struct shared *global = datainst->data;
	char buffer[1024];
	int buffer_len = 1024;
	int bytecount;
	memset(buffer, 0, buffer_len);
	if((bytecount = recv(csock, buffer, buffer_len, 0))== -1) {
	fprintf(stderr, "Error receiving data %d\n", errno);
	goto FINISH;
	}
	printf("Received bytes %d\nReceived string \"%s\"\n", bytecount, buffer);
	strcat(buffer, " SERVER ECHO");
	if(strstr(buffer,"GET /startrec HTTP/1.1")) {
	char *text="HTTP/1.0 200 OK\nContent-Type: text/plain\n\n Told to Start Recording : Sent at : ";
	send(csock,text,strlen(text),0);
	int outfd;
	char tbuf[80];
	time_t timer;
	struct tm* tm_info;
	time(&timer);
	tm_info = localtime(&timer);
	strftime(tbuf,80,"%Y-%m-%d %H:%M:%S", tm_info);
	send(csock,tbuf,strlen(tbuf),0);
	strftime(tbuf,80,"%Y%m%d%H%M%S.ts", tm_info);
	/** open the output file **/
	if(-1 == (outfd = open(tbuf, O_CREAT | O_RDWR, S_IRWXU | S_IRWXG))) {
	perror("Unable to open output file");
	} else {
	int started=0;
	pthread_mutex_lock(&(global->sharedlock));
	if(global->recording==0) {
	global->outfd=outfd;
	global->recording=1;
	started=1;
	}
	pthread_mutex_unlock(&(global->sharedlock));
	if(!started) {
	char *alreadyrecording="\n\nAlready recording, new request ignored\n";
	send(csock,alreadyrecording,strlen(alreadyrecording),0);
	} else {
	char *nowrecording="\n\nRecording now in progress\n";
	send(csock,nowrecording,strlen(nowrecording),0);
	}
	}
	fsync(csock);
	close(csock);
	printf("Sent bytes %d\n", bytecount);
	} else if(strstr(buffer,"GET /stoprec HTTP/1.1")) {
	char *text="HTTP/1.0 200 OK\nContent-Type: text/plain\n\n Told to Stop Recording : Sent at : ";
	send(csock,text,strlen(text),0);
	int outfd;
	char tbuf[80];
	time_t timer;
	struct tm* tm_info;
	time(&timer);
	tm_info = localtime(&timer);
	strftime(tbuf,80,"%Y-%m-%d %H:%M:%S", tm_info);
	send(csock,tbuf,strlen(tbuf),0);
	int stopped=0;
	pthread_mutex_lock(&(global->sharedlock));
	if(global->recording==1) {
	outfd=global->outfd;
	global->outfd=-1;
	global->recording=0;
	stopped=1;
	}
	pthread_mutex_unlock(&(global->sharedlock));
	if(!stopped) {
	char *notrecording="\n\nRecording not in progress, stop request ignored\n";
	send(csock,notrecording,strlen(notrecording),0);
	} else {
	fsync(outfd);
	close(outfd);
	char *stoppedok="\n\nRecording stopped.\n";
	send(csock,stoppedok,strlen(stoppedok),0);
	}
	fsync(csock);
	close(csock);
	printf("Sent bytes %d\n", bytecount);
	} else if(strstr(buffer,"GET /video HTTP/1.1")) {
	if(global->datarelayenabled < 128) {
	char *text="HTTP/1.0 200 OK\nContent-Type: video/h264\nSync-Point: no\nPre-roll: no\nMedia-Start: invalid\nMedia-End: invalid\nStream-Start: invalid\nStream-End: invalid\n\n";
	send(csock,text,strlen(text),0);
	char ofname[32];
	time_t curtime;
	struct tm *fmttime;
	/** set the output file name to time of creation **/
	time(&curtime);
	fmttime = localtime(&curtime);
	strftime(ofname, 32, "%Y%m%d%H%M%S.ts", fmttime);
	pthread_mutex_lock(&(global->sharedlock));
	global->outfds[global->datarelayenabled]=csock;
	global->datarelayenabled++;
	pthread_mutex_unlock(&(global->sharedlock));
	}
	} else {
	char *text="HTTP/1.0 200 OK\nContent-Type: text/plain\n\nUnknown URL requested, Response Sent at : ";
	send(csock,text,strlen(text),0);
	char tbuf[80];
	time_t timer;
	struct tm* tm_info;
	time(&timer);
	tm_info = localtime(&timer);
	strftime(tbuf,80,"%Y-%m-%d %H:%M:%S", tm_info);
	send(csock,tbuf,strlen(tbuf),0);
	char *text2="\nOriginal Headers:\n\n";
	send(csock,text2,strlen(text2),0);
	if((bytecount = send(csock, buffer, strlen(buffer), 0))== -1) {
	fprintf(stderr, "Error sending data %d\n", errno);
	goto FINISH;
	}
	fsync(csock);
	close(csock);
	printf("Sent bytes %d\n", bytecount);
	}
	FINISH:
	free(lp);
	return 0;
	}

view raw gistfile1.c hosted with ❤ by GitHub