ロボット制御やIoTデバイスのデータ収集のために，最も楽をしてOSC over ZeroMQを実装する方法

ネットワーク越しに小さいデータのやり取りをして，スマホからロボットに命令を送ったり，IoTデバイスのセンサーの値を遠隔で見たい，ということが最近特に増えてきました．
ローカルネットワーク内に繋がるデバイスでしたら，特に何も考えることなく，UDP越しのOSC(Open Sound Control)を使うのが簡単で良いと思うのですが，別のグローバルなネットワークとデータをやり取りしたいときには，UDPではデータロスが不安な場合があります．TCP越しにOSCを使えばいいのですが，何となく不安が残るのと，1対多でのやり取りを考えるときに面倒くさい、という問題があります．

そこで，以前紹介した
cvl-robot.hateblo.jp
の中のメッセージをOSCフォーマットにしてしまうことで，ZeroMQ越しのOSCを使えるようにしたいと思います．
（元の提案の実装を知らないので、OSC over ZeroMQっぽいもの、の方が正確かな。）

1. OpenFrameworksにofxZmq Addonをインストール

以前の記事を参考にしてください．
現在の最新版は，ZMQ-4.1.4です．

一か所追加で修正が必要です。
libs/src/Windows.hppの中に次の一行を加えてください．if_nametoindex関数を呼べるようにするためのヘッダーです．

#include <netioapi.h>

2. OscReceiveElements.hの中のReceiveMessageクラスにコンストラクタを一つ追加

OscReceiveElements.h

class ReceivedMessage{
    void Init( const char *bundle, osc_bundle_element_size_t size );
public:
    ReceivedMessage(const char* data, size_t size) : addressPattern_(data) { Init(data, size); } // added
    explicit ReceivedMessage( const ReceivedPacket& packet );
    explicit ReceivedMessage( const ReceivedBundleElement& bundleElement );
…

3. ofxOscSenderにserializemessage()関数を勝手に作って追加

ofxOscSender.h

namespace osc{
class ofxOscSender
{
public:
        ...

	void serializeMessage(std::string *oscm, ofxOscMessage& message, bool wrapInBundle = true); // added

        ...
};

ofxOscSender.cppに追加

void ofxOscSender::serializeMessage(string *oscm, ofxOscMessage& message, bool wrapInBundle)
{
	//setting this much larger as it gets trimmed down to the size its using before being sent.
	//TODO: much better if we could make this dynamic? Maybe have ofxOscMessage return its size?
	static const int OUTPUT_BUFFER_SIZE = 327680;
	char buffer[OUTPUT_BUFFER_SIZE];
	osc::OutboundPacketStream p(buffer, OUTPUT_BUFFER_SIZE);

	// serialise the message
	if (wrapInBundle) p << osc::BeginBundleImmediate;
	appendMessage(message, p);
	if (wrapInBundle) p << osc::EndBundle;
	
	oscm->append(p.Data(), p.Size());
}

4. ofxOscReceiverにDataToMessage()関数を勝手に作って追加

ofxOscReceiver.h

class ofxOscReceiver : public osc::OscPacketListener
{
public:
	...

	void DataToMessage(ofxOscMessage* ofMessage, const char *bundle, osc::osc_bundle_element_size_t size); // added

        ...
};

ofxOscReceiver.cppに追加

void ofxOscReceiver::DataToMessage(ofxOscMessage* ofMessage, const char *bundle, osc::osc_bundle_element_size_t size)
{
	osc::ReceivedMessage *m = new osc::ReceivedMessage(bundle, size);
	
	// convert the message to an ofxOscMessage

	// set the address
	ofMessage->setAddress(m->AddressPattern());

	// transfer the arguments
	for (osc::ReceivedMessage::const_iterator arg = m->ArgumentsBegin();
	arg != m->ArgumentsEnd();
		++arg)
	{
		if (arg->IsInt32())
			ofMessage->addIntArg(arg->AsInt32Unchecked());
		else if (arg->IsInt64())
			ofMessage->addInt64Arg(arg->AsInt64Unchecked());
		else if (arg->IsFloat())
			ofMessage->addFloatArg(arg->AsFloatUnchecked());
		else if (arg->IsDouble())
			ofMessage->addDoubleArg(arg->AsDoubleUnchecked());
		else if (arg->IsString())
			ofMessage->addStringArg(arg->AsStringUnchecked());
		else if (arg->IsSymbol())
			ofMessage->addSymbolArg(arg->AsSymbolUnchecked());
		else if (arg->IsChar())
			ofMessage->addCharArg(arg->AsCharUnchecked());
		else if (arg->IsMidiMessage())
			ofMessage->addMidiMessageArg(arg->AsMidiMessageUnchecked());
		else if (arg->IsBool())
			ofMessage->addBoolArg(arg->AsBoolUnchecked());
		else if (arg->IsInfinitum())
			ofMessage->addTriggerArg();
		else if (arg->IsTimeTag())
			ofMessage->addTimetagArg(arg->AsTimeTagUnchecked());
		else if (arg->IsRgbaColor())
			ofMessage->addRgbaColorArg(arg->AsRgbaColorUnchecked());
		else if (arg->IsBlob()) {
			const char * dataPtr;
			osc::osc_bundle_element_size_t len = 0;
			arg->AsBlobUnchecked((const void*&)dataPtr, len);
			ofBuffer buffer(dataPtr, len);
			ofMessage->addBlobArg(buffer);
		}
		else
		{
			ofLogError("ofxOscReceiver") << "ProcessMessage: argument in message " << m->AddressPattern() << " is not an int, float, or string";
		}
	}

	delete m;
}

5.テストプログラム

テスト用のプログラムはこんな感じ。
testApp.cpp

#include "testApp.h"

#include "ofxZmq.h"
#include "ofxOsc\src\ofxOsc.h"

ofxZmqSubscriber *subscriber;
ofxZmqPublisher *publisher;

int ZmqSender(ofxOscMessage &m)
{
	ofxOscSender sender;
	string s;
	sender.serializeMessage(&s, m, false);

	if (!publisher->send(s))
	{
		cerr << "send failed" << endl;
		return 0;
	}
	return s.size();
}

int ZmqUpdater(ofxOscMessage &m)
{
	ofBuffer data;
	subscriber->getNextMessage(data);

	ofxOscReceiver receiver;
	receiver.DataToMessage(&m, data.getData(), data.size());

	return data.size();
}

void dumpOSC(ofxOscMessage m) {
	string msg_string;
	msg_string = m.getAddress();
	for (int i = 0; i < m.getNumArgs(); i++) {
		msg_string += " ";
		if (m.getArgType(i) == OFXOSC_TYPE_INT32)
			msg_string += ofToString(m.getArgAsInt32(i));
		else if (m.getArgType(i) == OFXOSC_TYPE_FLOAT)
			msg_string += ofToString(m.getArgAsFloat(i));
		else if (m.getArgType(i) == OFXOSC_TYPE_STRING)
			msg_string += m.getArgAsString(i);
	}
	cout << msg_string << endl;
}

//--------------------------------------------------------------
void testApp::setup()
{
	subscriber = new ofxZmqSubscriber();
	publisher = new ofxZmqPublisher();

	// start server
	publisher->bind("tcp://*:9999");
	
	// start client
	subscriber->connect("tcp://localhost:9999");
}

//--------------------------------------------------------------
void testApp::update()
{
	while (subscriber->hasWaitingMessage())
	{
		ofxOscMessage m;
		ZmqUpdater(m);
		cout << "Received: ";
		dumpOSC(m);
	}
}

//--------------------------------------------------------------
void testApp::draw()
{
}

//--------------------------------------------------------------
void testApp::keyPressed(int key)
{
	ofxOscMessage m;
	m.setAddress("/key/pressed");
	m.addIntArg(key);

	ZmqSender(m);
}

//--------------------------------------------------------------
void testApp::keyReleased(int key)
{

}

//--------------------------------------------------------------
void testApp::mouseMoved(int x, int y)
{

}

//--------------------------------------------------------------
void testApp::mouseDragged(int x, int y, int button)
{

}

//--------------------------------------------------------------
void testApp::mousePressed(int x, int y, int button)
{
	ofxOscMessage m;               //OSCメッセージの準備
	m.setAddress("/mouse/button"); //OSCアドレスの指定
	m.addStringArg("pressed");     //OSC引数として、マウス状態"down"を送信
	m.addIntArg(x);                //OSC引数として、現在のマウスの座標(x, y)を送信
	m.addIntArg(y);
	m.addIntArg(button);

	ZmqSender(m);
}

//--------------------------------------------------------------
void testApp::mouseReleased(int x, int y, int button)
{

}

//--------------------------------------------------------------
void testApp::windowResized(int w, int h)
{

}

//--------------------------------------------------------------
void testApp::gotMessage(ofMessage msg)
{

}

//--------------------------------------------------------------
void testApp::dragEvent(ofDragInfo dragInfo)
{

}

マウスやキーボードをカチャカチャやると、こんな感じでOSCメッセージがZeroMQにラップされて送られてきます。
この場合は、ローカルでやっていますので自分から自分へ送っていますが，別のデバイスに送りたいときはclientのIPアドレスを変更してみてください．
f:id:cvl-robot:20160303174445p:plain