[UMEHOSHI ITA]をUART を介して、Bluetoothで制御する

001
002
003
004
005
006
007
008
009
010
011
012
013
014
015
016
017
018
019
020
021
022
023
024
025
026
027
028
029
030
031
032
033
034
035
036
037
038
039
040
041
042
043
044
045
046
047
048
049
050
051
052
053
054
055
056
057
058
059
060
061
062
063
064
065
066
067
068
069
070
071

import sys         # btc.py 
import time
import threading

# https://pybluez.readthedocs.io/en/latest/install.html
import bluetooth

devices = bluetooth.discover_devices(lookup_names=True)
print("Found {} devices.".format(len(devices)))

n=0
for addr, name in devices: # ペアリングデバイスリストアップ
    print("No={}  {} - {}".format(n, addr, name))
    n+=1

no=input("Device No=>>")

addr, name = devices[int(no)] # 上記選択デバイスをメモ
sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM) # ソケット生成
uuid = '00001101-0000-1000-8000-00805F9B34FB' # SPP (シリアルポートプロファイル) の UUID
service = []
retry = 1
while len(service) == 0: # 接続予定のデバイスのサービスが存在するか？のチェックで、見つかるまで繰り返す。
     print( "Looking for service on {0}, try {1}".format(addr, retry) )
     service = bluetooth.find_service(address=addr, uuid=uuid)
     retry = retry + 1
     time.sleep(1)

if len(service) != 1:
    print("not find service.")
    exit(1)

# print(service)
service = service[0]
for k in service: # 見つかったデバイス情報を表示
    print("\t", k, ":" , service[k])

print( "Service found. Name={0}".format(service["name"]) )
print( "Connecting to service." )
sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
try:
   sock.connect((addr, service["port"])) # 接続！！
   print( "Connected to service on {0} on port {1}".format(addr, service["port"]) )
except bluetooth.btcommon.BluetoothError as e:
   print( "Connection failed: {0}".format(e) )
   exit(0)

loop_flag=True
def read_loop(): # 受信スレッド用の関数
    global loop_flag
    while loop_flag:
        try:
            b=sock.recv(1) # binaryで1byte入力
            #print(type(b), b)
            print( chr(b[0]), end="") # 文字に変換して表示
        except:
           print("End Thread") 
           loop_flag = False
           return

# スレッドに read_loop 関数を渡す
t_id = threading.Thread(target=read_loop)
t_id.start() # スレッドスタート

while True: # 送信用のループ
    s = input("input>") # 送信文字列のキー入力
    if s == "": break # Enterだけすると、終了
    s += "\r\n"
    sock.send(s.encode('ascii')) # byteに変換して送信

sock.close()

001
002
003
004
005
006
007
008
009
010
011
012
013
014
015
016
017
018
019
020
021
022
023
024
025
026
027
028
029
030
031
032
033
034
035
036
037
038
039
040
041
042
043
044
045
046
047
048
049
050
051
052
053
054
055
056
057
058
059
060
061
062
063
064
065
066
067
068
069
070
071
072
073
074
075
076
077
078
079
080
081
082
083
084
085
086
087
088
089
090
091
092
093
094
095
096
097
098
099
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

import sys         # btc2.py 
import os
import time
import threading

# https://pybluez.readthedocs.io/en/latest/install.html
import bluetooth

devices = bluetooth.discover_devices(lookup_names=True)
print("Found {} devices.".format(len(devices)))

addr=None
addr, name ="00:06:66:7D:BE:FE","RNBT-BEFE" # 接続対象のBluetooth機器が分かればここで設定する

if addr is None:
    n=0
    for addr, name in devices: # ペアリングデバイスリストアップ
        print("No={}  {} - {}".format(n, addr, name))
        n+=1
    #
    no=input("Device No=>>")
    addr, name = devices[int(no)] # 上記選択デバイスをメモ

sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM) # ソケット生成
uuid = '00001101-0000-1000-8000-00805F9B34FB' # SPP (シリアルポートプロファイル) の UUID
service = []
retry = 1
while len(service) == 0: # 接続予定のデバイスのサービスが存在するか？のチェックで、見つかるまで繰り返す。
     print( "Looking for service on {0}, try {1}".format(addr, retry) )
     service = bluetooth.find_service(address=addr, uuid=uuid)
     retry = retry + 1
     time.sleep(0.5)

if len(service) != 1:
    print("not find service.")
    exit(1)

# print(service)
service = service[0]
for k in service: # 見つかったデバイス情報を表示
    print("\t", k, ":" , service[k])

print( "Service found. Name={0}".format(service["name"]) )
print( "Connecting to service." )
sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
try:
   sock.connect((addr, service["port"])) # 接続！！
   print( "Connected to service on {0} on port {1}".format(addr, service["port"]) )
except bluetooth.btcommon.BluetoothError as e:
   print( "Connection failed: {0}".format(e) )
   exit(0)

loop_flag=True
def read_loop(): # 受信スレッド用の関数
    global loop_flag
    while loop_flag:
        try:
            b=sock.recv(1) # binaryで1byte入力
            #print(type(b), b)
            print( chr(b[0]), end="") # 文字に変換して表示
        except:
           print("End Thread") 
           loop_flag = False
           return

# スレッドに read_loop 関数を渡す
t_id = threading.Thread(target=read_loop)
t_id.start() # スレッドスタート

def send_file( ): # hexファイルをbluetoothで転送して実行
    print(os.listdir("."))
    file = input("file name>")
    if not os.path.isfile( file ) : return
    with open ( file ) as f:
        datas = f.readlines()
    if file.endswith(".umh"): datas = datas[1:]
    for s in datas:
        s = s.strip() + "\r\n"
        print(s, end="")
        s = s.encode('ascii')
        sock.send(s) # byteに変換して送信
        time.sleep(0.1)

def get_check_sum(s): # チェックサム用２文字を取得
    ''' ck = get_check_sum("G108000900000") で"67"が取得
    '''
    ck = 0
    for c in s: ck += ord(c)
    ck &= 0x0ff
    return f"{ 0x100-ck :02X}" # check sum の16進2文字

def get_memory( ): # メモリ内容の表示
    try:
        adr=int(input("memory hex address>"), 16)
        size=int(input("get memory hex size>"), 16)
        cmd = f"G{size:02X}{adr:08X}00"
        cmd += get_check_sum( cmd ) + "\r\n"
        s = cmd.encode('ascii')
        #print(cmd, end="")
        sock.send(s) # byteに変換して送信
        time.sleep(0.5)
    #
    except Excption as e: 
        print(e)

def run_from_address( ): # 指定アドレスからの実行
    try:
        adr=int(input("memory hex address>"), 16)
        cmd = f"R00{adr:08X}00"
        cmd += get_check_sum( cmd ) + "\r\n"
        s = cmd.encode('ascii')
        #print(cmd, end="")
        sock.send(s) # byteに変換して送信
        time.sleep(0.5)
    #
    except Excption as e: 
        print(e)

while True: # 送信用のループ
    s = input("[quit|S|G|R|f|g|r]input>") .strip() # 送信文字列のキー入力
    if s == "quit": break # quit入力で、終了
    elif s == 'f': send_file( )
    elif s == 'g': get_memory( )
    elif s == 'r': run_from_address( )
    elif s=="" or not s[0] in "SGR":
        s += "\r\n"
        sock.send(s.encode('ascii')) # byteに変換して送信
        time.sleep(0.1)

sock.close()