qstream

#!/usr/bin/perl

# compute packet stream quality statistics for UDP/RTP from network or captured files(pcap)
# Base Olav Kvittem
# Mpeg support Odd Rune Mykkeltveit Lykkbo,  2009-2010
# Mos values computation Gurvinder Singh, 2010-05-07
# XML, RTMP, TCP support, Pål Moen Møst, 2011.

# use Socket;
use IO::Socket qw(:DEFAULT :crlf);
use IO::Socket::INET;
use IO::Socket::Multicast;
use IO::Select;
use Net::Pcap;
use NetPacket::Ethernet qw(:strip);
use NetPacket::IP;
use NetPacket::IP qw(:strip  :protos);;
use NetPacket::UDP;
use NetPacket::TCP;
use Net::IPv6Addr;
#use NetPacket::IPv6 qw(ipv6_strip);
#use Net::RTP;
use Net::RTP::Packet;
use Time::HiRes qw/ sleep time tv_interval gettimeofday/;
use Data::Dumper;
use threads;
use threads::shared;
use Pod::Usage;
#use bigint qw/ hex oct/;
use Math::BigInt;
use Math::BigFloat;
use Carp;
#RTMP implementation *experimental*
use rtmp::client qw(rtmp_connect rtmp_play rtmp_call get_chunk_ids);
use constant MTU => 1580;
#XML
use XML::DOM;
use XML::Twig;
#use mpegpackets;
$DEFAULT_PORT = 1955;
my $timeout=1 ; # frac seconds to wait for receiving packets

#my oh my,more globals
my ($rtmp,$period,$last,$debug,$swfurl,$pageurl,$flash_version,$reconnect);

require "newgetopt.pl";

@opts=('list', 'man','fullformat', 'net', 'rtp', 'mpeg', 'pcap', 'crude', 'bins=s', 'log', 'nperiod=s', 'period=s', 'last=s', 'packets=s', 'src=s', 'flow_key=s', 'flow_no=s', 'flow_min=s', 'nohead', 'dump=s', 'format=s', 'id=s', 'sum=s', 'ttl', 'rtt=s', 'codec=s', 'log', 'verbose', 'v', 'h', 'debug', 'av','xml=s', 'rtmp','tcp', 'port=s','swfurl=s','pageurl=s','flashversion=s','new','q','exclude=s','report=s', 'filename','test','ipv6','p=s','name=s','tcp_t');

&NGetOpt(@opts) || die pod2usage(1);
die pod2usage(1)  if $opt_h;
$continous= ! $opt_packets;
$opt_v=$opt_v or $opt_verbose;
my $codec = $opt_codec || 'G.711';
my $rtt = $opt_rtt || 10;		# ms round trip time of the connection
pod2usage(-verbose => 2 ) if $opt_man;

my @streams=@ARGV;

if ($opt_bins){
    @bins=split(/,/, $opt_bins);
}
if ($opt_report){
    @filter=split(/,/, $opt_report);
}
my $format = $opt_format || "pretty";

my %hdr=();
my $err="";
my $us=0; # current packet time in us
my $nkill=0;
my $nint=0;
my $n_packets=0;
my $endstream=0;
my $period_num = 0;
my $pcr_count = 0;
my $pts_count =  0;
my $doc;
my $xml;
my $xml_printed= 0;
my $last_stream;
my $file_flag;
my $offline;
my $lost_counter  = 0;

#TCP STUFF
my $thr;
my $rtmp_port:shared = 1935; 
my $rtmp_addr:shared;
my %tcp_packet_stats:shared = ();
my %bw_stats:shared = ();
my %seq_stats:shared = ();
my %seq_stats_n =();
my $n_seq_num:shared=0;
my $rtmp_stream_start:shared;
my $tcp_stop:shared;
my %hash : shared;

my $packet_re_num:shared=0;
$seq_stats{'p_seq_num'} = 0;
$seq_stats{'p_tcp_time_stamp'} = 0;
$seq_stats{'p_tcp_time_stamp_echo'} = 0;
$seq_stats{'real_last_time'} = 0;
#init packetstats
$tcp_packet_stats{'pkt_lost'} = 0;
$tcp_packet_stats{'pkt_late'} = 0;
$tcp_packet_stats{'pkt_dup'} = 0 ;
$tcp_packet_stats{'pkt_retrans'} = 0;

my %offset_bytes =();
my %network_info =();

my $start_time = time();
#MPEG-TS measurement globals (most are reset in init)
my $CHECK_NUM_TSP = 2;
#globals for mpeg ts 
#a tabel for transport streams pids. aleternativ to PAT impl.
my %pcr_pid = ();
my $TS;
my @cc = ();
my $discontinuities = 0;
my %last_ccs = ();
my $packets_between_pcr = 0;
#my $PTS_MAX = 95443.7176888888888888
my $PTS_MAX = 95443;
my $PCR_MAX = 95443;
# pcr_stats {filename/handle}
my %pcr_stats = ();
#PTS-stats {filename/handle}
my %pts_stats = ();
my %mpeg_stats = ();
my %jitter_stats = ();

#PTS stuff
#my $real_time = time();
my $late_video_packets = 0;

#CC distribution
my $num_cc = 0;
my %cc_dist = ();
my %pid_next_cc = ();

my %dist1 = ();
my %dist2 = ();

$opt_period = $opt_last if not $opt_period;

$file_flag = 1 if $opt_filename;

#MPEG-TS END


$SIG{USR2} = sub {confess "Caught by SIGUSR2"; };

$SIG{INT} = sub { $uninterrupted=0; # return if $nint++  < 1;
		  &display_stats() if  !$endstream;
		  &end_xml() if $opt_xml;
		  &handle_threads()  if ($opt_tcp and $opt_rtmp);
		  
		  die "End after interrupt.\n";exit(0) };

$SIG{TERM} = sub { $uninterrupted=0; # return if $nkill++ < 1;
		   &display_stats() if  !$endstream;  
		   &end_xml() if $opt_xml;
		   &handle_threads()  if ($opt_tcp and $opt_rtmp);

		  die "End after sigterm.\n";exit(0) };

$SIG{ALRM} = sub { 
#   Writes stats to file if option XML is given. -PMM
#    die "No packets received" if $n_packets < 1;
#    $uninterrupted=0;
    if (! $endstream) {
	$endstream=1;
	&display_stats();
    }
#	die "End after alarm\n"; 
#	return(0);
    print "endstream=1" if $opt_debug;
};

if ($opt_dump){
    open DUMP, ">$opt_dump";
}

if ($opt_test){
    if($opt_mpeg || $opt_rtp || $opt_tcp_t){
	&unit_test;
    }
    else {
	print "You need to choose a test. (mpeg, tcp or rtp) \n";
    }
}

if ($opt_pcap || ($opt_tcp && !$opt_rtmp)){ # listen in parallell
    alarm($opt_last) if $opt_last; 
    $uninterrupted=1;
    $endstream=0;
    eat_pcap_streams(@streams);
    &display_stats() if !$opt_xml and !$endstream;
} else { # listen serially
    my $stream_count;
    #my $ln = @streams;
    foreach $id (@streams) {
	alarm($opt_last) if $opt_last;
	++$stream_count;
	#$last_stream = 1 if $stream_count == ($ln);
	#print "count $stream_count  lengde  $ln \n";
	$uninterrupted=1;
	$endstream=0;
	init_file($id);
	eat_stream($id) if (!($opt_tcp || $opt_rtmp));
	&rtmp_eat_stream($id,$opt_period, $opt_last,0,$opt_debug,$opt_swfurl,$opt_pageurl,$opt_flashversion) if $opt_rtmp;
	#&eat_tcp($id,$opt_port) if $opt_tcp && !$opt_rtmp;
	if (! ( $opt_list || $opt_sum)){
	    &display_stats() if !$endstream;
	}
	last if ! $uninterrupted;
    }
    
}
&list_flows if $opt_list;
&display_stats() if $opt_sum and !$endstream;
&display_bins if $opt_bins;
&end_xml() if $opt_xml;

exit(0);
#
sub unit_test {
    my $test_data = shift;
    #streams =  "rtp_158.38.130.110_6711_1.pcap";
    if ($opt_mpeg){
	@streams = "testing/pcap/testdata_mpeg.pcap";
	@filter = qw "video audio";
    }
    @streams = "testing/pcap/testdata_rtp.pcap" if $opt_rtp;
    @streams = "testing/pcap/testdata_tcp.pcap" if $opt_tcp_t;
    $opt_xml = "tmp_testdata.xml";
    $opt_q =  1;
    $opt_v = 1;
    #print $streams[0], "\n";
}
sub unit_test_2{
    #my ($orgnial_test) = &feed_data("file",rtp_file.xml);
    my $test_data;
    $test_data = "testing/xml/testdata_mpeg.xml" if $opt_mpeg;
    $test_data = "testing/xml/testdata_rtp.xml" if $opt_rtp;
    #$test_data = "udp_testdata" if $opt_net;
    $test_data = "testing/xml/testdata_tcp.xml" if $opt_tcp;
    my ($orginal_test) = &xml_parser($test_data);
    my ($new_test) = &xml_parser("tmp_testdata.xml");
    print Dumper($orginal_test), "\n", Dumper($new_test)if $opt_debug;
    my $ignore_pattern = "jitter";  #PCR jitter is based on delta local time. 
    my $ignore_pattern2  = "dwell";
    my $ignore_pattern3 ="delay_factor";
    my $ignore_pattern4 ="gap";
    print "Checking values ....\n";
    for my $k1(keys %$orginal_test){
	if ($orginal_test->{ $k1} != $new_test->{ $k1} and $orginal_test->{ $k1} !~ m/HASH/) {
	    print "different value @ $k1, value: $orginal_test->{$k1}, differs from value: $new_test->{$k1}\n";
	}
	else {
	    printf "%2s %2s\n", "OK  @ ", "$k1 ";
	}
	for my $k2 ( keys %{$orginal_test->{ $k1}} ) {
	    if ( ($orginal_test->{ $k1}{$k2} != $new_test->{ $k1}{$k2}) and ($k2 !~ m/$ignore_pattern/) and ($k2 !~ m/$ignore_pattern2/) and ($k2 !~ m/$ignore_pattern3/) and ($k2 !~ m/$ignore_pattern4/)){
		print "different value @ $k1-> $k2, value: $orginal_test->{$k1}{$k2} differs from, value: $new_test->{$k1}{$k2}\n";
	    }
	    
	}
    }
    print "Test complete \n";
}

sub xml_parser {
    my $file = shift;
    my $parser = new XML::DOM::Parser;
    my $doc = $parser->parsefile ($file);
    my $root = $doc->getFirstChild();
    my $nodelist = $root->getChildNodes();
    my $node = $nodelist->item(1);
    my $dup;
    my $flow_childs = $node->getChildNodes();
    my @test_array = ();
    my %test_hash = ();
    foreach my $child ($node->getChildNodes()) {
	if ($child->getNodeType == ELEMENT_NODE) {
	    if ($child->getChildNodes()){
		my @children = $child->getChildNodes();
		my $l = @children;
		if ($l > 1){
		    #print "=== " ,$child->getNodeName(), " === \n";
		    if (exists $test_hash{$child->getNodeName()}){
			$dup +=1;
		    }
		    foreach my $grandchild ($child->getChildNodes()){
			if ($grandchild->getNodeType == ELEMENT_NODE){
			    #print $grandchild->getNodeName(), "\n";
			    eval {
				$test_hash{$child->getNodeName().$dup}{$grandchild->getNodeName()}=  $grandchild->getFirstChild()->getData();
			    };
			    if ($@){
			    };
			}
		    }
		}
		else {
		    $test_hash{'flow'}{$child->getNodeName()} =  $child->getFirstChild()->getData();
     		}	
	    }
	} 
    }
    return (\%test_hash)
}

sub http_stream_client {
    my $f = shift;
    my $playlist = get("$f");
    my @list_items  = split(/\n/,$playlist);
    my @streams;
    my $duration;
    my $regex = "http://";
    my $p_chunk;
    while(1){
	foreach my $item (@list_items){
	    my @tmp_streams= get("$item");
	    if ($item=~ /$regex/g){
		foreach my $stream (@tmp_streams){
		    if ($stream =~ /EXT-X-TARGETDURATION/){
			my @temp =  split(/:/,$stream);
			$duration = $temp[1];
		    }
		    if($stream !~ /#/g){
			push(@streams,$stream);
			my $l = @streams;
			my $current_chunk = $streams[$l-1];
			if ($current_chunk ne $p_chunk){
			    my $data  = get("$current_chunk");
			    $p_chunk = $current_chunk;
			    sleep(9);
			}
		    }
		}
	    }
	}
    }
}
sub get_stream {
    my $uri = shift;
    my $data = get("$uri");
}

##TCP impl. -PMM
sub handle_threads {
    $tcp_stop = 1;
    $thr->join();
}
sub end_xml{
    my $filename;
    my $twig = XML::Twig->new(pretty_print => 'indented');
    my ($date,$time)  = &get_date_time();
    $filename = $opt_xml;
    $filename = $date."_".$time.".xml" if $opt_xml eq "date_time";
    open (MYFILE, '>'.$filename) || die "Could not open $filename : $!";
    my $final_xml = $doc->toString();
    $twig->parse($final_xml);
    $twig->flush(\*MYFILE);
    #print "</qstream> \n"  if $opt_xml eq "-";
    $doc->dispose;
    close (MYFILE);
    $xml_printed = 1;
    &unit_test_2($filename) if $opt_test;
}

sub tcp_packet_stats {
    my $f = shift;
    my $ip = shift;
    my ($tcp_obj) = shift;
    my $len,$ip_hlen = 0;
    #ipv6.pm uses different variables. 
    if($opt_ipv6){
	$len = plen;
    }
    else {
	$len = len;
	$ip_hlen = 20; 
    }
    #my $ip = NetPacket::IP->decode(eth_strip($pkt));
    #my $tcp_obj = NetPacket::TCP->decode(ip_strip(eth_strip($pkt)));
    #my $f = $ip->{src_ip}.":".$tcp_obj->{src_port} if !$offline;
    my $seq_num = $tcp_obj->{seqnum};
    my $options = $tcp_obj->{options}; #12 bytes
    #print "$tcp_obj->{src_port} \n";
    if (!exists $tcp_packet_stats{$f}){
	&init_stats($f);
	$tcp_packet_stats{$f} = &share( {} );
	$seq_stats{$f} = &share( {} );
	$seq_stats_n{$f} = &share( {} );
	$tcp_packet_stats{$f}{'pkt_late'} = 0;
	$tcp_packet_stats{$f}{'pkt_retrans'} =0;
	$tcp_packet_stats{$f}{'pkt_dup'} = 0;
	
    }
    ++$tcp_packet_stats{$f}{'num_packets'};
    &tcp_jitter_stats($f,$options);
    #Bandwidth messurements
    $tcp_packet_stats{$f}{'num_bytes'} += $ip->{$len}; #counts the bytes
    #Check for duplicated,out-of-order, late and lost packets.
    $seq_stats{$f}{'seq_diff'} = $seq_num - $seq_stats{$f}{'p_seq_num'} - $seq_stats{$f}{'p_packet_len'};
    if ($seq_stats_n{$f}{'seq_num_p'} == $seq_num && $seq_stats{$f}{'p_seq_num'} != 0) {
	++$tcp_packet_stats{$f}{'pkt_dup'};
    }
    if ($seq_stats{$f}{'seq_diff'} != 0 && $seq_stats{$f}{'p_seq_num'} != 0){
	if ($seq_stats{$f}{'seq_diff'} < 0){
	    ++$tcp_packet_stats{$f}{'pkt_late'};
	    if ($seq_stats_n{$f}{'seq_num_n'} == $seq_num){
		++$tcp_packet_stats{$f}{'pkt_retrans'};
	    }
	}
	else {
	    ++$tcp_packet_stats{$f}{'pkt_lost'};
	}
    }
    if($seq_stats{$f}{'p_seq_num'} < $seq_num){
	$seq_stats{$f}{'n_seq_num'} = ($seq_num+$ip->{$len} -($ip_hlen)-($tcp_obj->{hlen}*4))
    } 
    $seq_stats{$f}{'p_seq_num'} = $seq_num;
    $seq_stats{$f}{'p_packet_len'} = $ip->{$len}-($ip_hlen)-($tcp_obj->{hlen}*4);
    $seq_stats_n{$f}{'seq_num_p'} = $seq_num;
    $seq_stats_n{$f}{'seq_num_n'} = $seq_stats{$f}{'n_seq_num'};
}
sub tcp_jitter_stats {
    my $f = shift;
    my $options = shift;
    my $tcp_time_stamp_echo = unpack("N",substr($options,8,4));#from our client
    my $tcp_time_stamp_delta = $tcp_time_stamp - $seq_stats{$f}{'p_tcp_time_stamp'};
    my $tcp_time_stamp_echo_delta = $tcp_time_stamp_echo - $seq_stats{$f}{'p_tcp_time_stamp_echo'}; 
    $seq_stats{$f}{'p_tcp_time_stamp'} = $tcp_time_stamp;
    $seq_stats{$f}{'p_tcp_time_stamp_echo'} = $tcp_time_stamp_echo;
    my $real_time_delta = time() - $seq_stats{$f}{'real_last_time'};
    $seq_stats{$f}{'real_last_time'} = time();
    
    #After AV streams starts
    #$first_av_packet = &get_stream_start() if $opt_rtmp;
    #$first_av_packet = $tcp_packet_stats{'num_packets'} if !$opt_rtmp and $opt_tcp;
    
    if ($tcp_packet_stats{$f}{'num_packets'} > 1){
	#gaps
	$tcp_packet_stats{$f}{'gap_sum'} += $real_time_delta;
	$tcp_packet_stats{$f}{'gap_square'} +=$real_time_delta**2;
	if ($tcp_packet_stats{$f}{'gap_max'} < $real_time_delta){
	    $tcp_packet_stats{$f}{'gap_max'} = (10**3)*$real_time_delta;
	}
	if ($tcp_packet_stats{$f}{'gap_min'} > $real_time_delta){
	    $tcp_packet_stats{$f}{'gap_min'} = (10**3)*$real_time_delta;
	}
	my $jitter  = abs($tcp_time_stamp_delta - $real_time_delta*10**3);
	if($jitter  > $tcp_packet_stats{$f}{'jitter_max'} ){
	    $tcp_packet_stats{$f}{'jitter_max'}  = $jitter; 
	}
	if($jitter  < $tcp_packet_stats{$f}{'jitter_min'} ){
	    $tcp_packet_stats{$f}{'jitter_min'}  = $jitter;
	}
	$tcp_packet_stats{$f}{'jitter_sum'} += $jitter;
	$tcp_packet_stats{$f}{'jitter_square'} += $jitter**2;
    }
    else {
	$tcp_packet_stats{$f}{'jitter_min'} = 2**31;
	$tcp_packet_stats{$f}{'jitter_max'} =  0;
	$tcp_packet_stats{$f}{'gap_max'} =  0;
	$tcp_packet_stats{$f}{'gap_min'} = 2*31;
    }
}
sub check_period {
    my $f =  shift;
    if((($opt_last)+$start_time) < time() || (($opt_period)+$start_time) < time()){
	$tcp_stop=1;
	my $span = time() - $start_time;
	&do_math($f,$span);
    }
}
sub eat_tcp {
    my $ip;
    my $port;
    if (!$opt_rtmp) {
	$uri = shift;
	$port = shift;
	if ($ip  !~ /^\d+\.\d+\.\d+\.\d+$/){
	    ($ip,$port) =  prepare_ip($uri);
	}
	my $f = $ip.":".$port;
	if (! exists($tstart{$f})){
	    $tstart{$f} = [gettimeofday]; 
	    $first_us{$f}=$prev1s_us{$f}=$prevt100ms_us{$f}=$us;
	    if (! exists($setuptime{$f}) ){
		if ( $pinterval){
		    $setuptime{$f}=0;
		} else {
		    $setuptime{$f}=tv_interval ( $tjoined{$f}, $tstart{$f}) * 10**3;
		}
		printf "setup $f : %.3f\n",  $setuptime{$f} if $opt_debug;
	    }
	}
    }
    $ip = $rtmp_addr if $opt_rtmp;
    $port = 1935 unless $port;
    my ($net, $netmask, $filter_t, $err, $dev);
    #my $time_out_ms = ($opt_last*10**3) || ($opt_period*10**3) if !$opt_last;
    my $time_out_ms = 2000;
    my $filter_str = "( (src $ip and port $port ) )";
    $dev = Net::Pcap::lookupdev(\$err);  # find a device	
    my $r = Net::Pcap::lookupnet( $dev, \$net, \$netmask, \$err);
    my $pcap = Net::Pcap::open_live($dev, 1514, 1, $time_out_ms, \$err) || die "You need to run this as sudo $err";
    $r=Net::Pcap::compile($pcap, \$filter_t, $filter_str, 0,$netmask);
    $r=Net::Pcap::setfilter($pcap, $filter_t);
    while (my $pkt=Net::Pcap::next($pcap,\%hdr)) { # get all packets
	
	my $ip_obj = NetPacket::IP->decode(eth_strip($pkt));
	my $tcp_obj = NetPacket::TCP->decode(ip_strip(eth_strip($pkt)));
	if($tcp_obj->{src_port} == $port && $ip_obj->{src_ip} eq $ip ){
	    my $f = $ip_obj->{src_ip}.":".$tcp_obj->{src_port};
	    check_period($f) if $opt_last;
	    last if $tcp_stop==1;
	    &tcp_packet_stats($pkt);
	}
    }
}
##RTMP impl. -PMM
sub rtmp_eat_stream {
    ($rtmp,$period,$last,$reconnect,$debug,$swfurl,$pageurl,$flash_version) = @_;
    ($address, $app, $play) = rtmp_prepare_url($rtmp);
    $rtmp_addr = $address;
    $port = 1935 unless $opt_port;
    my $f = $address.":".$port;
    while(!$endstream){
	$start_time = time();
	if (! exists($tstart{$f})){
	    $tstart{$f} = [gettimeofday]; 
	    $first_us{$f}=$prev1s_us{$f}=$prevt100ms_us{$f}=$us;
	    if (! exists($setuptime{$f}) ){
		if ( $pinterval){
		    $setuptime{$f}=0;
		} else {
		    $setuptime{$f}=tv_interval ( $tjoined{$f}, $tstart{$f}) * 10**3;
		}
		printf "setup $f : %.3f\n",  $setuptime{$f} if $opt_debug;
	    }
	}
	$thr = threads->new(\&eat_tcp) if $opt_tcp and !$reconnect;
	rtmp_connect($address, '1935', $app , $swfurl, $pageurl,$flash_version) if $opt_rtmp;
	rtmp_play($play, '-1.0', '-1.0', $period, $last,$debug)  if $opt_rtmp;
	&display_stats() if ($opt_period != $opt_last) and !$endstream;
	my $pkt_stat = $thr->join if $opt_tcp;
	last if  $endstream;
	
    }
}


sub rtmp_prepare_url{
    my $uri = shift;
    #	return 0  if $uri != m/^(rtmp)://; 
    $uri =~ s/^(rtmp)://;
    $uri =~ s#//##;
    @string = split(/\//, $uri);
    my $length = $#string;
    my $host = $string[0];
    my $app = '';
    if ($host !~ /^\d+\.\d+\.\d+\.\d+$/){ # hostname
	my $padr=gethostbyname($host) || die "Gethostbyname - $host - $padr : $!";
	$address=Socket::inet_ntoa($padr);
    }    
    for (my $i = 1; $i < $length;$i++){
	$app .= $string[$i];
	if ($i > 1 ){$app .="/";}
    }
    my $play = $string[$length];
    return ($address,$app,$play)
}
#------------------------------------------------------------------------------

# unpack uri's to ip-address
sub prepare_ip {
    my $uri=shift;
    $uri =~ s/^(udp|rtp|http)://;
    $uri =~ s#//##;
    die "Unsupported streaming $1" if $uri =~/^(\w:)/;
    if ($multicast = ($uri =~ /@/)){
	($source, $uri)=split(/@/, $uri);
    }

    my ($host, $port)=split(/[\/:]/, $uri);
    die $usage . "undefined address inn $host" unless (defined $host);
    $port = $DEFAULT_PORT unless (defined $port);
    my $address=$host;
    if ($host !~ /^\d+\.\d+\.\d+\.\d+$/){ # hostname
	my $padr=gethostbyname($host) || die "Gethostbyname - $host - $padr : $!";
	$address=Socket::inet_ntoa($padr);
    }   
    my @nib=split(/\./,$address);
    $multicast = $multicast || $nib[0] >= 224 && $nib[0] <= 240;
    return ($address, $port, $multicast);

}

sub eat_stream {
    my $id=shift;
    $tstart{$id}=[gettimeofday];
    if ($opt_net && $opt_rtp){
	# Create RTP socket
	my $elapsed=0;
	my $t0=0.0;
	$n_packets=0;
	my ($address, $port, $multicast) = prepare_ip($id);
	my $rtp = new Net::RTP( LocalPort=>$port,
				LocalAddr=>$address ) || die "Failed to create RTP socket: $!";
	my $select=IO::Select->new($rtp->fileno);

	my $tjoined=[gettimeofday];
	if ($opt_align){ # align to $opt_period
	}

	while ($uninterrupted && !($opt_packets && $n_packets > $opt_packets)) {
	    my $packet;
	    last if ! $uninterrupted || $endstream;
	    if ($select->can_read($timeout) && ($packet = $rtp->recv())){
		my $ssrc = $packet->ssrc();
		$cur_id=$packet->source_ip . ", $address:$port";

		if ( !$file{$cur_id}){
		    push(@myflows, $cur_id);
		    $file{$cur_id}=$id;
		}
		$n_packets++;
		next if $opt_src and $opt_src ne $cur_id;

		$tc=[gettimeofday]; # current flow time
		if( $t0){
		    $elapsed=tv_interval ( $t0, $tc);
		} else {
		    $t0 = $tc;
		    $elapsed=0.0;
		    &init_stats($cur_id);
		}
		$us=$elapsed*10**6;
		$tjoined{$cur_id} = $tjoined if !$tjoined{$cur_id};
		# No stats for that SSRC yet?
		&pkt_stats($cur_id, $packet->encode, $elapsed * 10**6, $packet->size());
		
		if ($opt_dump){
		    print DUMP $packet;
		}
	    } else { # anticipate interrupted by alarm	
#		warn "Failed to receive $n_packets packets from $address:$port : $!";
#		next; #### exit
	    }
	}
#	&display_stats();

#	$mc->shutdown; # gives error message

    } elsif ($opt_net) {			# udp stream

	# Create UDP socket
	my $elapsed=0;
	my $t0=0.0;
        $n_packets=0;
	my ($address, $port, $multicast) = prepare_ip($id);
	my $mc;

	if ( $multicast){
	    $mc = IO::Socket::Multicast->new(LocalPort=>$port, LocalAddr=>$address, ReuseAddr=>1) ||
		die "Could not connect to port $port : $!";
	    $mc->mcast_add($address) || die "could not join $address:$port : $!";
	} else { # unicast
	    $mc = IO::Socket::INET->new( LocalAddr=> $address, LocalPort => $port);
	}
	my $tjoined=[gettimeofday];

	my $select=IO::Select->new($mc->fileno);

	if ($opt_align){ # align to $opt_period
	}

	while ($uninterrupted && ! $endstream && !($opt_packets && $n_packets > $opt_packets)){
	    
	    if ($select->can_read($timeout) && $mc->recv($packet, 1560)) {
		last if ! $uninterrupted || $endstream;
		$n_packets++;
		my $ssrc = $mc->peerhost;
		my $cur_id = $mc->peerhost . ":" . $mc->peerport  . "->$address:$port";
		if ( !$file{$cur_id}){
		    push(@myflows, $cur_id);
		    $file{$cur_id}=$id;
		}
		
		next if $opt_src and $opt_src ne $cur_id;

		$tc=[gettimeofday]; # current flow time
		if( $t0){
		    $elapsed=tv_interval ( $t0, $tc);
		} else {
		    $t0 = $tc;
		    $elapsed=0.0;
		    &init_stats($cur_id);
		}
		$us=$elapsed*10**6;
		
		$tjoined{$cur_id} = $tjoined if !$tjoined{$cur_id};
		# No stats for that SSRC yet?
		&pkt_stats($cur_id, $packet, $elapsed * 10**6, length($packet));
		if ($opt_dump){
		    print DUMP $packet;
		}
		$endstream = tv_interval($t0, $tc) >= $opt_last if $opt_last && !$endstream;
	    } else { # assume that we got an interrupt
#		die "Failed to receive packet: $!"; 
	    }
	}
#	&display_stats();
#        $mc->mcast_drop($address);

    } else {			# files
	my $f=$id;
	my $pcap; 
	my $tmp_file;

	my $pcap_file=$f;
	if ($f =~ /\.gz$|\.Z/){ # gzipped ?
	    $tmp_file="/tmp/pcap$$";
	    $pcap_file=$tmp_file;
	    system( "/bin/zcat $f > $tmp_file") ; # rc code ? || die "Could not unpack $f to $tmp_file : $!";
	}
	if ($opt_crude){
	    my $flow_key;
	    open CRUDE, "<$pcap_file" || die "Could not open $f : $err";
	    while(<CRUDE>){  # crude.sourceforge.net
		my $tx0;
		
		if (($fid, $seq, $src, $dst, $tx, $rx, $size)=
		    /ID=(\d+)\s+SEQ=(\d+)\s+SRC=([\d.:]+)\s+DST=([\d.:]+)\s+Tx=([\d.,]+)\s+Rx=([\d.,]+)\s+SIZE=(\d+)/){

		    $flow_key="$src->$dst";
		    $flow_pkts{$flow_key}++;
		    $tc=[int($rx), ($tx-int($rx))*10^6] ; # tval
		    if ($flow_pkts{$flow_key} <= 1) { # new flow
			push(@flows, $flow_key);  # 
			$flow_no++; 
			$file{$flow_key}=$f;
			$source{$flow_key}=$src;
			$rx0{$flow_key}=$rx;
			$tx0{$flow_key}=$tx;
			$tjoined{$flow_key}=$tc;
		    } 
		    
		    next if $opt_list;  # just count packets per flow
		    next if $opt_flow_key && $flow_key !~ $wanted_flow;
		    next if $opt_flow_no && $opt_flow_no !~ /\b$flow_no\b/;
		    
		    if ($flow_pkts{$flow_key} <= 1) { #
			&init_stats($flow_key);
			push (@myflows, $flow_key);
		    }

		    $seq_num{$flow_key}=$seq;
		    $ptx{$flow_key}=$tx{$flow_key}; # previous
		    $tx{$flow_key}=$tx;
		    $rx{$flow_key}=$rx;
		    $mtime{$flow_key}=($tx-$tx0{$flow_key})*10**6;
		    $us=($rx-$rx0{$flow_key})*10**6; # us
		    &pkt_stats($flow_key, 0, $us, $size);
		}
	    }
	    
	} else {
	    if ($pcap=Net::Pcap::open_offline($pcap_file, \$err)){ 
		#	my $select=IO::Select->new(Net::Pcap::fileno($pcap));
		until ( ! $uninterrupted || $endstream || ($opt_packets && $n_packets > $opt_packets) 
			|| ! (my $pkt=Net::Pcap::next($pcap, \%hdr) ) ) {
		    #$file_flag =1;
		    $offline =1;
		    eat_pcap($f, $pcap, $pkt);
		    $n_packets++;
		}
		printf "%s\n", $ttl_log if  $opt_ttl;

	    } else { # open failed 
		if ($format ne "full"){ # always put out data when format full
		    die "Could not open $f : $err";
		}
	    }
	}
	undef $t0;
	unlink $tmpfile if $tmp_file;
    } # of files
} # of eat_stream


sub eat_pcap_streams{ # open groups and listen on pcap
    my @streams=@_; 
    my $filter="( ";
    my ($address, $port, $multicast);
    my $mc = IO::Socket::Multicast->new(LocalPort=>$DEFAULT_PORT, ReuseAddr=>1) ||
	die "Could not connect to port $port : $!";
    foreach $id (@streams){ # listen to all groups
	if ($opt_ipv6){
	    $address = Net::IPv6Addr::ipv6_parse($id);
	    $port =  $opt_p;
	}
	else {
	    ($address, $port, $multicast) = prepare_ip($id);
	}
	$filter .= " or " if $filter =~ /\(\s+\(/;
	if ($multicast){
	    $mc->mcast_add($address) || die "could not join $address:$port : $!";
	    $filter .= "(dst $address and port $port)"; 	
	} else {
	    $filter .= "(src $address and port $port)"; 
	}
    }
    $tjoined{$f}=[gettimeofday];
    $filter .= " )";
#    $filter .= " and ip multicast";
#    $filter ="ip";
    my ($net, $netmask, $filter_t, $err, $dev);
#    printf $filter;
    $dev = Net::Pcap::lookupdev(\$err) || die "Could not lookupdev $dev : $err : $! .. Needs to run as sudo ..";
    my $r = Net::Pcap::lookupnet( $dev, \$net, \$netmask, \$err);
    die "Failed lookupnet for $dev : $r : $err : $!" if $r != 0;
    my $snap_length = 100;
    $snap_length = 1514 if $opt_mpeg;
    $pcap=Net::Pcap::open_live( $dev, $snap_length, 1, 0,\$err) || die "Could not open $dev  : $err";
    $r=Net::Pcap::compile($pcap, \$filter_t, $filter, 0, $netmask);
    die "Could not compile filter : $r : $filter : $!)" if $r != 0;
    $r=Net::Pcap::setfilter($pcap, $filter_t);
    die "Could not set pcap filter : $r : $filter : $! " if $r != 0 ;
    my $select=IO::Select->new( Net::Pcap::fileno($pcap));
    while ($select->can_read($timeout) && (my $pkt=Net::Pcap::next($pcap, \%hdr)) ) { # get all packets
	last if ! $uninterrupted || $endstream;
	eat_pcap("pcap", $pcap, $pkt);
	$n_packets++;
	last if  $opt_packets && $n_packets > $opt_packets;
    }
    printf "%s\n", $ttl_log if  $opt_ttl;

}

sub set_network_info{
    my ($f,$src_ip,$src_port,$dst_ip,$dst_port) = @_;
    $network_info{$f}{'src_ip'} = $src_ip;
    $network_info{$f}{'dst_ip'} = $dst_ip;
    $network_info{$f}{'src_port'} = $src_port;
    $network_info{$f}{'dst_port'} = $dst_port;	
}
sub eat_pcap { # process pcap packets	
    my ($f, $pcap, $pkt)=@_;
    my $ip=null;
    my $ip_payload;# udp or tcp packet.
    my $rtp = new Net::RTP::Packet(); # spare rtp obj
    my $proto,$len=null;
    if (Net::Pcap::datalink($pcap) == 1){ # ethernet
    	if ($opt_ipv6){
	    $ip = NetPacket::IPv6->decode(eth_strip($pkt));
	    $proto = $ip->{nxt};
	    $len = plen;
	    $hlen =  0;
	}
	else {
	    $ip = NetPacket::IP->decode(eth_strip($pkt));
	    $proto = $ip->{proto};
	    $len = len;
	    $hlen = ($ip->{hlen}*4);
	}
    } elsif (Net::Pcap::datalink($pcap) == 113) { # DLT_LINUX_SLL
	my ($head, $data)=unpack('a16a*', $pkt);  
	$ip = NetPacket::IP->decode($data);
    } else {
	die "Invalid link layer type : ".Net::Pcap::datalink($pcap);
    }
    
    if ($proto == IP_PROTO_TCP){
    	$opt_tcp = 1; 
	$ip_payload = NetPacket::TCP->decode($ip->{data});
    }
    else {
    	$ip_payload = NetPacket::UDP->decode($ip->{data});
    }
    my $dlen = $ip->{$len} - $hlen;
    my $flow_key;
    if ($opt_sum){
	$flow_key=$opt_sum;
    }
    elsif ($file_flag){
    	$flow_key = $f;
    }
    else {
	$flow_key=sprintf "%s:%d->%s:%d", 
        $ip->{src_ip}, $ip_payload->{src_port}, $ip->{dest_ip}, $ip_payload->{dest_port};
    }
    set_network_info($f,$ip->{src_ip}, $ip_payload->{src_port}, $ip->{dest_ip}, $ip_payload->{dest_port});
    $flow_pkts{$flow_key}++;
    #next if $ip->{proto} != IP_PROTO_UDP;
    if ($flow_pkts{$flow_key} <= 1) { # new flow
	push(@flows, $flow_key);  # 
	$flow_no++; 
	$file{$flow_key}=$f;
	$source{$flow_key} =sprintf "%s:%d", $ip->{src_ip}, $ip_payload->{src_port};
    }
    if ($opt_ttl){
	if ($pttl{$flow_key} && ($opt_log ||  ($pttl{$flow_key} !=  $ip->{ttl}))){ # change in ttl
	    printf "%s\n",  $ttl_log;
	    $nttl{$flow_key}=0;		
	}
	$pttl{$flow_key}= $ip->{ttl};
	$nttl{$flow_key}++;
	($sequence_number{$flow_key}, $tx_time_seconds{$flow_key}, $tx_time_useconds{$flow_key}, $flow_id{$flow_key})= unpack('NNNN', $ip_payload->{data});
	$ttl_log=sprintf "ID=%u SEQ=%u SRC=%s DST=%s Tx=%u.%06u  Rx=%u.%06u TTL=%u",  
	$flow_id{$flow_key}, $sequence_number{$flow_key}, 
	$ip->{src_ip}, $ip->{dest_ip}, $tx_time_seconds{$flow_key}, 
	$tx_time_useconds{$flow_key}, $hdr{tv_sec}, $hdr{tv_usec}, $pttl{$flow_key};
	$ttl_log.= sprintf(" No=%08d\n",  $nttl{$flow_key},) if !$opt_log;
    }
    next if $opt_list;  # just count packets per flow
    next if $opt_flow_key && $flow_key !~ $wanted_flow;
    next if $opt_flow_no && $opt_flow_no !~ /\b$flow_no\b/;

    if ($flow_pkts{$flow_key} <= 1) { #
	&init_stats($flow_key);
	push (@myflows, $flow_key);
    }
    $tc= [$hdr{tv_sec}, $hdr{tv_usec}];
    if (! defined($t0)) {
	$t0=$tc;
    }
    $tjoined{$f}=$tc if !$tjoined{$f};
    $tjoined{$flow_key} = $tjoined{$f} if !exists($tjoined{$flow_key});
    $us = tv_interval ( $t0, $tc) * 10**6 ; # us relative packet time 	  
# fixate on first flow to appear
#    if ( $first_port) {
#	next if $ip_payload->{dest_port} != $first_port;
#    } else {
#	$first_port =  $ip_payload->{dest_port};
#    }
    if ($opt_tcp){
	&tcp_packet_stats($flow_key,$ip,$ip_payload);
    }
    &pkt_stats($flow_key, $ip_payload->{data}, $us, $dlen);
}

#--------------------------------------------------------------------------------
sub init_stats {
    my $f=shift;
    $pinterval=0;
    $npkt{$f}=$sumbyte{$f}=$ssbyte{$f}=0;
    $first_sec=0;
    delete $pmtime{$f}; delete $pus{$f} ; delete $first_us{$f};  delete $last_us{$f}; 
    delete $prev_1s{$f}; delete $prev_100ms{$f};
    delete $per1s{$f}; delete $per100ms{$f};
    delete $sum1s{$f}; delete $sum100ms{$f};
    $n100ms{$f}=0; $sum100ms{$f}=0; $ss100bps{$f}=0; $akk100ms{$f}=0;
    $sumbyte{$f}=$sumps{$f}=0;			# sum byte per second
    $maxbps{$f}=0; $max100ms{$f}=0;
    $dup{$f}=$lost{$f}=$late{$f}=0;
    $maxbyte{$f}=0; $minbyte{$f}=2**31;
    $sumgap{$f}=$ssgap{$f}=0;
    $ngap{$f}=0; $maxgap{$f}=0, $mingap{$f}=2**63-5;
    $njitter{$f}=$sumjitter{$f}=$ssjitter{$f}=0;
    $maxjitter{$f}=-2**63-5; $minjitter{$f}=2**63-5; delete $est_jitter{$f}; delete $pest_jitter{$f}; 
    $first_port = 0;  $n_seq=-1;
    delete $tstart{$f}; delete $tend{$f}; 
    

#MPEG-TS initalizing and resetting (new period!)
    
    $discontinuities = 0;
    %last_ccs = ();
    %mpeg_stats = ();
    @cc = ();
# r_stats {filename/handle}
    @pcr_stats{$f} = ();
#P-stats {filename/handle}
    @pts_stats{$f,$TS} = ();
    @mpeg_stats{$f} = ();
#P stuff
    $late_video_packets = 0;
#MPEG-TS END
    %dist1 = ();
    %dist2 = ();
    $num_cc = 0;
}
sub format_numbers{
    my $value =  shift;
    if ($value =~ /^[a-zA-z][\D]*$/ || $value  !~ /^\d+\.\d+$/) {
	return sprintf("%1.0f",$value);
    }
    else {
	return sprintf("%0.5f",$value);
    }
}
sub xml_dump_rtmp_tcp {
    my $f = $_[0];  # *Not* shift. This gets called from a signal handler.
    return 0 if $xml_printed;
    my ($rtmp_jitter_stats,$rtmp_stats,$span)  = get_rtmp_data() if $opt_rtmp;
    my ($tcp_packet_stats) = (\%tcp_packet_stats) if $opt_tcp;
    my $xml_version="1.0";
    if ($period_num <= 1 ){
	$doc = XML::DOM::Document->new;
    	my $decl = $doc->createXMLDecl('1.0');
	$doc-> setXMLDecl($decl);
	$qstream = $doc->createElement('qstream');
	$qstream->setAttribute('version',$xml_version);
	$doc->appendChild($qstream);
    }
    ($address,$app,$play) = rtmp_prepare_url($f) if $rtmp;
    #($address, $port) = prepare_ip($f) if $opt_tcp && !$opt_rtmp;
    #$port  = 1935 unless $opt_port;
    #my $id=$address.":".$port;
    my $id = $f;
    my ($sec,$min,$hour,$mday,$mon,$year,$wday,$yday,$isdst) = localtime($tstart{$f}->[0]);
    my $time=sprintf "%02d:%02d:%02d", $hour, $min, $sec;
    my $date=sprintf "%4d-%02d-%02d", $year+1900, $mon+1, $mday;
    my $span=&tv_interval($tstart{$f}, $tend{$f}); # second
    &do_math($id,$span);
    my $flow = $doc->createElement('flow');
    my $flow_id = $doc->createElement('flow_id');
    my $rtmp_stats_xml  = $doc->createElement('rtmp_stats');
    my $dateXML = $doc->createElement('date');
    my $timeXML = $doc->createElement('time');
    my $spanXML = $doc->createElement('span'); 
    $qstream->appendChild($flow);
    $flow->appendChild($flow_id);
    $flow->appendChild($dateXML);
    $flow->appendChild($timeXML);
    $flow->appendChild($rtmp_stats_xml) if $opt_rtmp;
    $rtmp_stats_xml->appendChild($spanXML);    

    #source and destination ip. 
    $flow_id->addText($id);
    
    #datetime 
    $dateXML->addText($date);
    $timeXML->addText($time);
    $spanXML->addText($span);

    if ($opt_rtmp) {
	for my $k1 ( sort keys %$rtmp_stats ) {
	    my $element = $doc->createElement($k1);	
	    $rtmp_stats_xml->appendChild($element);
	    $element->addText(&format_numbers($rtmp_stats->{ $k1 }));	
	}
	foreach $k1 (sort keys %$rtmp_jitter_stats ){
	    if ($rtmp_jitter_stats->{ $k1 }{ 'type' } eq "Video" || $rtmp_jitter_stats->{ $k1 }{ 'type'} eq "Audio" ){
		my $chunk = $doc->createElement('chunk');
		$chunk->setAttribute('id', $k1);
		$rtmp_stats_xml->appendChild($chunk);
		for my $k2 (sort keys %{$rtmp_jitter_stats->{ $k1 }} ) {
		    my $element = $doc->createElement($k2);
		    $chunk->appendChild($element);
		    $value = &format_numbers($rtmp_jitter_stats->{ $k1 }{ $k2 }); 
		    $element->addText(&format_numbers($value));
		}   
	    }     
	}
    }
    #bandwidth	
    if ($opt_tcp){
    	$f = $id;
    	my $bw_stats = $doc->createElement('bandwidth_stats');
	my $avg_bw = $doc->createElement('bandwidth_avg');
    	my $max_bw = $doc->createElement('bandwidth_max_1s');
	my $max_bw_100ms = $doc->createElement('bandwidth_max_100ms');
    	my $sdv_bw = $doc->createElement('bandwidth_sdv');
    	$flow->appendChild($bw_stats);
    	$bw_stats->appendChild($avg_bw);
    	$bw_stats->appendChild($max_bw);
    	$bw_stats->appendChild($max_bw_100ms);
    	$bw_stats->appendChild($sdv_bw);
    	$avg_bw->addText(&format_numbers( avg( $bw_stats{$f}{'sumbyte'}*8, $span )));
	$max_bw->addText(&format_numbers($bw_stats{$f}{'maxbps'}));
	$max_bw_100ms->addText(&format_numbers($bw_stats{$f}{'max100ms'}));
	$sdv_bw->addText(&format_numbers(sdv($bw_stats{$f}{'n100ms'},
					     $bw_stats{$f}{'akk100ms'}, 
					     $bw_stats{$f}{'ss100bps'})));
    	my $child = $doc->createElement('tcp_packet_stats');
    	$flow->appendChild($child);
	for my $k2 (sort keys %{$tcp_packet_stats->{ $id }} ) {
	    my $grandchild = $doc->createElement($k2);
	    $value = &format_numbers($tcp_packet_stats->{$id}{$k2});
	    $grandchild->addText($value);	
	    $child->appendChild($grandchild);
	}
    } 
    #my $final_xml = $flow->toString();
    &init_stats($id);
}

sub xml_new {
    my ($pcr_stats,$pts_stats,$mpeg_stats) = (\%pcr_stats, \%pts_stats,\%mpeg_stats) if $opt_mpeg;
    my $doc = XML::DOM::Document->new;
    my $decl = $doc->createXMLDecl('1.0');
    my $flow = $doc->createElement('flow');
    my $src_dst = $doc->createElement('src_dst');
    my $dateXML = $doc->createElement('date');
    my $timeXML = $doc->createElement('time');
    my @elements_network_stats  = qw "src_dst date time network_stats setuptime gap";
    my @elements_pcr  = qw "pcr_avg_jitter pcr_sdv_jitter pcr_jitter_min pcr_jitter_max";
    my @elements_pts  = qw "dwell_sum dwell_num";
    my @elements_rtp = qw " ";
    my $mpeg_stats_parrent = $doc->createElement('mpeg_stats');
    my $pcr_stats_parrent = $doc->createElement('pcr_stats');
    my $transport_streams = $doc->createElement('transport_streams');
    my $pts_stats_parrent = $doc->createElement('pts_stats');

    #append static childs
    $doc->appendChild($flow);
    $flow->appendChild($pcr_stats_parrent);
    $flow->appendChild($transport_streams);
    $transport_streams->appendChild($pts_stats_parrent);
    $transport_streams->appendChild($mpeg_stats_parrent);
    
    #loop through hashes
    for my $k1 ( sort keys %mpeg_stats ) {
    	for my $k2 ( keys %{$mpeg_stats->{ $k1 }} ) {
	    for my $k3 (keys %{$mpeg_stats->{ $k1 } { $k2 }}){
		#$child = $doc->createElement($k3);
		#$child->addText(&lesbar($mpeg_stats->{ $k1 }{ $k2 }{ $k3 }));
		#$mpeg_stats_parrent->appendChild($child);		
	    }
    	}
    }
    
    for my $k1 ( sort keys %$pcr_stats ) {
    	for my $k2 ( keys %{$pcr_stats->{ $k1 }} ) {
	    foreach my $element (@elements_pcr){
		if ($element eq $k2){
		    $child = $doc->createElement($k2);
		    $child->addText(&lesbar($pcr_stats->{ $k1 }{ $k2 }));
		    $pcr_stats_parrent->appendChild($child);		
		}
	    }
      	}
    }
    
    for my $k1 ( sort keys %$pts_stats ) {
    	for my $k2 ( keys %{$pts_stats->{ $k1 }} ) {
	    for my $k3 (keys %{$pts_stats->{ $k1 } { $k2 }}){
		foreach my $element (@elements_pts){
		    if ($element eq $k3){
			$child = $doc->createElement($k3);
			$child->addText(&lesbar($pts_stats->{ $k1 }{ $k2 }{ $k3 }));
			$pcr_stats_parrent->appendChild($child);		
		    }	
		}
	    }
      	}
    }
}


sub xml_dump {
    my $f = $_[0];  # *Not* shift. This gets called from a signal handler.
    return 0 if $xml_printed;
    my ($network_info) = (\%network_info);
    my ($pcr_stats,$pts_stats) = (\%pcr_stats, \%pts_stats) if $opt_mpeg;
    my $xml_version="1.0";
    if ($period_num <= 1 ){
	$doc = XML::DOM::Document->new;
    	my $decl = $doc->createXMLDecl('1.0');
	$doc-> setXMLDecl($decl);
	$qstream = $doc->createElement('qstream');
	$qstream->setAttribute('version',$xml_version);
	$doc->appendChild($qstream);
    }
    my $flow = $doc->createElement('flow');
    my $flow_id = $doc->createElement('flow_id');
    my $flow_nameXML = $doc->createElement('flow_name');
    my $dateXML = $doc->createElement('date');
    my $timeXML = $doc->createElement('time');
    my $spanXML = $doc->createElement('span');
    my $pcr_jitter_parrent = $doc->createElement('pcr_jitter') if $opt_mpeg;
    my $network_info_parrent = $doc->createElement('network_info');
    $qstream->appendChild($flow);
    $flow->appendChild($flow_id);
    $flow->appendChild($flow_nameXML);
    $flow->appendChild($dateXML);
    $flow->appendChild($timeXML);
    $flow->appendChild($spanXML);
    $flow->appendChild($network_info_parrent);
    $flow->appendChild($pcr_jitter_parrent) if $opt_mpeg;
    my $media="file";
    $media = "host:port" if $opt_net;
    my $id=$f;
    $flow_name = $id;
    $id = $opt_id if $opt_id;
    $flow_name = $opt_name if $opt_name;
    next if $npkt{$f} < $opt_flow_min;
    my $source=$f; 
    $source=$source{$f} if  ! $opt_v && $source{$f} ne "";
    $source=$opt_id if $opt_id;
    my $jitter_stat= $opt_rtp || $opt_crude;

    my ($sec,$min,$hour,$mday,$mon,$year,$wday,$yday,$isdst) = localtime($tstart{$f}->[0]);
    my $time=sprintf "%02d:%02d:%02d", $hour, $min, $sec;
    if ($format eq "full"){
	$time.=sprintf ".%06d", $tstart{$f}->[1];
    } else {
	my $time=sprintf "%02d:%02d:%02d", $hour, $min, $sec;
    }
    my $date=sprintf "%4d-%02d-%02d", $year+1900, $mon+1, $mday;
    my $span=&tv_interval($tstart{$f}, $tend{$f}); # second
    &do_math($f,$span);
    &burst($f); # in case a second is interrupted
    #source and destination ip. 
    $flow_id->addText($id);
    $flow_nameXML->addText($flow_name);
    #datetime 
    $dateXML->addText($date);
    $timeXML->addText($time);
    $spanXML->addText($span);
    for my $k1 ( sort keys %network_info ) {
    	for my $k2 ( keys %{$network_info->{ $k1 }} ) {
	    $child = $doc->createElement($k2);
	    $child->addText($network_info->{ $k1 }{ $k2 });
	    $network_info_parrent->appendChild($child);
    	}
    }
    #network_stats
    my $network_stats = $doc->createElement('network_stats');
    my $setuptimeXML = $doc->createElement('setuptime');
    my $gap_avgXML = $doc->createElement('gap_avg');
    my $gap_sdvXML = $doc->createElement('gap_sdv');
    my $gap_minXML = $doc->createElement('gap_min');
    my $gap_maxXML = $doc->createElement('gap_max');
    $network_stats->appendChild($setuptimeXML);
    $network_stats->appendChild($gap_avgXML);
    $network_stats->appendChild($gap_sdvXML);
    $network_stats->appendChild($gap_minXML);
    $network_stats->appendChild($gap_maxXML); 
    $flow->appendChild($network_stats);
    #packets_stats
    my $packet_stats = $doc->createElement('packet_stats');
    my $packets_numbXML = $doc->createElement('packet_count');
    my $packets_sizeXML = $doc->createElement('packets_size');
    $flow->appendChild($packet_stats);
    $packet_stats->appendChild($packets_numbXML);
    
    #bandwidth
    my $bw_stats = $doc->createElement('bandwidth_stats');
    my $avg_bw = $doc->createElement('bandwidth_avg');
    my $max_bw = $doc->createElement('bandwidth_max');
    my $max_bw_100ms = $doc->createElement('bandwidth_max_100ms');
    my $sdv_bw = $doc->createElement('bandwidth_sdv');
    $bw_stats->appendChild($avg_bw);
    $bw_stats->appendChild($max_bw);
    $bw_stats->appendChild($max_bw_100ms);
    $bw_stats->appendChild($sdv_bw);
    $flow->appendChild($bw_stats);

    #network_stats
    $setuptimeXML->addText(&format_numbers($setuptime{$f}));
    $gap_avgXML->addText(&format_numbers(avg( $sumgap{$f}, $ngap{$f}) / 1000 ));
    $gap_sdvXML->addText(&format_numbers(sdv($ngap{$f}, $sumgap{$f},$ssgap{$f})/1000));
    $gap_minXML->addText(&format_numbers($mingap{$f}/1000));
    $gap_maxXML->addText(&format_numbers($mingap{$f}/1000));
    
    #Jitter / $opt_crue or $opt_rtp
    if ($jitter_stat){
	#create elements
	my $MOS_XML = $doc->createElement('MOS');
	my $rtp_timestamp = $doc->createElement('rtp_timestamp');
	my $rtp_dup = $doc->createElement('rtp_duplicated');
	my $rtp_late = $doc->createElement('rtp_pkt_late');
	my $rtp_lost = $doc->createElement('rtp_pkt_lost');
	my $rtp_gaps = $doc->createElement('rtp_pkt_gaps');
	my $rtp_avg = $doc->createElement('rtp_pkt_lost_avg');
	my $rtp_sdv = $doc->createElement('rtp_pkt_lost_sdv');
	my $rtp_jitter_numb = $doc->createElement('rtp_jitter_numb');
	my $rtp_jitter_avgXML = $doc->createElement('rtp_jitter_avg');
	my $rtp_jitter_sdvXML = $doc->createElement('rtp_jitter_sdv');
	my $rtp_jitter_minXML = $doc->createElement('rtp_jitter_min');
	my $rtp_jitter_maxXML = $doc->createElement('rtp_jitter_max');
	my $rtp_jitter_rfc = $doc->createElement('rtp_jitter_rfc');
	my $packets_dupXML = $doc->createElement('packet_duplicated');
    	my $packets_lateXML = $doc->createElement('packet_late');
    	my $packets_lostXML = $doc->createElement('packet_lost');
	my $packets_sizeXML = $doc->createElement('packet_size');
	
	#append childs
	$network_stats->appendChild($MOS_XML);
	$network_stats->appendChild($rtp_timestamp);
	$network_stats->appendChild($rtp_dup);
	$network_stats->appendChild($rtp_late);
	$network_stats->appendChild($rtp_lost);
	$network_stats->appendChild($rtp_gaps);
	$network_stats->appendChild($rtp_avg);
	$network_stats->appendChild($rtp_sdv);
	$network_stats->appendChild($rtp_jitter_numb);
	$network_stats->appendChild($rtp_jitter_avgXML);
	$network_stats->appendChild($rtp_jitter_sdvXML);
	$network_stats->appendChild($rtp_jitter_minXML);
	$network_stats->appendChild($rtp_jitter_maxXML);
	$network_stats->appendChild($rtp_jitter_rfc);
	
	$packet_stats->appendChild($packets_dupXML);
    	$packet_stats->appendChild($packets_lateXML);
    	$packet_stats->appendChild($packets_lostXML);
    	$packet_stats->appendChild($packets_sizeXML);
    	
    	$packets_dupXML->addText(&format_numbers($dup{$f}));
	$packets_lateXML->addText(&format_numbers($late{$f}));
	$packets_lostXML->addText(&format_numbers($nloss{$f}));
	my $n=$npkt{$f}; $n=1 if $npkt{$f} < 1; # divide by zero protection
	my $njitter = $njitter{$f}; $njitter =1 if $njitter{$f} < 1;
	$packets_sizeXML->addText(&format_numbers($sumbyte{$f}/$n));
	
	#adds values
	$MOS_XML->addText(&format_numbers(&mos_r($codec, $rtt,  avg( $sumgap{$f}, $ngap{$f}) / 1000,
						 avg($sumjitter{$f}, $njitter)/1000, 
						 100*$nloss{$f}/$n)));
	
	$rtp_timestamp->addText(&format_numbers($ntimeerr{$f}));
	$rtp_dup->addText(&format_numbers($dup{$f}));
	$rtp_late->addText(&format_numbers($late{$f}));
	$rtp_lost->addText(&format_numbers($lost{$f}));
	$rtp_gaps->addText(&format_numbers($nloss{$f}));
	$rtp_avg->addText(&format_numbers( avg( $sumloss{$f}, $nloss{$f}) ));
	$rtp_sdv->addText(&format_numbers( sdv($lost{$f}, $sumloss{$f},$ssumloss{$f})));
	$rtp_jitter_numb->addText(&format_numbers($njitter{$f}));			
	$rtp_jitter_avgXML->addText(&format_numbers($sumjitter{$f}/$njitter/1000));
	$rtp_jitter_sdvXML->addText(&format_numbers($sumjitter{$f}/$njitter/1000));
	$rtp_jitter_minXML->addText(&format_numbers($minjitter{$f}/1000));
	$rtp_jitter_maxXML->addText(&format_numbers($maxjitter{$f}/1000));
	$rtp_jitter_rfc->addText(&format_numbers($maxjitter{$f}/1000));
    }	
    #packet stats
    
    $packets_numbXML->addText(&format_numbers($npkt{$f}));
    #$packets_sizeXML->addText();

    #bandwidth	
    $avg_bw->addText(&format_numbers( avg( $sumbyte{$f}*8, $span )));
    $max_bw->addText(&format_numbers($maxbps{$f}));
    $max_bw_100ms->addText(&format_numbers($max100ms{$f}));
    $sdv_bw->addText(&format_numbers(sdv($n100ms{$f},
					 $akk100ms{$f}, 
					 $ss100bps{$f})));	
    if ($opt_mpeg) {			
	#PCR_stats
	my $pcr_jitter_avg_XML = $doc->createElement('pcr_jitter_avg');
	my $pcr_jitter_sdv_XML = $doc->createElement('pcr_jitter_sdv');
	my $pcr_jitter_min_XML = $doc->createElement('pcr_jitter_min');
	my $pcr_jitter_max_XML = $doc->createElement('pcr_jitter_max');
	$pcr_jitter_avg_XML->addText(&format_numbers( (10**3) * avg($pcr_stats{$f}{$pcr_pid{$f}}{'pcr_jitter_sum'}, $pcr_stats{$f}{$pcr_pid{$f}}{'num'})));
	$pcr_jitter_sdv_XML->addText(&format_numbers( (10**3) * sdv($pcr_stats{$f}{$pcr_pid{$f}}{'num'}, $pcr_stats{$f}{$pcr_pid{$f}}{'pcr_jitter_sum'},$pcr_stats{$f}{$pcr_pid{$f}}{'pcr_jitter_square'})));
	$pcr_jitter_min_XML->addText(&format_numbers( (10**3) * $pcr_stats{$f}{$pcr_pid{$f}}{'min_jit'}));
	$pcr_jitter_max_XML->addText(&format_numbers( (10**3) * $pcr_stats{$f}{$pcr_pid{$f}}{'max_jit'}));