想把odas这个项目弄成分布式的，用一个主机同时监控多个树莓派进行波达方向定位。源码备份在doa-project，防丢用。

树莓派#

新的 raspbian buster 自带了python 3.7，安装pipenv即可：

1
proxychains4 pip install pipenv

环境准备#

初始化虚拟环境#

为odas项目新建文件夹odas，初始化pipenv环境：

1
mkdir odas
2
cd odas
3
proxychains4 pipenv --python 3.7
4
# 或 pipenv --python /usr/bin/python3

安装声卡驱动#

1
mkdir 3rd-lib
2
cd 3rd-lib
3

4
# 安装声卡驱动
5
git clone https://github.com/respeaker/seeed-voicecard.git
6
cd seeed-voicecard
7
sudo proxychains4 ./install.sh
8
sudo reboot -h now
9

10
# 测试声卡
11
arecord -L
12
aplay -L

编译`odas`项目#

1
# 编译odas
2
cd 3rd-lib/
3
proxychains4 git clone https://github.com/introlab/odas.git
4
mkdir odas/build
5
cd odas/build
6
cmake ..
7
make
8

9
# 做软链接方便使用
10
ln -s ~/odas/3rd-lib/odas/bin/odaslive ~/.local/bin/odaslive
11

12
# 测试odas
13
odaslive -c ~/respeaker-6mic-odas.cfg

安装`pyzmq`#

新的buster自带了libzmq 4.3.1，安装pyzmq即可：

1
proxychains4 pip install pyzmq

测试：

1
>>> import zmq
2
>>> zmq.zmq_version()
3
'4.3.1'

配置防火墙#

检查端口是否开启可以使用nc -zvw3 192.168.1.108 5557（z用于扫描端口的零I/O模式，v显示详情，w3三秒后超时）。

为zmq打开防火墙端口：

1
sudo ufw allow proto tcp to 0.0.0.0/0 port 5556 comment "zmq publish port of odas for sound source tracking"
2
sudo ufw allow proto tcp to 0.0.0.0/0 port 5557 comment "zmq publish port of odas for sound source localization"

注：我一开始为了用上Python 3.7，早早更新了buster的测试版，现在buster正式发布了，更新时报错：

1
E: Repository 'http://mirrors.aliyun.com/raspbian/raspbian buster InRelease' changed its 'Suite' value from 'testing' to 'stable'
2
N: This must be accepted explicitly before updates for this repository can be applied. See apt-secure(8) manpage for details.

按照Explicitly accept change for PPA ‘Label’方法中说的，把apt-get改成apt即可。

使用`pyzmq`发布定位数据#

1
from multiprocessing import Process
2
import os
3
import socket
4
import zmq
5

6

7
def sst_write(rpi_host='0.0.0.0', rpi_port=9000, zmq_host='*', zmq_port=5556):
8
    """
9
    A process, listened on socket://rpi_host:port,
10
        receive odas Sound Source Tracking(SST) json format data in bytes.
11
    Reassemble json byte frames and publish using zmq on tcp://zmq_host:zmq_port.
12
    :return: None
13
    """
14
    write_pid = os.getpid()
15
    print('#Write Process# %s start' % write_pid)
16
    bind_ip, bind_port = rpi_host, rpi_port
17
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
18
    server.bind((bind_ip, bind_port))
19
    server.listen(5)  # max backlog of connections
20
    print('#Write Process# listening on {}:{}'.format(bind_ip, bind_port))
21
    client_socket, address = server.accept()
22
    print('#Write Process# accepted connection from {}:{}'.format(address[0], address[1]))
23

24
    context = zmq.Context()
25
    zmq_socket = context.socket(zmq.PUB)
26
    zmq_socket.bind(f"tcp://{zmq_host}:{zmq_port}")
27

28
    temp = b''
29
    while True:
30
        t = 0
31
        temp = client_socket.recv(1024)
32
        for i in range(5):
33
            t = temp.find(b'}', t) + 1
34
        zmq_socket.send(temp[:t])
35
        temp = temp[t:]
36

37

38
def ssl_write(rpi_host='0.0.0.0', rpi_port=9001, zmq_host='*', zmq_port=5557):
39
    """
40
    A process, listened on socket://rpi_host:port,
41
        receive odas Sound Source Localization(SSL) json format data in bytes.
42
    Reassemble json byte frames and publish using zmq on tcp://zmq_host:zmq_port.
43
    :return: None
44
    """
45
    write_pid = os.getpid()
46
    print('#Write Process# %s start' % write_pid)
47
    bind_ip, bind_port = rpi_host, rpi_port
48
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
49
    server.bind((bind_ip, bind_port))
50
    server.listen(5)  # max backlog of connections
51
    print('#Write Process# listening on {}:{}'.format(bind_ip, bind_port))
52
    client_socket, address = server.accept()
53
    print('#Write Process# accepted connection from {}:{}'.format(address[0], address[1]))
54

55
    context = zmq.Context()
56
    zmq_socket = context.socket(zmq.PUB)
57
    zmq_socket.bind(f"tcp://{zmq_host}:{zmq_port}")
58

59
    temp = b''
60
    while True:
61
        t = 0
62
        temp = client_socket.recv(1024)
63
        for i in range(5):
64
            t = temp.find(b'}', t) + 1
65
        zmq_socket.send(temp[:t])
66
        temp = temp[t:]
67

68

69
if __name__ == '__main__':
70
    try:
71
        p_ssl_w = Process(target=ssl_write)
72
        p_sst_w = Process(target=sst_write)
73
        # p_ssl_r = Process(target=ssl_read, kwargs={"verbose": True})
74
        # 启动子进程pw，写入:
75
        p_ssl_w.start()
76
        p_sst_w.start()
77
        # 等待pw结束:
78
        p_ssl_w.join()
79
        p_sst_w.join()
80
    except KeyboardInterrupt:
81
        p_ssl_w.terminate()
82
        p_sst_w.terminate()
83
        print('KeyboardInterrupt pw pr Terminated')

远端主机#

使用`pyzmq`订阅定位数据#

1
from multiprocessing import Process
2
import os
3
import json
4
import asyncio
5
from aiohttp import web, WSMsgType
6
import zmq
7

8

9
def sst_read(zmq_host='192.168.1.108', zmq_port=5556, aio_host='0.0.0.0', aio_port=8080, verbose=False):
10
    """
11
    A process, used for subscribing zmq socket on tcp://zmq_host:zmq_port,
12
        receiving Sound Source Tracking(SST) data frames.
13
    In the meantime the process runs an aiohttp server on ws://aio_host:aio_port,
14
        sending SST data frames to user browser via websocket connection.
15
    User browser websocket-client should send a 'received' message after every data frame received,
16
        to tell the server 'last message received, please send next.'
17
    :output: Sound Source Tracking output like this:
18
    {
19
        "timeStamp": 45602,
20
        "src": [
21
            { "id": 43, "tag": "dynamic", "x": 0.025, "y": 0.128, "z": 0.991, "activity": 1.000 },
22
            { "id": 0, "tag": "", "x": 0.000, "y": 0.000, "z": 0.000, "activity": 0.000 },
23
            { "id": 0, "tag": "", "x": 0.000, "y": 0.000, "z": 0.000, "activity": 0.000 },
24
            { "id": 0, "tag": "", "x": 0.000, "y": 0.000, "z": 0.000, "activity": 0.000 }
25
        ]
26
    }
27
    :return: None
28
    """
29
    read_pid = os.getpid()
30
    print('#Read  Process# %s start' % read_pid)
31
    routes = web.RouteTableDef()
32

33
    context = zmq.Context()
34
    zmq_socket = context.socket(zmq.SUB)
35

36
    print("Collecting updates from odas publisher...")
37
    zmq_socket.connect(f"tcp://{zmq_host}:{zmq_port}")
38
    zmq_socket.setsockopt_string(zmq.SUBSCRIBE, '{')
39

40
    @routes.get('/ws')
41
    async def websocket_handler(request):
42
        print('websocket connection open')
43

44
        ws = web.WebSocketResponse()
45
        await ws.prepare(request)
46
        c = 0
47
        await ws.send_str(zmq_socket.recv_string())
48
        async for msg in ws:
49
            if msg.type == WSMsgType.TEXT:
50
                if msg.data == 'close':
51
                    print('websocket connection close')
52
                    await ws.close()
53
                else:
54
                    c += 1
55
                    tmp = zmq_socket.recv_string()
56
                    await ws.send_str(tmp)
57
                    if verbose:
58
                        c += 1
59
                        print(f'\r{c}', end='', flush=True)
60
                    # TODO: debug jump, delete later
61
                    # for i in range(100):
62
                    #     zmq_socket.recv_string()
63
            elif msg.type == WSMsgType.ERROR:
64
                print('ws connection closed with exception %s' %
65
                      ws.exception())
66
        return ws
67

68
    app = web.Application()
69
    app.add_routes(routes)
70
    web.run_app(app, host=aio_host, port=aio_port)
71

72

73
def ssl_read(zmq_host='192.168.1.108', zmq_port=5557, aio_host='0.0.0.0', aio_port=8081, verbose=False):
74
    """
75
    A process, used for subscribing zmq socket on tcp://zmq_host:zmq_port,
76
        receiving Sound Source Localization(SSL) data frames.
77
    In the meantime the process runs an aiohttp server on ws://aio_host:aio_port/ws,
78
        sending SSL data frames to user browser via websocket connection.
79
    User browser websocket-client should send a 'received' message after every data frame received,
80
        to tell the server 'last message received, please send next.'
81
    :output: Sound Source Localization output like this:
82
    {
83
        "timeStamp": 45608,
84
        "src": [
85
            { "x": 0.132, "y": 0.181, "z": 0.975, "E": 0.557 },
86
            { "x": 0.198, "y": 0.342, "z": 0.918, "E": 0.130 },
87
            { "x": 0.000, "y": 0.273, "z": 0.962, "E": 0.018 },
88
            { "x": 0.000, "y": 0.339, "z": 0.941, "E": 0.006 }
89
        ]
90
    }
91
    :return: None
92
    """
93
    read_pid = os.getpid()
94
    print('#Read  Process# %s start' % read_pid)
95
    routes = web.RouteTableDef()
96

97
    context = zmq.Context()
98
    zmq_socket = context.socket(zmq.SUB)
99

100
    print("Collecting updates from odas publisher...")
101
    zmq_socket.connect(f"tcp://{zmq_host}:{zmq_port}")
102
    zmq_socket.setsockopt_string(zmq.SUBSCRIBE, '{')
103

104
    @routes.get('/ws')
105
    async def websocket_handler(request):
106
        print('websocket connection open')
107

108
        ws = web.WebSocketResponse()
109
        await ws.prepare(request)
110
        c = 0
111
        await ws.send_str(zmq_socket.recv_string())
112
        async for msg in ws:
113
            if msg.type == WSMsgType.TEXT:
114
                if msg.data == 'close':
115
                    print('websocket connection close')
116
                    await ws.close()
117
                else:
118
                    tmp = zmq_socket.recv_string()
119
                    await ws.send_str(tmp)
120
                    if verbose:
121
                        c += 1
122
                        print(f'\r{c}', end='', flush=True)
123
                    # TODO: debug jump, delete later
124
                    # for i in range(100):
125
                    #     zmq_socket.recv_string()
126
            elif msg.type == WSMsgType.ERROR:
127
                print('ws connection closed with exception %s' %
128
                      ws.exception())
129
        return ws
130

131
    app = web.Application()
132
    app.add_routes(routes)
133
    web.run_app(app, host=aio_host, port=aio_port)
134

135

136
if __name__ == '__main__':
137
    try:
138
        p_ssl_r = Process(target=ssl_read, kwargs={"verbose": True})
139
        # p_ssl_r = Process(target=ssl_read)
140
        p_sst_r = Process(target=sst_read)
141
        # 启动子进程pr，读取:
142
        p_ssl_r.start()
143
        p_sst_r.start()
144
    except KeyboardInterrupt:
145
        p_ssl_r.terminate()
146
        p_sst_r.terminate()
147
        print('KeyboardInterrupt pw pr Terminated')

注意： 这里我使用了aiohttp将数据发布为websocket，在浏览器上进行可视化。

使用`three.js`可视化数据#

封装ant-design组件#

为了以后用着方便，我特意将可视化代码封装为ant-design的组件：

1
# 新建文件夹
2
mkdir odas-front
3
# 安装antd测试
4
npm install antd --save

注意： 在电脑上用nvm控制node版本，最新的已经是版本12.x了。但是port安装的yarn使用的是node --version 10.16.0，因此还是要配合安装nvm install 10.16.0，使默认node变成10.16.0（12版本会报错）。

1
安装`create-umi`
2
npm create umi
3
选择`app`，并同意使用`typescript`，再选择`dva`和`antd`

代码基本照抄odas的Element版本。