USB develop

Development Platform:
Unix_Linux

triggerthread.cc：Code Content
							#include "../oconfig.h"
#include "../core/triggerthread.h"
#include <Qt/qapplication.h>
#include <unistd.h>
template<typename T>inline T MAX(T a,T b)
	{  return(a>b ? a : b);  }
DataBuffer::IData *TriggerThread::_DoDownSampling(Channel *chan,
	const DataBuffer *db)
{
	// FIXME: We may end up with one sample per DataBuffer if we have 
	//        a downsampling factor which is greater than the (standard) 
	//        buffer size. 
	
	size_t buflen=db->length();
	size_t input_samples = chan->avg_nsamp+buflen;
	size_t output_samples = input_samples/tp.downsampling_fact;
	DataBuffer::IData *dest=NULL;
	if(output_samples)
	{
		// Warn user. This is the result of the FIXME above!
		if(output_samples<10)
		{  fprintf(stderr,"output_samples very low (%u)...n",output_samples);  }
		
		dest = new(output_samples) DataBuffer::IData(output_samples);
	}
	// else: idata stays NULL. 
	
	int nsamp=chan->avg_nsamp;
	int maxsamp=(int)tp.downsampling_fact;
	assert(nsamp<maxsamp);
	size_t dest_i=0;
	const unsigned char *data=(const unsigned char*)db->buf();
	if(tp.downsampling_avg)
	{
		int accu=chan->avg_accu;
		for(size_t i=0; i<buflen; i++)
		{
			accu+=(int)(data[i]);
			if(++nsamp==maxsamp)
			{
				dest->buf[dest_i++]=(unsigned char)(accu/nsamp);
				accu=0;
				nsamp=0;
			}
		}
		chan->avg_accu=accu;
	}
	else
	{
		size_t i=maxsamp-nsamp-1;
		for(; i<buflen; i+=maxsamp)
		{  dest->buf[dest_i++]=data[i];  }
		//fprintf(stderr,"dest_i=%u, dest->len=%un",dest_i,dest->len);
		nsamp=i-buflen;
	}
	chan->avg_nsamp=nsamp;
	assert(!dest || dest_i==dest->len);
	
	return(dest);
}
TriggerThread::NBState TriggerThread::_GetNextBuffer(StreamBuffer *store_here)
{
	iterate:;
	
	// Get data from queue or wait for new data. 
	NBState rv=NB_OK;
	bool buf_ok=0;
	mutex.lock();
	for(;;)
	{
		// Do not change the order here!
		if(do_quit)
		{  rv=NB_Quit;  break;  }
		if(do_restart)
		{  rv=NB_Restart;  do_restart=0;  break;  }
		
		// If we have data in the queue, process it. 
		if(!bq.IsEmpty())
		{
			bq.PopTail(store_here);
			buf_ok=1;
			stream_tot_size-=store_here->dbA.length()+store_here->dbB.length();
			//char c='0'+(char)trig_state;  write(1,&c,1);
			if(trigger_now)
			{  rv=NB_TriggerNow;  trigger_now=0;  seen_gui_ack=0;  }
			if(seen_gui_ack)
			{  rv=NB_GUIAck;  seen_gui_ack=0;  }
			//else rv=NB_OK...
			break;
		}
		
		bool cond_rv=wcond.wait(&mutex,/*msec_timeout=*/ULONG_MAX);
		// cond_rv=true if woken up and false on timeout. 
	}
	mutex.unlock();
	
	// Do down-sampling if needed. 
	if(buf_ok && tp.downsampling_fact>1)
	{
		DataBuffer::IData *nbA=_DoDownSampling(&A,&store_here->dbA);
		DataBuffer::IData *nbB=_DoDownSampling(&B,&store_here->dbB);
		// If we do SOO much downsampling that the buffer is empty, 
		// get a new one. 
		assert(!nbA==!nbB);
		if(!nbA)
		{  goto iterate;  }
		// Otherwise, use down-sampled buffers. 
		store_here->dbA=DataBuffer(nbA);
		store_here->dbB=DataBuffer(nbB);
	}
	
	return(rv);
}
size_t TriggerThread::_TrigScanSignal(size_t bufi,const DataBuffer *db,
	int slope_mul)
{
	Channel *chan=NULL;
	switch(tp.trig_src)
	{
		case TS_None:  break;  // chan stays NULL
		case TS_ChanA:  chan=&A;  break;
		case TS_ChanB:  chan=&B;  break;
		default:  assert(0);
	}
	if(!chan)
	{  return(db->length());  }
	
	switch(chan->cmode)
	{
		case CM_Off:  return(db->length());
		case CM_Analog:
		{
			// Analog trigger. 
			size_t buflen=db->length();
			const unsigned char *data=(unsigned char*)db->buf();
			int lvl=tp.trig_level+127;
			// slope=+1 -> trigger if v>level
			// slope=-1 -> trigger if v<level
			if(tp.trig_slope*slope_mul>0)
			{
				for(size_t i=bufi; i<buflen; i++)
				{
					int v=(int)data[i];
					if(v>lvl)  return(i);
				}
			}
			else if(tp.trig_slope*slope_mul<0)
			{
				for(size_t i=bufi; i<buflen; i++)
				{
					int v=(int)data[i];
					if(v<lvl)  return(i);
				}
			}
			else assert(0);
		} break;
		case CM_Digital:
			fprintf(stderr,"Digital trigger not yet supported.n");
			return(db->length());
		default:  assert(0);
	}
	
	return(db->length());
}
void TriggerThread::_AppendSampleBuf(Channel *chan,const DataBuffer *db)
{
	if(chan->cmode==CM_Off)
	{
		chan->sb.clear();
		return;
	}
	
	chan->sb.append(*db);
}
void TriggerThread::_SamplingDone()
{
	// Okay, so we need to transfer the data to the GUI thread for 
	// display. 
	
	fprintf(stderr,"Delivering result (%u, %u netto samples (should be %u),"
		"st=%u,%u)... ",
		A.sb.NettoSamples(),B.sb.NettoSamples(),tp.nsamples,
		A.sb.SkipTail(),B.sb.SkipTail());
	if(data_receiver)
	{
		// We clear our buffers after the copy so that we don't get problems 
		// when continuing to sample and throwing in data when cut_head is set..
		SampleDataEvent *ev = new SampleDataEvent();
		ev->sbA.ShallowCopy(A.sb);  A.sb.clear();
		ev->sbB.ShallowCopy(B.sb);  B.sb.clear();
		
		QApplication::postEvent(data_receiver,ev);
	}
	fprintf(stderr,"donen");
}
void TriggerThread::_SendTrigStateEvent()
{
	if(data_receiver)
	{
		TriggerStateEvent *ev = new TriggerStateEvent();
		ev->trig_state=trig_state;
		
		QApplication::postEvent(data_receiver,ev);
	}
}
void TriggerThread::_ResetRestartTrigger()
{
	trig_state=TS_Arming;  _SendTrigStateEvent();
	A.avg_nsamp=0;  A.avg_accu=0;  A.sb.clear();
	B.avg_nsamp=0;  B.avg_accu=0;  B.sb.clear();
	fprintf(stderr,"Trigger: restartn");
}
void TriggerThread::_RunTriggerThread()
{
	
	// Trigger states: 
	//  TS_Arming:
	//     copy samples to arm sample buffer for triggering
	//  TS_WaitForTrigger0: 
	//     sample input and wait for signal to go below trigger level 
	//     (when triggering on rising edge; otherwise wait to go above 
	//     trigger level) 
	//  TS_WaitForTrigger1: 
	//     sample input and wait for signal to go above trigger level 
	//     (when triggering on rising edge; otherwise wait to go below 
	//     trigger level) and fire trigger in that case (-> TS_Triggered)
	//  TS_Triggered:
	//     simply copy data until samples are done
	//  TS_HoldOff
	//     discart data for holdoff time
	//  TS_WaitACK
	//     wait for ACK by GUI thread so that we don'r run too fast!
	
	// Current buffer (already dequeued). 
	StreamBuffer curr;
	// bufi: Index in current buffer. 
	//       If bufi==buffer length, get new buffer. 
	size_t bufi=0;
	// Buffer length. 
	size_t buflen=0;
	for(;;)
	{
		restart:;
		// Get next buffer or wait for it. 
		bool do_trig_now=0;
		bool got_gui_ack=0;
		if(bufi==buflen)  // ==curr.dbB.length() ALWAYS
		{
			NBState nbs=_GetNextBuffer(&curr);
			switch(nbs)
			{
				case NB_OK:       break;
				case NB_TriggerNow:  do_trig_now=1;  break;
				case NB_GUIAck:      got_gui_ack=1;  break;
				case NB_Quit:     return;
				case NB_Restart:
				{
					_CopyNewParam();
					_ResetRestartTrigger();
					goto restart;
				}
				default: assert(0); break;
			}
			bufi=0;
			buflen=curr.dbA.length();
			assert(buflen);
			assert(buflen==curr.dbB.length());
			
//write(1,"u",1);
			
			// For now, we put everything into the sample buffer (unless the 
			// channel is switched off) and trim it afterwards if needed. 
			_AppendSampleBuf(&A,&curr.dbA);
			_AppendSampleBuf(&B,&curr.dbB);
			
			// Make sure we don't collect too much data: 
			//fprintf(stderr,"TrimMin(%u+%u)n",tp.trig_h_pos,buflen);
			switch(trig_state)
			{
				case TS_Arming:           // fall 
				case TS_WaitForTrigger0:  //   through
				case TS_WaitForTrigger1:  //     here
					if(A.cmode!=CM_Off)
					{  A.sb.TrimMin(tp.trig_h_pos+buflen);  }
					if(B.cmode!=CM_Off)
					{  B.sb.TrimMin(tp.trig_h_pos+buflen);  }
					break;
				case TS_Triggered:
				case TS_HoldOff:
				case TS_WaitACK:
					if(A.cmode!=CM_Off)
					{  A.sb.TrimMin(tp.nsamples+buflen+A.sb.SkipTail()+A.sb.CutHead());  }
					if(B.cmode!=CM_Off)
					{  B.sb.TrimMin(tp.nsamples+buflen+B.sb.SkipTail()+B.sb.CutHead());  }
					break;
				default:  assert(0);
			}
				
//write(1,"v",1);
			
			// See if trigger is finally armed. 
			// This is the case when all enabled channels have enough samples 
			// and at least one channel IS enabled. 
			if(trig_state==TS_Arming)
			{
				if((A.cmode || B.cmode) && 
					(A.cmode==CM_Off || A.sb.NSamples()>=tp.trig_h_pos) && 
					(B.cmode==CM_Off || B.sb.NSamples()>=tp.trig_h_pos) )
				{
					// Make sure we consume all the samples up to the armed 
					// point. 
					// NOTE: The MAX should just clip off zero!
					size_t nsamp=MAX(A.sb.NSamples(),B.sb.NSamples());
					assert(!A.sb.NSamples() || !B.sb.NSamples() || 
						A.sb.NSamples()==B.sb.NSamples());
					size_t beyond = nsamp-tp.trig_h_pos;
					if(beyond<buflen)
					{  bufi=buflen-beyond;  }
					trig_state=TS_WaitForTrigger0;  _SendTrigStateEvent();
				}
				else
				{
					// Otherwise, go on arming...
					bufi=buflen;
				}
			}
		}
//{ char c='a'+(char)trig_state;  write(1,&c,1);  }
		
		if(trig_state==TS_WaitACK)
		{
			if(got_gui_ack || do_trig_now)
			{
				_ResetRestartTrigger();  // Will start with TS_Arming. 
			}
			else
			{
				// Discard. 
				bufi=buflen;
			}
		}
		
		// Depending on the trigger state, we do nothing or search for 
		// the trigger signal. 
		// If do_trig_now is set, trigger at bufi. 
		if(do_trig_now)
		{
			if(trig_state==TS_Triggered || trig_state==TS_Arming)
			{  do_trig_now=0;  }  // ignore
		}
		else if(trig_state==TS_WaitForTrigger0 || 
		   trig_state==TS_WaitForTrigger1 )
		{
			DataBuffer *trig_db=NULL;
			switch(tp.trig_src)
			{
				case TS_ChanA:  trig_db=&curr.dbA;  break;
				case TS_ChanB:  trig_db=&curr.dbB;  break;
				default:
					/* trig_db stays NULL esp. for TS_None. */
					bufi=buflen;
					break;
			}
			
			if(trig_db && trig_state==TS_WaitForTrigger0)
			{
				bufi=_TrigScanSignal(bufi,trig_db,-1);
				if(bufi<buflen)
				{  trig_state=TS_WaitForTrigger1;  }
			}
			/*no else*/if(trig_db && trig_state==TS_WaitForTrigger1)
			{
				// Scan to see if we can get triggered finally :)
				bufi=_TrigScanSignal(bufi,trig_db,+1);
				if(bufi<buflen)
				{  do_trig_now=1;  }
			}
		}
//write(1,"E",1);
		// See if we're triggered now. 
		if(do_trig_now)
		{
			assert(trig_state!=TS_Arming);
			
			// Since we want to trigger at bufi, trim the sample buffer 
			// in a way that exactly trig_h_pos-many samples are left of 
			// bufi, or more exactly, that the bufi is the trig_h_pos-th 
			// sample. 
			if(A.cmode!=CM_Off)
			{  A.sb.TrimTriggered(tp.trig_h_pos,bufi);  }
			if(B.cmode!=CM_Off)
			{  B.sb.TrimTriggered(tp.trig_h_pos,bufi);  }
			
			// Go into triggered state. Trigger at bufi. 
			trig_state=TS_Triggered;  _SendTrigStateEvent();
			fprintf(stderr,"Trigger: triggered (st=%u,%u)n",
				A.sb.SkipTail(),B.sb.SkipTail());
		}
		
//write(1,"F",1);
		if(trig_state==TS_Triggered)
		{
			// Just check when we're done. 
			ssize_t rvA=-1,rvB=-1;
			if(A.cmode!=CM_Off)
			{  rvA=A.sb.CutToLength(tp.nsamples);  assert(rvA!=-2);  }
			if(B.cmode!=CM_Off)
			{  rvB=B.sb.CutToLength(tp.nsamples);  assert(rvB!=-2);  }
			if(rvA>=0 || rvB>=0)
			{
				assert(A.cmode==CM_Off || rvA>=0);
				assert(B.cmode==CM_Off || rvB>=0);
				if(A.cmode!=CM_Off && B.cmode!=CM_Off) assert(rvA==rvB);
				
				if(A.cmode!=CM_Off)  bufi=rvA;
				else if(B.cmode!=CM_Off)  bufi=rvB;
				else assert(0);  // <-- bufi=buflen;
				
				// So, we're now done. 
				_SamplingDone();
				trig_state=TS_HoldOff;  _SendTrigStateEvent();
			}
			else
			{  bufi=buflen;  }
		}
		
//write(1,"G",1);
		if(trig_state==TS_HoldOff)
		{
			// FIXME: Should process holdoff. 
			bufi=buflen;
			
			// For single trigger, we stay in HoldOff until reset. 
			if(tp.trig_mode!=TM_Single)
			{
				// When holdoff done: 
				trig_state=TS_WaitACK;  _SendTrigStateEvent();
				_ResetRestartTrigger();  // Will start with TS_Arming. 
			}
			else
			{  bufi=buflen;  }
		}
		
//write(1,"x",1);
		
	}
}
void TriggerThread::run()
{
	fprintf(stderr,"Trigger thread runningn");
	
	USBIOThread::usb_thread->RegisterStreamHandler(this);
	
	thread_running.post();
	
	_RunTriggerThread();
	
	USBIOThread::usb_thread->UnregisterStreamHandler(this);
	
	fprintf(stderr,"Trigger thread exitingn");
}
void TriggerThread::_CopyNewParam()
{
	param_mutex.lock();
	
	tp=new_param;  // C++-assignment. 
	fprintf(stderr,"Trigger: copying new params. ns=%un",tp.nsamples);
	
	A.cmode=new_param.cmodeA;
	B.cmode=new_param.cmodeB;
	
	param_mutex.unlock();
}
void TriggerThread::QuitThread()
{
	mutex.lock();
	do_quit=1;
	mutex.unlock();
	wcond.wakeOne();
}
void TriggerThread::TriggerNow()
{
	mutex.lock();
	trigger_now=1;
	mutex.unlock();
	wcond.wakeOne();
}
void TriggerThread::TriggerReset()
{
	mutex.lock();
	do_restart=1;  // Well, this may not be prefect...
	mutex.unlock();
	wcond.wakeOne();
}
void TriggerThread::GUIDataAck()
{
	mutex.lock();
	seen_gui_ack=1;
	mutex.unlock();
	wcond.wakeOne();
}
void TriggerThread::_SignalTrigReStart()
{
	mutex.lock();
	do_restart=1;
	mutex.unlock();
	wcond.wakeOne();
}
void TriggerThread::SetTriggerHPosAndSamples(size_t h_pos,size_t nsamples)
{
	param_mutex.lock();
	new_param.trig_h_pos=h_pos;
	new_param.nsamples=nsamples;
	assert(h_pos<nsamples);
	param_mutex.unlock();
	
	_SignalTrigReStart();
}
void TriggerThread::SetTriggerLevelAndSource(int level,TriggerSource tsrc)
{
	param_mutex.lock();
	new_param.trig_level=level;
	new_param.trig_src=tsrc;
	param_mutex.unlock();
	
	_SignalTrigReStart();
}
void TriggerThread::SetTriggerModeAndSlope(TriggerMode tmode,int slope)
{
	param_mutex.lock();
	new_param.trig_mode=tmode;
	new_param.trig_slope=slope;
	param_mutex.unlock();
	
	_SignalTrigReStart();
}
void TriggerThread::SetChannelMode(int chan_num,ChannelMode cmode)
{
	//Channel *c=chan_num==1 ? &A : &B;
	param_mutex.lock();
	     if(chan_num==1) new_param.cmodeA=cmode;
	else if(chan_num==2) new_param.cmodeB=cmode;
	else assert(0);
	param_mutex.unlock();
	
	_SignalTrigReStart();
}
void TriggerThread::SetHorizDownsampling(int fact,bool average)
{
	param_mutex.lock();
	new_param.downsampling_fact=fact;
	new_param.downsampling_avg=average;
	param_mutex.unlock();
	
	_SignalTrigReStart();
}
void TriggerThread::FeedBuffer(const DataBuffer &A,const DataBuffer &B)
{
	// This function is executed in USBIOThread context and must be as fast 
	// as possible. We just enqueue the buffers; NO PROCESSING WHATSOEVER. 
	
	size_t tlen=A.length()+B.length();
	assert(A.length()==B.length());
	
	mutex.lock();
	if(stream_tot_size+tlen<stream_buf_max_size)
	{
		bq.PushHead(StreamBuffer(A,B));
		stream_tot_size+=tlen;
		
		//write(1,"+",1);
		
		// There is only ONE thread waiting by construction so we don't 
		// have to wake ALL. 
		wcond.wakeOne();
	}
	else
	{
		++stream_dropped_buffers;
		fprintf(stderr,"OOPS: Dropping buffers.n");
	}
	mutex.unlock();
}
TriggerThread::TriggerThread(QObject *_data_receiver) : 
	QThread(),
	USBIOThread::StreamHandler(),
	mutex(),
	wcond(),
	bq(32,16),
	A(),B(),
	tp(),
	new_param(),
	param_mutex()
{
	mutex.lock();
	data_receiver=_data_receiver;
	stream_tot_size=0;
	stream_dropped_buffers=0;
	stream_buf_max_size=16*1024*1024;
	do_quit=0;
	trigger_now=0;
	seen_gui_ack=0;
	do_restart=0;
	mutex.unlock();
	
	trig_state=TS_Arming;
}
TriggerThread::~TriggerThread()
{
	data_receiver=NULL;
}
//------------------------------------------------------------------------------
TriggerThread::Channel::Channel() : 
	sb()
{
	cmode=CM_Off;
	avg_nsamp=0;
	avg_accu=0;
}
TriggerThread::TrigParam::TrigParam()
{
	trig_h_pos=0;
	nsamples=1000;
	trig_level=0;
	trig_src=TS_None;
	trig_mode=TM_Norm;
	trig_slope=+1;
	downsampling_fact=1;
	downsampling_avg=0;
}
TriggerThread::TrigParamExt::TrigParamExt() : 
	TrigParam()
{
	cmodeA=CM_Off;
	cmodeB=CM_Off;
}

						