PythonでBLEを使ってスマホを検知する

　iPhoneならiBeacon、androidならEddyStoneという名前になっているBLEを使ってスマホの検知を行う。スマートホームなどのIoTを作成する場合、重要となるセンサーが人間の感知である。赤外線を使用したセンサーは動作を検知するものなので座っていたり寝ていたりすると反応しないため人間のセンサーには向かない。そこでスマホが部屋にあるかを判断して、人がいるかを判断することにした。

　まず、初めに使ったものはWi-Fiである。Wi-FiのpingコマンドでスマホがWi-Fiに接続されているかを監視する。Pingの監視はサーバーなどの管理に使用されているようでググるといろいろ出てきた。検出してみたがスリープにして5分でpingが通らなくなることが分かった。自分は、androidを使用しているのだがどうも電力設定が邪魔しているようだ。そこで、BLEに目を付けた。

BLE 【 Bluetooth Low Energy 】とは
Bluetooth 4.0の規格の一部である。物理設計からモデル使用に及ぶあらゆる面で電力消費を最小に保つよう設計されている。

　自分はNuAns NEO [Reloaded] というandroidを使っているのだがたまたまBLEに対応していた。最近のスマホはほとんど対応していると思う。Raspberry Piも3から対応している。なぜBLEかというと消費電力が少ないためである。Pingコマンドの方法では部屋にいるときに電池の消費が激しくなるという問題があった。だが、BLEは24時間送信していてもほとんど電池を食わない。また、通信距離を電波の強弱で制御することができるため部屋ごとに判断するといった使い方もできる。

Raspberry Piでスマホを検知する

スマホにBLEのツールをインストールする。AndroidならBeacon Simulator、iPhoneはもっていないのでBLEなどと検索して探してみてください。ここではandroid用の間隔や強さを設定できるものが使いやすい。AndroidのBeacon Simulatorで説明していきます。

アプリを起動し左下の＋で新規追加を行う。

UUIDはランダムな値がはじめ入力されている。
Major:0
Minor:0
Broadcast settings
Power:Medium(電池の消費を抑えたい場合Lowに)
Frequency: Balanced(電池の消費を抑えたい場合Lowに)

チェックを押し、初期画面でtestを有効にする。

これよりPCでの作業になる。
ここから実行ファイルをダウンロードし、解凍したものをRaspberry Piにコピーする。コピー先は、/home/piとする。端末、ターミナルを開き以下のコマンドを実行する。
コマンド
sudo python iBeacon-Scanner–master/testblescan.py

すると以下の様な文字列が表示される。Ctrl+cで終了する。00:bb:c1:37:70:34,00000277c0a80b040000000000000000,0,0,-56,-83

左からMACアドレス、UUID、Major、Minor、TxPower、RSSIとなる。設定していた値と同じものが表示されればBLEを認識できている。

次にUUIDで識別し、スマホが認識した場合は１を、認識できなかった場合は0を返すようにプログラムを編集する。
コマンド
cd iBeacon-Scanner—master
nano blescan.py

# BLE iBeaconScanner based on https://github.com/adamf/BLE/blob/master/ble-scanner.py
# JCS 06/07/14

DEBUG = False
# BLE scanner based on https://github.com/adamf/BLE/blob/master/ble-scanner.py
# BLE scanner, based on https://code.google.com/p/pybluez/source/browse/trunk/examples/advanced/inquiry-with-rssi.py

# https://github.com/pauloborges/bluez/blob/master/tools/hcitool.c for lescan
# https://kernel.googlesource.com/pub/scm/bluetooth/bluez/+/5.6/lib/hci.h for opcodes
# https://github.com/pauloborges/bluez/blob/master/lib/hci.c#L2782 for functions used by lescan

# performs a simple device inquiry, and returns a list of ble advertizements 
# discovered device

# NOTE: Python's struct.pack() will add padding bytes unless you make the endianness explicit. Little endian
# should be used for BLE. Always start a struct.pack() format string with "<"

import os
import sys
import struct
import bluetooth._bluetooth as bluez

LE_META_EVENT = 0x3e
LE_PUBLIC_ADDRESS=0x00
LE_RANDOM_ADDRESS=0x01
LE_SET_SCAN_PARAMETERS_CP_SIZE=7
OGF_LE_CTL=0x08
OCF_LE_SET_SCAN_PARAMETERS=0x000B
OCF_LE_SET_SCAN_ENABLE=0x000C
OCF_LE_CREATE_CONN=0x000D

LE_ROLE_MASTER = 0x00
LE_ROLE_SLAVE = 0x01

# these are actually subevents of LE_META_EVENT
EVT_LE_CONN_COMPLETE=0x01
EVT_LE_ADVERTISING_REPORT=0x02
EVT_LE_CONN_UPDATE_COMPLETE=0x03
EVT_LE_READ_REMOTE_USED_FEATURES_COMPLETE=0x04

# Advertisment event types
ADV_IND=0x00
ADV_DIRECT_IND=0x01
ADV_SCAN_IND=0x02
ADV_NONCONN_IND=0x03
ADV_SCAN_RSP=0x04

def returnnumberpacket(pkt):
    myInteger = 0
    multiple = 256
    for c in pkt:
        myInteger +=  struct.unpack("B",c)[0] * multiple
        multiple = 1
    return myInteger 

def returnstringpacket(pkt):
    myString = "";
    for c in pkt:
        myString +=  "%02x" %struct.unpack("B",c)[0]
    return myString 

def printpacket(pkt):
    for c in pkt:
        sys.stdout.write("%02x " % struct.unpack("B",c)[0])

def get_packed_bdaddr(bdaddr_string):
    packable_addr = []
    addr = bdaddr_string.split(':')
    addr.reverse()
    for b in addr: 
        packable_addr.append(int(b, 16))
    return struct.pack("<BBBBBB", *packable_addr)

def packed_bdaddr_to_string(bdaddr_packed):
    return ':'.join('%02x'%i for i in struct.unpack("<BBBBBB", bdaddr_packed[::-1]))

def hci_enable_le_scan(sock):
    hci_toggle_le_scan(sock, 0x01)

def hci_disable_le_scan(sock):
    hci_toggle_le_scan(sock, 0x00)

def hci_toggle_le_scan(sock, enable):
# hci_le_set_scan_enable(dd, 0x01, filter_dup, 1000);
# memset(&scan_cp, 0, sizeof(scan_cp));
 #uint8_t         enable;
 #       uint8_t         filter_dup;
#        scan_cp.enable = enable;
#        scan_cp.filter_dup = filter_dup;
#
#        memset(&rq, 0, sizeof(rq));
#        rq.ogf = OGF_LE_CTL;
#        rq.ocf = OCF_LE_SET_SCAN_ENABLE;
#        rq.cparam = &scan_cp;
#        rq.clen = LE_SET_SCAN_ENABLE_CP_SIZE;
#        rq.rparam = &status;
#        rq.rlen = 1;

#        if (hci_send_req(dd, &rq, to) < 0)
#                return -1;
    cmd_pkt = struct.pack("<BB", enable, 0x00)
    bluez.hci_send_cmd(sock, OGF_LE_CTL, OCF_LE_SET_SCAN_ENABLE, cmd_pkt)

def hci_le_set_scan_parameters(sock):
    old_filter = sock.getsockopt( bluez.SOL_HCI, bluez.HCI_FILTER, 14)

    SCAN_RANDOM = 0x01
    OWN_TYPE = SCAN_RANDOM
    SCAN_TYPE = 0x01

    
def parse_events(sock, loop_count=100):
    old_filter = sock.getsockopt( bluez.SOL_HCI, bluez.HCI_FILTER, 14)

    # perform a device inquiry on bluetooth device #0
    # The inquiry should last 8 * 1.28 = 10.24 seconds
    # before the inquiry is performed, bluez should flush its cache of
    # previously discovered devices
    flt = bluez.hci_filter_new()
    bluez.hci_filter_all_events(flt)
    bluez.hci_filter_set_ptype(flt, bluez.HCI_EVENT_PKT)
    sock.setsockopt( bluez.SOL_HCI, bluez.HCI_FILTER, flt )
    done = False
    results = []
    myFullList = []
    for i in range(0, loop_count):
        pkt = sock.recv(255)
        ptype, event, plen = struct.unpack("BBB", pkt[:3])
        #print "--------------" 
        if event == bluez.EVT_INQUIRY_RESULT_WITH_RSSI:
        i =0
        elif event == bluez.EVT_NUM_COMP_PKTS:
                i =0 
        elif event == bluez.EVT_DISCONN_COMPLETE:
                i =0 
        elif event == LE_META_EVENT:
            subevent, = struct.unpack("B", pkt[3])
            pkt = pkt[4:]
            if subevent == EVT_LE_CONN_COMPLETE:
                le_handle_connection_complete(pkt)
            elif subevent == EVT_LE_ADVERTISING_REPORT:
                #print "advertising report"
                num_reports = struct.unpack("B", pkt[0])[0]
                report_pkt_offset = 0
                for i in range(0, num_reports):
        
            if (DEBUG == True):
            print "-------------"
                        #print "\tfullpacket: ", printpacket(pkt)
                print "\tUDID: ", printpacket(pkt[report_pkt_offset -22: report_pkt_offset - 6])
                print "\tMAJOR: ", printpacket(pkt[report_pkt_offset -6: report_pkt_offset - 4])
                print "\tMINOR: ", printpacket(pkt[report_pkt_offset -4: report_pkt_offset - 2])
                        print "\tMAC address: ", packed_bdaddr_to_string(pkt[report_pkt_offset + 3:report_pkt_offset + 9])
                # commented out - don't know what this byte is.  It's NOT TXPower
                        txpower, = struct.unpack("b", pkt[report_pkt_offset -2])
                        print "\t(Unknown):", txpower
    
                        rssi, = struct.unpack("b", pkt[report_pkt_offset -1])
                        print "\tRSSI:", rssi
            # build the return string
                    Adstring = packed_bdaddr_to_string(pkt[report_pkt_offset + 3:report_pkt_offset + 9])
            Adstring += ","
            Adstring += returnstringpacket(pkt[report_pkt_offset -22: report_pkt_offset - 6]) 
            Adstring += ","
            Adstring += "%i" % returnnumberpacket(pkt[report_pkt_offset -6: report_pkt_offset - 4]) 
            Adstring += ","
            Adstring += "%i" % returnnumberpacket(pkt[report_pkt_offset -4: report_pkt_offset - 2]) 
            Adstring += ","
            Adstring += "%i" % struct.unpack("b", pkt[report_pkt_offset -2])
            Adstring += ","
            Adstring += "%i" % struct.unpack("b", pkt[report_pkt_offset -1])

            #print "\tAdstring=", Adstring
            myFullList.append(Adstring)
                done = True
    sock.setsockopt( bluez.SOL_HCI, bluez.HCI_FILTER, old_filter )
    return myFullList

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

# BLE iBeaconScanner based on https://github.com/adamf/BLE/blob/master/ble-scanner.py

# JCS 06/07/14

DEBUG = False

# BLE scanner based on https://github.com/adamf/BLE/blob/master/ble-scanner.py

# BLE scanner, based on https://code.google.com/p/pybluez/source/browse/trunk/examples/advanced/inquiry-with-rssi.py

# https://github.com/pauloborges/bluez/blob/master/tools/hcitool.c for lescan

# https://kernel.googlesource.com/pub/scm/bluetooth/bluez/+/5.6/lib/hci.h for opcodes

# https://github.com/pauloborges/bluez/blob/master/lib/hci.c#L2782 for functions used by lescan

# performs a simple device inquiry, and returns a list of ble advertizements

# discovered device

# NOTE: Python's struct.pack() will add padding bytes unless you make the endianness explicit. Little endian

# should be used for BLE. Always start a struct.pack() format string with "<"

import os

import sys

import struct

import bluetooth._bluetooth as bluez

LE_META_EVENT = 0x3e

LE_PUBLIC_ADDRESS=0x00

LE_RANDOM_ADDRESS=0x01

LE_SET_SCAN_PARAMETERS_CP_SIZE=7

OGF_LE_CTL=0x08

OCF_LE_SET_SCAN_PARAMETERS=0x000B

OCF_LE_SET_SCAN_ENABLE=0x000C

OCF_LE_CREATE_CONN=0x000D

LE_ROLE_MASTER = 0x00

LE_ROLE_SLAVE = 0x01

# these are actually subevents of LE_META_EVENT

EVT_LE_CONN_COMPLETE=0x01

EVT_LE_ADVERTISING_REPORT=0x02

EVT_LE_CONN_UPDATE_COMPLETE=0x03

EVT_LE_READ_REMOTE_USED_FEATURES_COMPLETE=0x04

# Advertisment event types

ADV_IND=0x00

ADV_DIRECT_IND=0x01

ADV_SCAN_IND=0x02

ADV_NONCONN_IND=0x03

ADV_SCAN_RSP=0x04

def returnnumberpacket(pkt):

myInteger = 0

multiple = 256

for c in pkt:

myInteger += struct.unpack("B",c)[0] * multiple

multiple = 1

return myInteger

def returnstringpacket(pkt):

myString = "";

for c in pkt:

myString += "%02x" %struct.unpack("B",c)[0]

return myString

def printpacket(pkt):

for c in pkt:

sys.stdout.write("%02x " % struct.unpack("B",c)[0])

def get_packed_bdaddr(bdaddr_string):

packable_addr = []

addr = bdaddr_string.split(':')

addr.reverse()

for b in addr:

packable_addr.append(int(b, 16))

return struct.pack("<BBBBBB", *packable_addr)

def packed_bdaddr_to_string(bdaddr_packed):

return ':'.join('%02x'%i for i in struct.unpack("<BBBBBB", bdaddr_packed[::-1]))

def hci_enable_le_scan(sock):

hci_toggle_le_scan(sock, 0x01)

def hci_disable_le_scan(sock):

hci_toggle_le_scan(sock, 0x00)

def hci_toggle_le_scan(sock, enable):

# hci_le_set_scan_enable(dd, 0x01, filter_dup, 1000);

# memset(&scan_cp, 0, sizeof(scan_cp));

#uint8_t enable;

# uint8_t filter_dup;

# scan_cp.enable = enable;

# scan_cp.filter_dup = filter_dup;

# memset(&rq, 0, sizeof(rq));

# rq.ogf = OGF_LE_CTL;

# rq.ocf = OCF_LE_SET_SCAN_ENABLE;

# rq.cparam = &scan_cp;

# rq.clen = LE_SET_SCAN_ENABLE_CP_SIZE;

# rq.rparam = &status;

# rq.rlen = 1;

# if (hci_send_req(dd, &rq, to) < 0)

# return -1;

cmd_pkt = struct.pack("<BB", enable, 0x00)

bluez.hci_send_cmd(sock, OGF_LE_CTL, OCF_LE_SET_SCAN_ENABLE, cmd_pkt)

def hci_le_set_scan_parameters(sock):

old_filter = sock.getsockopt( bluez.SOL_HCI, bluez.HCI_FILTER, 14)

SCAN_RANDOM = 0x01

OWN_TYPE = SCAN_RANDOM

SCAN_TYPE = 0x01

def parse_events(sock, loop_count=100):

old_filter = sock.getsockopt( bluez.SOL_HCI, bluez.HCI_FILTER, 14)

# perform a device inquiry on bluetooth device #0

# The inquiry should last 8 * 1.28 = 10.24 seconds

# before the inquiry is performed, bluez should flush its cache of

# previously discovered devices

flt = bluez.hci_filter_new()

bluez.hci_filter_all_events(flt)

bluez.hci_filter_set_ptype(flt, bluez.HCI_EVENT_PKT)

sock.setsockopt( bluez.SOL_HCI, bluez.HCI_FILTER, flt )

done = False

results = []

myFullList = []

for i in range(0, loop_count):

pkt = sock.recv(255)

ptype, event, plen = struct.unpack("BBB", pkt[:3])

#print "--------------"

if event == bluez.EVT_INQUIRY_RESULT_WITH_RSSI:

i =0

elif event == bluez.EVT_NUM_COMP_PKTS:

i =0

elif event == bluez.EVT_DISCONN_COMPLETE:

i =0

elif event == LE_META_EVENT:

subevent, = struct.unpack("B", pkt[3])

pkt = pkt[4:]

if subevent == EVT_LE_CONN_COMPLETE:

le_handle_connection_complete(pkt)

elif subevent == EVT_LE_ADVERTISING_REPORT:

#print "advertising report"

num_reports = struct.unpack("B", pkt[0])[0]

report_pkt_offset = 0

for i in range(0, num_reports):

if (DEBUG == True):

print "-------------"

#print "\tfullpacket: ", printpacket(pkt)

print "\tUDID: ", printpacket(pkt[report_pkt_offset -22: report_pkt_offset - 6])

print "\tMAJOR: ", printpacket(pkt[report_pkt_offset -6: report_pkt_offset - 4])

print "\tMINOR: ", printpacket(pkt[report_pkt_offset -4: report_pkt_offset - 2])

print "\tMAC address: ", packed_bdaddr_to_string(pkt[report_pkt_offset + 3:report_pkt_offset + 9])

# commented out - don't know what this byte is. It's NOT TXPower

txpower, = struct.unpack("b", pkt[report_pkt_offset -2])

print "\t(Unknown):", txpower

rssi, = struct.unpack("b", pkt[report_pkt_offset -1])

print "\tRSSI:", rssi

# build the return string

Adstring = packed_bdaddr_to_string(pkt[report_pkt_offset + 3:report_pkt_offset + 9])

Adstring += ","

Adstring += returnstringpacket(pkt[report_pkt_offset -22: report_pkt_offset - 6])

Adstring += ","

Adstring += "%i" % returnnumberpacket(pkt[report_pkt_offset -6: report_pkt_offset - 4])

Adstring += ","

Adstring += "%i" % returnnumberpacket(pkt[report_pkt_offset -4: report_pkt_offset - 2])

Adstring += ","

Adstring += "%i" % struct.unpack("b", pkt[report_pkt_offset -2])

Adstring += ","

Adstring += "%i" % struct.unpack("b", pkt[report_pkt_offset -1])

#print "\tAdstring=", Adstring

myFullList.append(Adstring)

done = True

sock.setsockopt( bluez.SOL_HCI, bluez.HCI_FILTER, old_filter )

return myFullList

青色部分を以下に置き換える。

            Adstring = returnstringpacket(pkt[report_pkt_offset -22: report_pkt_offset - 6])
                    if (Adstring=='アプリで設定したUUID'):
                        return 1
    return 0

同様にしてtestblescan.pyも書き換える。
コマンド
nano testblescan.py

# test BLE Scanning software
# jcs 6/8/2014

import blescan
import sys

import bluetooth._bluetooth as bluez

dev_id = 0
try:
    sock = bluez.hci_open_dev(dev_id)
    print "ble thread started"

except:
    print "error accessing bluetooth device..."
        sys.exit(1)

blescan.hci_le_set_scan_parameters(sock)
blescan.hci_enable_le_scan(sock)

while True:
    returnedList = blescan.parse_events(sock, 10)
    print "----------"
    for beacon in returnedList:
        print beacon

# test BLE Scanning software

# jcs 6/8/2014

import blescan

import sys

import bluetooth._bluetooth as bluez

dev_id = 0

try:

sock = bluez.hci_open_dev(dev_id)

print "ble thread started"

except:

print "error accessing bluetooth device..."

sys.exit(1)

blescan.hci_le_set_scan_parameters(sock)

blescan.hci_enable_le_scan(sock)

while True:

returnedList = blescan.parse_events(sock, 10)

print "----------"

for beacon in returnedList:

print beacon

青色部分を以下に置き換える。

returned = blescan.parse_events(sock, 30print returned

30は検索回数である。増やすと精度が上がるが時間がかかる。
以下のコマンドを実行し確認する。
コマンド
sudo python iBeacon-Scanner–master/testblescan.py

BLEを送信しているとき1が表示されBLEを切断しているときに0が表示されれば完成となる。