366 lines
13 KiB
C++
366 lines
13 KiB
C++
/** @file
|
|
* @author Edouard DUPIN
|
|
* @copyright 2015, Edouard DUPIN, all right reserved
|
|
* @license APACHE v2.0 (see license file)
|
|
*/
|
|
|
|
#include "Node.h"
|
|
#include <river/debug.h>
|
|
|
|
#undef __class__
|
|
#define __class__ "io::Node"
|
|
|
|
|
|
river::io::Node::Node(const std::string& _name, const std11::shared_ptr<const ejson::Object>& _config) :
|
|
m_config(_config),
|
|
m_name(_name),
|
|
m_isInput(false) {
|
|
static uint32_t uid=0;
|
|
m_uid = uid++;
|
|
RIVER_INFO("-----------------------------------------------------------------");
|
|
RIVER_INFO("-- CREATE NODE --");
|
|
RIVER_INFO("-----------------------------------------------------------------");
|
|
drain::IOFormatInterface interfaceFormat;
|
|
drain::IOFormatInterface hardwareFormat;
|
|
/**
|
|
io:"input", # input, output or aec
|
|
frequency:48000, # frequency to open device
|
|
channel-map:[ # mapping of the harware device (to change map if needed)
|
|
"front-left", "front-right",
|
|
"read-left", "rear-right",
|
|
],
|
|
# format to open device (int8, int16, int16-on-ont32, int24, int32, float)
|
|
type:"int16",
|
|
# muxer/demuxer format type (int8-on-int16, int16-on-int32, int24-on-int32, int32-on-int64, float)
|
|
mux-demux-type:"int16_on_int32",
|
|
*/
|
|
std::string interfaceType = m_config->getStringValue("io");
|
|
if ( interfaceType == "input"
|
|
|| interfaceType == "PAinput"
|
|
|| interfaceType == "aec") {
|
|
m_isInput = true;
|
|
} else {
|
|
m_isInput = false;
|
|
}
|
|
|
|
int32_t frequency = m_config->getNumberValue("frequency", 1);
|
|
// Get audio format type:
|
|
std::string type = m_config->getStringValue("type", "int16");
|
|
enum audio::format formatType = audio::getFormatFromString(type);
|
|
// Get volume stage :
|
|
std::string volumeName = m_config->getStringValue("volume-name", "");
|
|
if (volumeName != "") {
|
|
RIVER_INFO("add node volume stage : '" << volumeName << "'");
|
|
// use global manager for volume ...
|
|
m_volume = river::io::Manager::getInstance()->getVolumeGroup(volumeName);
|
|
}
|
|
// Get map type :
|
|
std::vector<audio::channel> map;
|
|
const std11::shared_ptr<const ejson::Array> listChannelMap = m_config->getArray("channel-map");
|
|
if ( listChannelMap == nullptr
|
|
|| listChannelMap->size() == 0) {
|
|
// set default channel property:
|
|
map.push_back(audio::channel_frontLeft);
|
|
map.push_back(audio::channel_frontRight);
|
|
} else {
|
|
for (size_t iii=0; iii<listChannelMap->size(); ++iii) {
|
|
std::string value = listChannelMap->getStringValue(iii);
|
|
map.push_back(audio::getChannelFromString(value));
|
|
}
|
|
}
|
|
hardwareFormat.set(map, formatType, frequency);
|
|
|
|
|
|
std::string muxerDemuxerConfig = m_config->getStringValue("mux-demux-type", "int16-on-int32");
|
|
enum audio::format muxerFormatType = audio::getFormatFromString(muxerDemuxerConfig);
|
|
if (m_isInput == true) {
|
|
if (muxerFormatType != audio::format_int16) {
|
|
RIVER_CRITICAL("not supported demuxer type ... " << muxerFormatType << " for INPUT set in file:" << muxerDemuxerConfig);
|
|
}
|
|
} else {
|
|
if (muxerFormatType != audio::format_int16_on_int32) {
|
|
RIVER_CRITICAL("not supported demuxer type ... " << muxerFormatType << " for OUTPUT set in file:" << muxerDemuxerConfig);
|
|
}
|
|
}
|
|
// no map change and no frequency change ...
|
|
interfaceFormat.set(map, muxerFormatType, frequency);
|
|
// configure process interface
|
|
if (m_isInput == true) {
|
|
m_process.setInputConfig(hardwareFormat);
|
|
m_process.setOutputConfig(interfaceFormat);
|
|
} else {
|
|
m_process.setOutputConfig(hardwareFormat);
|
|
m_process.setInputConfig(interfaceFormat);
|
|
}
|
|
//m_process.updateInterAlgo();
|
|
}
|
|
|
|
river::io::Node::~Node() {
|
|
RIVER_INFO("-----------------------------------------------------------------");
|
|
RIVER_INFO("-- DESTROY NODE --");
|
|
RIVER_INFO("-----------------------------------------------------------------");
|
|
};
|
|
|
|
size_t river::io::Node::getNumberOfInterface(enum river::modeInterface _interfaceType) {
|
|
size_t out = 0;
|
|
for (size_t iii=0; iii<m_list.size(); ++iii) {
|
|
if (m_list[iii] == nullptr) {
|
|
continue;
|
|
}
|
|
if (m_list[iii]->getMode() == _interfaceType) {
|
|
out++;
|
|
}
|
|
}
|
|
return out;
|
|
}
|
|
size_t river::io::Node::getNumberOfInterfaceAvaillable(enum river::modeInterface _interfaceType) {
|
|
size_t out = 0;
|
|
for (size_t iii=0; iii<m_listAvaillable.size(); ++iii) {
|
|
std11::shared_ptr<river::Interface> element = m_listAvaillable[iii].lock();
|
|
if (element == nullptr) {
|
|
continue;
|
|
}
|
|
if (element->getMode() == _interfaceType) {
|
|
out++;
|
|
}
|
|
}
|
|
return out;
|
|
}
|
|
|
|
void river::io::Node::registerAsRemote(const std11::shared_ptr<river::Interface>& _interface) {
|
|
std::vector<std11::weak_ptr<river::Interface> >::iterator it = m_listAvaillable.begin();
|
|
while (it != m_listAvaillable.end()) {
|
|
if (it->expired() == true) {
|
|
it = m_listAvaillable.erase(it);
|
|
}
|
|
++it;
|
|
}
|
|
m_listAvaillable.push_back(_interface);
|
|
}
|
|
|
|
void river::io::Node::interfaceAdd(const std11::shared_ptr<river::Interface>& _interface) {
|
|
{
|
|
std11::unique_lock<std11::mutex> lock(m_mutex);
|
|
for (size_t iii=0; iii<m_list.size(); ++iii) {
|
|
if (_interface == m_list[iii]) {
|
|
return;
|
|
}
|
|
}
|
|
RIVER_INFO("ADD interface for stream : '" << m_name << "' mode=" << (m_isInput?"input":"output") );
|
|
m_list.push_back(_interface);
|
|
}
|
|
if (m_list.size() == 1) {
|
|
startInGroup();
|
|
}
|
|
}
|
|
|
|
void river::io::Node::interfaceRemove(const std11::shared_ptr<river::Interface>& _interface) {
|
|
{
|
|
std11::unique_lock<std11::mutex> lock(m_mutex);
|
|
for (size_t iii=0; iii< m_list.size(); ++iii) {
|
|
if (_interface == m_list[iii]) {
|
|
m_list.erase(m_list.begin()+iii);
|
|
RIVER_INFO("RM interface for stream : '" << m_name << "' mode=" << (m_isInput?"input":"output") );
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
if (m_list.size() == 0) {
|
|
stopInGroup();
|
|
}
|
|
return;
|
|
}
|
|
|
|
|
|
void river::io::Node::volumeChange() {
|
|
for (size_t iii=0; iii< m_listAvaillable.size(); ++iii) {
|
|
std11::shared_ptr<river::Interface> node = m_listAvaillable[iii].lock();
|
|
if (node != nullptr) {
|
|
node->systemVolumeChange();
|
|
}
|
|
}
|
|
}
|
|
|
|
int32_t river::io::Node::newInput(const void* _inputBuffer,
|
|
uint32_t _nbChunk,
|
|
const std11::chrono::system_clock::time_point& _time) {
|
|
if (_inputBuffer == nullptr) {
|
|
return -1;
|
|
}
|
|
const int16_t* inputBuffer = static_cast<const int16_t *>(_inputBuffer);
|
|
for (size_t iii=0; iii< m_list.size(); ++iii) {
|
|
if (m_list[iii] == nullptr) {
|
|
continue;
|
|
}
|
|
if (m_list[iii]->getMode() != river::modeInterface_input) {
|
|
continue;
|
|
}
|
|
RIVER_VERBOSE(" IO name="<< m_list[iii]->getName());
|
|
m_list[iii]->systemNewInputData(_time, inputBuffer, _nbChunk);
|
|
}
|
|
RIVER_VERBOSE("data Input size request :" << _nbChunk << " [ END ]");
|
|
return 0;
|
|
}
|
|
|
|
int32_t river::io::Node::newOutput(void* _outputBuffer,
|
|
uint32_t _nbChunk,
|
|
const std11::chrono::system_clock::time_point& _time) {
|
|
if (_outputBuffer == nullptr) {
|
|
return -1;
|
|
}
|
|
std::vector<int32_t> output;
|
|
RIVER_VERBOSE("resize=" << _nbChunk*m_process.getInputConfig().getMap().size());
|
|
output.resize(_nbChunk*m_process.getInputConfig().getMap().size(), 0);
|
|
const int32_t* outputTmp = nullptr;
|
|
std::vector<uint8_t> outputTmp2;
|
|
RIVER_VERBOSE("resize=" << sizeof(int32_t)*m_process.getInputConfig().getMap().size()*_nbChunk);
|
|
outputTmp2.resize(sizeof(int32_t)*m_process.getInputConfig().getMap().size()*_nbChunk, 0);
|
|
for (size_t iii=0; iii< m_list.size(); ++iii) {
|
|
if (m_list[iii] == nullptr) {
|
|
continue;
|
|
}
|
|
if (m_list[iii]->getMode() != river::modeInterface_output) {
|
|
continue;
|
|
}
|
|
RIVER_VERBOSE(" IO name="<< m_list[iii]->getName());
|
|
// clear datas ...
|
|
memset(&outputTmp2[0], 0, sizeof(int32_t)*m_process.getInputConfig().getMap().size()*_nbChunk);
|
|
RIVER_VERBOSE(" request Data="<< _nbChunk << " time=" << _time);
|
|
m_list[iii]->systemNeedOutputData(_time, &outputTmp2[0], _nbChunk, sizeof(int32_t)*m_process.getInputConfig().getMap().size());
|
|
RIVER_VERBOSE(" Mix it ...");
|
|
outputTmp = reinterpret_cast<const int32_t*>(&outputTmp2[0]);
|
|
// Add data to the output tmp buffer :
|
|
for (size_t kkk=0; kkk<output.size(); ++kkk) {
|
|
output[kkk] += outputTmp[kkk];
|
|
}
|
|
}
|
|
RIVER_VERBOSE(" End stack process data ...");
|
|
m_process.processIn(&outputTmp2[0], _nbChunk, _outputBuffer, _nbChunk);
|
|
RIVER_VERBOSE(" Feedback :");
|
|
for (size_t iii=0; iii< m_list.size(); ++iii) {
|
|
if (m_list[iii] == nullptr) {
|
|
continue;
|
|
}
|
|
if (m_list[iii]->getMode() != river::modeInterface_feedback) {
|
|
continue;
|
|
}
|
|
RIVER_VERBOSE(" IO name="<< m_list[iii]->getName() << " (feedback) time=" << _time);
|
|
m_list[iii]->systemNewInputData(_time, _outputBuffer, _nbChunk);
|
|
}
|
|
RIVER_VERBOSE("data Output size request :" << _nbChunk << " [ END ]");
|
|
return 0;
|
|
}
|
|
|
|
static void link(etk::FSNode& _node, const std::string& _first, const std::string& _op, const std::string& _second) {
|
|
if (_op == "->") {
|
|
_node << " " << _first << " -> " << _second << ";\n";
|
|
} else if (_op == "<-") {
|
|
_node << " " << _first << " -> " <<_second<< " [color=transparent];\n";
|
|
_node << " " << _second << " -> " << _first << " [constraint=false];\n";
|
|
}
|
|
}
|
|
|
|
|
|
void river::io::Node::generateDot(etk::FSNode& _node) {
|
|
_node << "subgraph clusterNode_" << m_uid << " {\n";
|
|
_node << " color=blue;\n";
|
|
_node << " label=\"[" << m_uid << "] IO::Node : " << m_name << "\";\n";
|
|
if (m_isInput == true) {
|
|
_node << " node [shape=rarrow];\n";
|
|
_node << " NODE_" << m_uid << "_HW_interface [ label=\"HW interface\n interface=ALSA\n stream=MICROPHONE\n type=input\" ];\n";
|
|
_node << " subgraph clusterNode_" << m_uid << "_process {\n";
|
|
_node << " label=\"Drain::Process\";\n";
|
|
_node << " node [shape=ellipse];\n";
|
|
_node << " node_ALGO_" << m_uid << "_in [ label=\"format=xxx\n freq=yyy\n channelMap={left,right}\" ];\n";
|
|
_node << " node_ALGO_" << m_uid << "_out [ label=\"format=xxx\n freq=yyy\n channelMap={left,right}\" ];\n";
|
|
|
|
_node << " }\n";
|
|
_node << " node [shape=square];\n";
|
|
_node << " NODE_" << m_uid << "_demuxer [ label=\"DEMUXER\n format=xxx\" ];\n";
|
|
// Link all nodes :
|
|
_node << " NODE_" << m_uid << "_HW_interface -> node_ALGO_" << m_uid << "_in [arrowhead=\"open\"];\n";
|
|
_node << " node_ALGO_" << m_uid << "_in -> node_ALGO_" << m_uid << "_out [arrowhead=\"open\"];\n";
|
|
_node << " node_ALGO_" << m_uid << "_out -> NODE_" << m_uid << "_demuxer [arrowhead=\"open\"];\n";
|
|
} else {
|
|
size_t nbOutput = getNumberOfInterfaceAvaillable(river::modeInterface_output);
|
|
size_t nbfeedback = getNumberOfInterfaceAvaillable(river::modeInterface_feedback);
|
|
_node << " node [shape=larrow];\n";
|
|
_node << " NODE_" << m_uid << "_HW_interface [ label=\"HW interface\n interface=ALSA\n stream=SPEAKER\n type=output\" ];\n";
|
|
if (nbOutput>0) {
|
|
_node << " subgraph clusterNode_" << m_uid << "_process {\n";
|
|
_node << " label=\"Drain::Process\";\n";
|
|
_node << " node [shape=ellipse];\n";
|
|
_node << " node_ALGO_" << m_uid << "_out [ label=\"format=xxx\n freq=yyy\n channelMap={left,right}\" ];\n";
|
|
_node << " node_ALGO_" << m_uid << "_in [ label=\"format=xxx\n freq=yyy\n channelMap={left,right}\" ];\n";
|
|
_node << " }\n";
|
|
}
|
|
_node << " node [shape=square];\n";
|
|
if (nbOutput>0) {
|
|
_node << " NODE_" << m_uid << "_muxer [ label=\"MUXER\n format=xxx\" ];\n";
|
|
}
|
|
if (nbfeedback>0) {
|
|
_node << " NODE_" << m_uid << "_demuxer [ label=\"DEMUXER\n format=xxx\" ];\n";
|
|
}
|
|
// Link all nodes :
|
|
if (nbOutput>0) {
|
|
link(_node, "NODE_" + etk::to_string(m_uid) + "_HW_interface", "<-", "node_ALGO_" + etk::to_string(m_uid) + "_out");
|
|
link(_node, "node_ALGO_" + etk::to_string(m_uid) + "_out", "<-", "node_ALGO_" + etk::to_string(m_uid) + "_in");
|
|
link(_node, "node_ALGO_" + etk::to_string(m_uid) + "_in", "<-", "NODE_" + etk::to_string(m_uid) + "_muxer");
|
|
}
|
|
if (nbfeedback>0) {
|
|
_node << " NODE_" << m_uid << "_HW_interface -> NODE_" << m_uid << "_demuxer [arrowhead=\"open\"];\n";
|
|
}
|
|
if ( nbOutput>0
|
|
&& nbfeedback>0) {
|
|
_node << " { rank=same; NODE_" << m_uid << "_demuxer; NODE_" << m_uid << "_muxer }\n";
|
|
}
|
|
|
|
}
|
|
_node << "}\n";
|
|
|
|
for (size_t iii=0; iii< m_listAvaillable.size(); ++iii) {
|
|
if (m_listAvaillable[iii].expired() == true) {
|
|
continue;
|
|
}
|
|
std11::shared_ptr<river::Interface> element = m_listAvaillable[iii].lock();
|
|
if (element == nullptr) {
|
|
continue;
|
|
}
|
|
bool isLink = false;
|
|
for (size_t jjj=0; jjj<m_list.size(); ++jjj) {
|
|
if (element == m_list[jjj]) {
|
|
isLink = true;
|
|
}
|
|
}
|
|
if (element != nullptr) {
|
|
if (element->getMode() == modeInterface_input) {
|
|
element->generateDot(_node, "NODE_" + etk::to_string(m_uid) + "_demuxer", isLink);
|
|
} else if (element->getMode() == modeInterface_output) {
|
|
element->generateDot(_node, "NODE_" + etk::to_string(m_uid) + "_muxer", isLink);
|
|
} else if (element->getMode() == modeInterface_feedback) {
|
|
element->generateDot(_node, "NODE_" + etk::to_string(m_uid) + "_demuxer", isLink);
|
|
} else {
|
|
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
void river::io::Node::startInGroup() {
|
|
std11::shared_ptr<river::io::Group> group = m_group.lock();
|
|
if (group != nullptr) {
|
|
group->start();
|
|
} else {
|
|
start();
|
|
}
|
|
}
|
|
|
|
void river::io::Node::stopInGroup() {
|
|
std11::shared_ptr<river::io::Group> group = m_group.lock();
|
|
if (group != nullptr) {
|
|
group->stop();
|
|
} else {
|
|
stop();
|
|
}
|
|
} |