HeadCountAlgo/detector/detdef/source/DetStreamPipe.cpp

#include "DetStreamPipe.h"

#include "SysUtils.h"

#define ASYNC_CHECK_DELAY_MS 20
#define DET_WAIT_MSECOND 100

NAMESPACE_MAS_BEGIN

DetStreamPipe::DetStreamPipe(TZ_INT rscID,
							 TZ_INT detId, TZ_INT intype, const ExcepInfo &einfo)
	: IStreamPipe(rscID, detId, EMPT_DET, einfo),
	  m_iType(intype),
	  m_detLib(nullptr),
	  m_detector(nullptr) {}

DetStreamPipe::~DetStreamPipe()
{
	this->Dispose();
}

TZ_INT DetStreamPipe::Initialize(SPtr<DetectorLib> & detLib, const std::string & initParam)
{
	m_detLib = detLib;
	m_detLib->GetDetLibKey(m_detKey);

	m_detector = m_detLib->BuildDetector();
	if (!m_detector)
	{
		TZLogWarn("%s BuildDetector return nullptr!!!", m_detKey.c_str());
		return MEC_NULL_OBJ;
	}

	auto irst = m_detector->Initialize(initParam);
	if (irst) return irst;

	this->Start();

	return MEC_OK;
}

TZ_INT DetStreamPipe::Dispose()
{
	this->StopAndWait();

	m_detLib->DestroyDetector(m_detector);

	m_detector = nullptr;

	return MEC_OK;
}

std::tuple<TZ_BOOL, TZ_BOOL> DetStreamPipe::streamArrived(SPtr<StreamInfo> streamInfo)
{
	if (m_detSema.Count() > 3000)
	{
		TZLogInfo("sema is more than 3000!!!");
		return std::make_tuple(FALSE, FALSE);
	}

	m_lock.Lock();
	m_cacheInfo.push_back(streamInfo);
	m_lock.Unlock();

	m_detSema.Signal();

	return std::make_tuple(FALSE, FALSE);
}

TZ_INT DetStreamPipe::TurnOnGPU()
{
	return m_detector->TurnOnGPU();
}

TZ_INT DetStreamPipe::TurnOffGPU()
{
	return m_detector->TurnOffGPU();
}

TZ_INT DetStreamPipe::SetDetectCfg(const std::string param)
{
	return m_detector->SetDetectCfg(param);
}

void DetStreamPipe::Entry()
{
	TZ_INT irst;
	while (!this->IsStop())
	{
		if (!m_detSema.Wait(DET_WAIT_MSECOND)) continue;

		m_lock.Lock();
		if (!m_cacheInfo.size())
		{
			m_lock.Unlock();
			continue;
		}

		SPtr<StreamInfo> info = m_cacheInfo.front();
		m_cacheInfo.pop_front();

		m_lock.Unlock();

		irst = m_detector->DoDetect(info, this->detCallback, this);
		if (irst)
		{
			TZLogWarn("Pipe %llu DoDetect failed %d!!", this->GetHandle(), irst);
		}
		//TZLogInfo("after %s-------------", m_detKey.c_str());
	}
}

TZ_INT DetStreamPipe::detCallback(SPtr<StreamInfo> & streamInfo, void * ctx)
{
	DetStreamPipe * self = (DetStreamPipe *)ctx;

	self->DeliverStream(streamInfo);

	return MEC_OK;
}

NAMESPACE_MAS_END