Apple Watch から BLE でサーボモーターを動かす！（PythonとSwift）

はじめに

前回動かしたサーボモーターを Apple Watch から動かしたいと思います。

デジタル・マイクロサーボ SG90 (1個) 新品価格 ￥440から (2021/3/20 16:31時点)

ラズパイは Raspberry Pi 3 Model B です。
その他必要なやつ

オスメスジャンバ線３本を使いました。

配線

前回と同じです。

公式ドキュメントから引用

1. ジャンバ線１のオスをモーターの茶色に挿す。
2. ジャンバ線１のメスをラズパイの 5V 横の GND に挿す。
3. ジャンバ線２のオスをモーターの黄色に挿す。
4. ジャンバ線２のメスをラズパイの GPIO14 に挿す。
5. ジャンバ線３のオスをモーターの赤色に挿す。
6. ジャンバ線３のメスをラズパイの 5V に挿す。

ラズパイ側の設定

コードを書いていきます（なんとなく pipenv 使います）。

1. SSH 接続する。

2. 必要なものを入れていく（実行時にパーミッションエラーになったので sudo でやってます。よくわかってない。。。）。

   # BLEに必要なやつ $ sudo apt-get install bluetooth bluez libbluetooth-dev libudev-dev # ディレクトリを作成 $ mkdir ble_servo # ディレクトリに移動 $ cd ble_servo # python3系でpipenv初期化 $ pipenv --python 3 # RPi.GPIOインストール $ sudo pipenv install RPi.GPIO # pyblenoインストール $ sudo pipenv install pybleno # servo.pyファイル作成 $ touch servo.py # servo.pyファイル編集 $ vim servo.py

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

   # BLEに必要なやつ $ sudo apt-get install bluetooth bluez libbluetooth-dev libudev-dev   # ディレクトリを作成 $ mkdir ble_servo   # ディレクトリに移動 $ cd ble_servo   # python3系でpipenv初期化 $ pipenv --python 3   # RPi.GPIOインストール $ sudo pipenv install RPi.GPIO   # pyblenoインストール $ sudo pipenv install pybleno   # servo.pyファイル作成 $ touch servo.py   # servo.pyファイル編集 $ vim servo.py

3. i と入力し入力モードにする。

4. 下記ソースをコピペする。

   from pybleno import * import RPi.GPIO as GPIO import time import sys import signal bleno = Bleno() class ServoCharacteristic(Characteristic): def __init__(self): Characteristic.__init__(self, { 'uuid': 'ff11', 'properties': ['read', 'write'], 'value': None }) self._value = 0 self._updateValueCallback = None def move(self): GPIO.setmode(GPIO.BCM) GPIO.setup(14, GPIO.OUT) servo = GPIO.PWM(14, 50) servo.start(0.0) servo.ChangeDutyCycle(2.5) time.sleep(0.6) servo.ChangeDutyCycle(10.0) def onWriteRequest(self, data, offset, withoutResponse, callback): self._value = data print('EchoCharacteristic - %s - onWriteRequest: value = %s' % (self['uuid'], [hex(c) for c in self._value])) self.move() if self._updateValueCallback: self._updateValueCallback(self._value) callback(Characteristic.RESULT_SUCCESS) def onReadRequest(self, offset, callback): print('onReadRequest') callback(Characteristic.RESULT_SUCCESS, self._value) def onSubscribe(self, maxValueSize, updateValueCallback): print('onSubscribe') self._updateValueCallback = updateValueCallback def onUnsubscribe(self): print('onUnsubscribe') self._updateValueCallback = None def onStateChange(state): print('stateChange: ' + state) if (state == 'poweredOn'): bleno.startAdvertising('Servo', ['FF10']) else: bleno.stopAdvertising() bleno.on('stateChange', onStateChange) servoCharacteristic = ServoCharacteristic() def onAdvertisingStart(error): print('advertisingStart: ' + ('error ' + error if error else 'success')) if not error: bleno.setServices([ BlenoPrimaryService({ 'uuid': 'FF10', 'characteristics': [servoCharacteristic] }) ]) bleno.on('advertisingStart', onAdvertisingStart) bleno.start() # Ctrl+CによってSIGINTシグナルが送信された時のハンドラ。終了前にGPIO.cleanupを呼び出す def handler(signum, frame): print('Signal handler called with signal', signum) bleno.stopAdvertising() bleno.disconnect() GPIO.cleanup() sys.exit(0) # ハンドラの登録 signal.signal(signal.SIGINT, handler) while True: time.sleep(0.1)

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93

   from pybleno import * import RPi.GPIO as GPIO import time import sys import signal   bleno = Bleno()   class ServoCharacteristic(Characteristic):       def __init__(self):         Characteristic.__init__(self, {             'uuid': 'ff11',             'properties': ['read', 'write'],             'value': None         })           self._value = 0         self._updateValueCallback = None       def move(self):         GPIO.setmode(GPIO.BCM)         GPIO.setup(14, GPIO.OUT)           servo = GPIO.PWM(14, 50)         servo.start(0.0)           servo.ChangeDutyCycle(2.5)         time.sleep(0.6)         servo.ChangeDutyCycle(10.0)       def onWriteRequest(self, data, offset, withoutResponse, callback):         self._value = data           print('EchoCharacteristic - %s - onWriteRequest: value = %s' % (self['uuid'], [hex(c) for c in self._value]))           self.move()           if self._updateValueCallback:             self._updateValueCallback(self._value)           callback(Characteristic.RESULT_SUCCESS)       def onReadRequest(self, offset, callback):         print('onReadRequest')         callback(Characteristic.RESULT_SUCCESS, self._value)       def onSubscribe(self, maxValueSize, updateValueCallback):         print('onSubscribe')         self._updateValueCallback = updateValueCallback       def onUnsubscribe(self):         print('onUnsubscribe')         self._updateValueCallback = None   def onStateChange(state):     print('stateChange: ' + state)       if (state == 'poweredOn'):         bleno.startAdvertising('Servo', ['FF10'])     else:         bleno.stopAdvertising()   bleno.on('stateChange', onStateChange) servoCharacteristic = ServoCharacteristic()   def onAdvertisingStart(error):     print('advertisingStart: ' + ('error ' + error if error else 'success'))       if not error:         bleno.setServices([             BlenoPrimaryService({                 'uuid': 'FF10',                 'characteristics': [servoCharacteristic]             })         ])   bleno.on('advertisingStart', onAdvertisingStart) bleno.start()   # Ctrl+CによってSIGINTシグナルが送信された時のハンドラ。終了前にGPIO.cleanupを呼び出す def handler(signum, frame):     print('Signal handler called with signal', signum)     bleno.stopAdvertising()     bleno.disconnect()     GPIO.cleanup()     sys.exit(0)   # ハンドラの登録 signal.signal(signal.SIGINT, handler)   while True:     time.sleep(0.1)

Watch側の設定

watch 側のコードを書いていきます（L チカやってたので名称がそのままになってます）。

1. Xcode の新規プロジェクト作成で「Watch App」を選択する。
2. Storyboard で InterfaceController を下記のように設定する。
   
3. Watch Kit の info.plist に NSBluetoothAlwaysUsageDescription キーを追加する。

4. InterfaceController.swift に下記をコピペする。

   import WatchKit import Foundation import CoreBluetooth class InterfaceController: WKInterfaceController { private var centralManager: CBCentralManager! private var peripheral: CBPeripheral? private var lightCharacteristic: CBCharacteristic? @IBOutlet private weak var ledSwitch: WKInterfaceSwitch! override func awake(withContext context: Any?) { centralManager = CBCentralManager(delegate: self, queue: nil) } @IBAction private func scanForPeripherals() { centralManager.scanForPeripherals(withServices: nil, options: nil) } @IBAction private func connectPeripheral(_ sender: Any) { if let peripheral = peripheral { centralManager.connect(peripheral, options: nil) } } @IBAction private func switchLED(_ value: Bool) { guard let peripheral = peripheral, let lightCharacteristic = lightCharacteristic else { ledSwitch.setOn(!value) return } var intValue: CUnsignedInt = value ? 1 : 0 print("write characteristic and \(value)") let data = Data(bytes: &intValue, count: 1) peripheral.writeValue(data, for: lightCharacteristic, type: .withResponse) } } extension InterfaceController: CBCentralManagerDelegate { func centralManagerDidUpdateState(_ central: CBCentralManager) { switch central.state { case .unknown: print("state: unknown") case .resetting: print("state: resetting") case .unsupported: print("state: unsupported") case .unauthorized: print("state: unauthorized") case .poweredOff: print("state: poweredOff") case .poweredOn: print("state: poweredOn") @unknown default: assertionFailure("想定外") } } func centralManager(_ central: CBCentralManager, didDiscover peripheral: CBPeripheral, advertisementData: [String : Any], rssi RSSI: NSNumber) { guard let peripheralName = peripheral.name else { return } print("peripheral: \(peripheralName)") if peripheralName == "raspberrypi" { centralManager.stopScan() self.peripheral = peripheral } } func centralManager(_ central: CBCentralManager, didConnect peripheral: CBPeripheral) { print("connected") peripheral.delegate = self peripheral.discoverServices(nil) } func centralManager(_ central: CBCentralManager, didFailToConnect peripheral: CBPeripheral, error: Error?) { print("fail connect") } } extension InterfaceController: CBPeripheralDelegate { func peripheral(_ peripheral: CBPeripheral, didDiscoverServices error: Error?) { guard let services = peripheral.services else { if let error = error { print("error: \(error)") } return } print("Found \(services.count) services: \(services)") for service in services { print(service.uuid.uuidString) if service.uuid.uuidString == "FF10" { print("discovering characteristics") peripheral.discoverCharacteristics(nil, for: service) } } } func peripheral(_ peripheral: CBPeripheral, didDiscoverCharacteristicsFor service: CBService, error: Error?) { guard let characteristics = service.characteristics else { if let error = error { print("error in dicovercharc: \(error)") } return } print("Found \(characteristics) characteristics! : \(characteristics)") for characteristic in characteristics { if characteristic.uuid.uuidString == "FF11" { self.lightCharacteristic = characteristic } } } }

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121

   import WatchKit import Foundation import CoreBluetooth   class InterfaceController: WKInterfaceController {       private var centralManager: CBCentralManager!     private var peripheral: CBPeripheral?     private var lightCharacteristic: CBCharacteristic?       @IBOutlet private weak var ledSwitch: WKInterfaceSwitch!       override func awake(withContext context: Any?) {         centralManager = CBCentralManager(delegate: self, queue: nil)     }       @IBAction private func scanForPeripherals() {         centralManager.scanForPeripherals(withServices: nil, options: nil)     }       @IBAction private func connectPeripheral(_ sender: Any) {         if let peripheral = peripheral {             centralManager.connect(peripheral, options: nil)         }     }       @IBAction private func switchLED(_ value: Bool) {         guard let peripheral = peripheral,               let lightCharacteristic = lightCharacteristic else {             ledSwitch.setOn(!value)             return         }           var intValue: CUnsignedInt = value ? 1 : 0         print("write characteristic and \(value)")         let data = Data(bytes: &intValue, count: 1)         peripheral.writeValue(data, for: lightCharacteristic, type: .withResponse)     } }   extension InterfaceController: CBCentralManagerDelegate {       func centralManagerDidUpdateState(_ central: CBCentralManager) {         switch central.state {         case .unknown:             print("state: unknown")         case .resetting:             print("state: resetting")         case .unsupported:             print("state: unsupported")         case .unauthorized:             print("state: unauthorized")         case .poweredOff:             print("state: poweredOff")         case .poweredOn:             print("state: poweredOn")         @unknown default:             assertionFailure("想定外")         }     }       func centralManager(_ central: CBCentralManager, didDiscover peripheral: CBPeripheral,                         advertisementData: [String : Any], rssi RSSI: NSNumber) {         guard let peripheralName = peripheral.name else {             return         }         print("peripheral: \(peripheralName)")         if peripheralName == "raspberrypi" {             centralManager.stopScan()             self.peripheral = peripheral         }     }       func centralManager(_ central: CBCentralManager, didConnect peripheral: CBPeripheral) {         print("connected")         peripheral.delegate = self         peripheral.discoverServices(nil)     }       func centralManager(_ central: CBCentralManager,                         didFailToConnect peripheral: CBPeripheral, error: Error?) {         print("fail connect")     } }   extension InterfaceController: CBPeripheralDelegate {       func peripheral(_ peripheral: CBPeripheral, didDiscoverServices error: Error?) {         guard let services = peripheral.services else {             if let error = error {                 print("error: \(error)")             }             return         }           print("Found \(services.count) services: \(services)")         for service in services {             print(service.uuid.uuidString)             if service.uuid.uuidString == "FF10" {                 print("discovering characteristics")                 peripheral.discoverCharacteristics(nil, for: service)             }         }     }       func peripheral(_ peripheral: CBPeripheral,                     didDiscoverCharacteristicsFor service: CBService, error: Error?) {         guard let characteristics = service.characteristics else {             if let error = error {                 print("error in dicovercharc: \(error)")             }             return         }         print("Found \(characteristics) characteristics! : \(characteristics)")         for characteristic in characteristics {             if characteristic.uuid.uuidString == "FF11" {                 self.lightCharacteristic = characteristic             }         }     } }

起動

ラズパイ側で下記コマンドを実行する！

Watch のアプリを起動して Scan ボタン押下、Connect ボタン押下、スイッチ操作でモーターが動きます！

ちょっと工夫

よくあるモーターを動かして電気を消灯したりするやつをやってみたい！と思いやってみました。

追加で用意するもの。

• そのへんにあった 100 t ハンマーの上の部分
• 綿棒
• シャーペンの消しゴム
• セロハンテープ

完成品（ポイントは力を面から点にしているとこです）。
watch 操作 -> サーボモーターが動く -> エンター押下 -> Slackにメッセージ送信

これはもうwatchからslackに送信したと言っても過言ではない😇 pic.twitter.com/HfFj5ku2BC

— am10 (@am103141592) March 18, 2021

おわりに

これで Apple Watch（iPhone でも可）からモーターを動かすことができるようになりました。遊びの幅が広がりそうだけど何も思いつかない。。。

参考

• サーボモーターで自動エサやりマシーンを作る（前半）
• bleno で Raspberry Pi を BLE Peripheral として動かしてみる