pico/micro-python support

[J0EK3R]

@elclaudio
I'm developing this project on windows and raspberry pi 4.
micro-python-support is also needed.

Here is a link to a micropython-pi-pico simulator
Here is a link to bluetooth advertising with pico-micro
Here is a link to a simple python example of an ibeacon
Here is a link to a Bluetooth library for Raspberry Pi
Here is a link to Linux Beacon Zone
Here is a link to btmgmt - Basic Python Wrapper for the BlueZ Management API
Here is a link to python-btsocket - Python library to interact with the Bluez Bluetooth Management API
Ultimate Python Cheat Sheet: Practical Python For Everyday Tasks
How to create a Python library

[elclaudio]

very nice work you've done J0EKER !

in the test "C2: halfspeed backwards" rawdata = mk6_0.SetChannel(1, -0.5) 6d b6 43 cf 7e 8f 47 11 84 66 59 f8 11 7a aa 34 67 4a 3d f9 15 16 17 18, channel B is running in the good direction (backward) but Channel A is also activated at the same time. I tested others channels C,D,E,F they are not activated.

Since there is no hcitool in a microcontroller, the trick would be to send thoses values directly to the BT controller. This is apparently possible with (at least) an arduino nano ESP32 with their C Library https://www.arduino.cc/reference/en/libraries/arduinoble/ however, the library work only in C, not in python or micropython

I think I will order both an arduino ESP32 nano and a pico w to make further tests. In the mean time, I received yesterday my second mould king 6 hub so I could use it to test for exemple dual hub connection

[J0EK3R]

The value for each channel [0..5] is stored and sent.
You have to call the Methode "stop" to set all channels to 0.
...and there is a copy-and-paste-bug :(

[elclaudio]

I detected 2 issues:

• stop command is stopping all the motors instead of a particular channel

• starting a motor on another channel stop the motor that is currently running on another channel

I've made a little script to play interactively:
run it with : python3 -i thisscript.py

#!/usr/bin/python3

try:
  from MouldKing.MouldKingCrypt import MouldKingCrypt
  from MouldKing.MouldKing_6 import MouldKing_6
  from MouldKing.MouldKingDevice import MouldKingDevice
except ImportError:
  from MouldKingCrypt import MouldKingCrypt
  from MouldKing_6 import MouldKing_6

import subprocess
import platform
import time

hcitool_path = '/usr/bin/hcitool'

def _getChannelId(channel):
    switch={
    'A': 0,
    'B': 1,
    'C': 2,
    'D': 3,
    'E': 4,
    'F': 5,
    }
    return switch.get(channel,"")

def _getHubId(deviceId):
    if deviceId == 0:
        hub = MouldKing_6(0)
    elif deviceId == 1:
        hub = MouldKing_6(1)
    elif deviceId == 2:
        hub = MouldKing_6(2)
    else:
        return
    return hub

def mkconnect(deviceId=0):
    hub = _getHubId(deviceId)
    rawdata = hub.Connect()
    hcitool_args1 = hcitool_path + ' -i hci0 cmd 08 0008 ' + MouldKingCrypt.CreateTelegramForHCITool(MouldKingDevice.ManufacturerID, rawdata) + ' &> /dev/null'
    hcitool_args2 = hcitool_path + ' -i hci0 cmd 0x08 0x0006 A0 00 A0 00 03 00 00 00 00 00 00 00 00 07 00' + ' &> /dev/null'
    hcitool_args3 = hcitool_path + ' -i hci0 cmd 0x08 0x000a 01' + ' &> /dev/null'

    if platform.system() == 'Linux':
        subprocess.run(hcitool_args1, shell=True, executable="/bin/bash")
        subprocess.run(hcitool_args2, shell=True, executable="/bin/bash")
        subprocess.run(hcitool_args3, shell=True, executable="/bin/bash")
    elif platform.system() == 'Windows':
        print('Connect command :')
        print(hcitool_args1)
        print(hcitool_args2)
        print(hcitool_args3)
    else:
        print('Unsupported OS')

    return

def mkstop(deviceId=0):
    hub = _getHubId(deviceId)
    rawdata = hub.Stop()
    hcitool_args = hcitool_path + ' -i hci0 cmd 08 0008 ' + MouldKingCrypt.CreateTelegramForHCITool(MouldKingDevice.ManufacturerID, rawdata) + ' &> /dev/null'

    if platform.system() == 'Linux':
        subprocess.run(hcitool_args, shell=True, executable="/bin/bash")
    elif platform.system() == 'Windows':
        print('Control command selected:')
        print(hcitool_args)
    else:
        print('Unsupported OS')

    return        

def mkcontrol(deviceId=0, channel=0, powerAndDirection=1):
    hub = _getHubId(deviceId)
    rawdata = hub.SetChannel(channel, powerAndDirection)
    hcitool_args = hcitool_path + ' -i hci0 cmd 08 0008 ' + MouldKingCrypt.CreateTelegramForHCITool(MouldKingDevice.ManufacturerID, rawdata) + ' &> /dev/null'

    if platform.system() == 'Linux':
        subprocess.run(hcitool_args, shell=True, executable="/bin/bash")
    elif platform.system() == 'Windows':
        print('Control command selected:')
        print(hcitool_args)
    else:
        print('Unsupported OS')

    return

print("\nReady to execute commands\n")
print("For connecting: mkconnect(hubId) ex: mkconnect(0) or mkconnect(1) for the second hub\n")
print(" Available commands: mkconnect(hubId)\n mkstop(hubId)\n mkcontrol(deviceId, channel, powerAndDirection)\n")
print("ex: mkcontrol(0, 0, 0.5) ; mkcontrol(0, 'B', -1)\n the minus sign - indicate reverse motor direction\n")

[J0EK3R]

@elclaudio
I have added your testscript as consoletest.py to the project and did an important fix within the script:

def _getHubId(deviceId):
    if deviceId == 0:
        hub = MouldKing_6(0)
    elif deviceId == 1:
        hub = MouldKing_6(1)
    elif deviceId == 2:
        hub = MouldKing_6(2)
    else:
        return
    return hub

Everytime you call _getHubId a new instance of MouldKing_6 is created - and the internal stored values of the new instance are zero...
The result is that you can't run more than one motor at the same time.

Here is my fix:
three instances are created and stored in the fields - _getHubId returns one of the instances

hub0 = MouldKing_6(0)
hub1 = MouldKing_6(1)
hub2 = MouldKing_6(2)

def _getHubId(deviceId):
    if deviceId == 0:
        return hub0
    elif deviceId == 1:
        return hub1
    elif deviceId == 2:
        return hub2
    else:
        raise Exception("deviceId 0..2")

[elclaudio]

@elclaudio I have added your testscript as consoletest.py to the project and did an important fix within the script:

def _getHubId(deviceId):
    if deviceId == 0:
        hub = MouldKing_6(0)
    elif deviceId == 1:
        hub = MouldKing_6(1)
    elif deviceId == 2:
        hub = MouldKing_6(2)
    else:
        return
    return hub

Everytime you call _getHubId a new instance of MouldKing_6 is created - and the internal stored values of the new instance are zero... The result is that you can't run more than one motor at the same time.

Here is my fix: three instances are created and stored in the fields - _getHubId returns one of the instances

hub0 = MouldKing_6(0)
hub1 = MouldKing_6(1)
hub2 = MouldKing_6(2)

def _getHubId(deviceId):
    if deviceId == 0:
        return hub0
    elif deviceId == 1:
        return hub1
    elif deviceId == 2:
        return hub2
    else:
        raise Exception("deviceId 0..2")

Indeed, that was it ! I confirm it works now, good job

I've updated my files with your modifications.
I don't know how much v6 hub do you got, but I've tried with my second hub and noticed something strange :

When 2 hub are switched on and waiting for connect telegram, if trying to connect to hub1 ( MouldKing_6(1)) it connect to ALL hub (actually tested with 2 hub)

But when sending a command to hub 1 ( mkcontrol(1,0,1) ) there is no reaction, for ex, hub 1, channel 0, fullspeed, forwards :
data send: 25 02 01 02 1B FF F0 FF 6D B6 43 CF 7E 8F 47 11 87 66 59 47 D1 7A AA 34 67 49 1F F1 15 16 17 18

When choosing hub 0 instead, same command and channel ( mkcontrol(0,0,1) data send = 25 02 01 02 1B FF F0 FF 6D B6 43 CF 7E 8F 47 11 84 66 59 47 D1 7A AA 34 67 4A ED B7 15 16 17 18), the motors of both hub 0 and 1 run simultaneously. In facts, it seems that both hubs react when deviceId is 0 but not when another device is selected

ps: I updated the consoletest.py and added automated test function and also added debug option to functions that show the raw command line executed

#!/usr/bin/python

# to run:  python -i consoletest.py

try:
  from MouldKing.MouldKingCrypt import MouldKingCrypt
  from MouldKing.MouldKing_6 import MouldKing_6
except ImportError:
  from MouldKingCrypt import MouldKingCrypt
  from MouldKing_6 import MouldKing_6

import subprocess
import platform
import time
import numpy as np
import sys

hcitool_path = '/usr/bin/hcitool'
hub0 = MouldKing_6(0)
hub1 = MouldKing_6(1)
hub2 = MouldKing_6(2)

def _getChannelId(channel):
    switch={
    'A': 0,
    'B': 1,
    'C': 2,
    'D': 3,
    'E': 4,
    'F': 5,
    }
    return switch.get(channel,"")

def _getHubId(deviceId):
    if deviceId == 0:
        return hub0
    elif deviceId == 1:
        return hub1
    elif deviceId == 2:
        return hub2
    else:
        raise Exception("deviceId 0..2")

def _automate(deviceId, channel):
    userinput = input("\nDo you want to test channel "+ str(channel) +" ? enter y/n\n")

    if (userinput != str("y")):
        return

    print("HUB: "+  str(deviceId) +", FORWARD : Power ramp up from 0 to 100% on channel :" + str(channel))
    for i in np.arange(0, 1.1, 0.1):
        print("Power : " + str(i) + "\n")
        mkcontrol(deviceId,channel,i)
        time.sleep(1)

    mkstop(deviceId)

    print("HUB: "+  str(deviceId) +", REVERSE: Power ramp up from 0 to 100% on channel :" + str(channel))
    for i in np.arange(-0, -1.1, -0.1):
        print("Power : " + str(i) + "\n")
        mkcontrol(deviceId,channel,i)
        time.sleep(1)

    mkstop(deviceId)

def mkconnect(deviceId=0, debug=False):
    hub = _getHubId(deviceId)
    rawdata = hub.Connect()
    hcitool_args1 = hcitool_path + ' -i hci0 cmd 08 0008 ' + MouldKingCrypt.CreateTelegramForHCITool(MouldKing_6.ManufacturerID, rawdata)
    hcitool_args2 = hcitool_path + ' -i hci0 cmd 0x08 0x0006 A0 00 A0 00 03 00 00 00 00 00 00 00 00 07 00'
    hcitool_args3 = hcitool_path + ' -i hci0 cmd 0x08 0x000a 01'

    if platform.system() == 'Linux':
        subprocess.run(hcitool_args1 + ' &> /dev/null', shell=True, executable="/bin/bash")
        subprocess.run(hcitool_args2 + ' &> /dev/null', shell=True, executable="/bin/bash")
        subprocess.run(hcitool_args3 + ' &> /dev/null', shell=True, executable="/bin/bash")
    else:
        print('Unsupported OS or debug mode, this is the command(s) that should be run :')

    if (debug or platform.system() != 'Linux'):
        print(str(hcitool_args1) + '\n' + str(hcitool_args2) + '\n' + str(hcitool_args3) + '\n')

    return

def mkstop(deviceId=0, debug=False):
    hub = _getHubId(deviceId)
    rawdata = hub.Stop()
    hcitool_args = hcitool_path + ' -i hci0 cmd 08 0008 ' + MouldKingCrypt.CreateTelegramForHCITool(MouldKing_6.ManufacturerID, rawdata)

    if platform.system() == 'Linux':
        subprocess.run(hcitool_args + ' &> /dev/null', shell=True, executable="/bin/bash")
    else:
        print('Unsupported OS or debug mode, this is the command that should be run :')

    if (debug or platform.system() != 'Linux'):
        print(str(hcitool_args) + '\n')

    return        

def mkcontrol(deviceId=0, channel=0, powerAndDirection=1, debug=False):
    hub = _getHubId(deviceId)
    rawdata = hub.SetChannel(channel, powerAndDirection)
    hcitool_args = hcitool_path + ' -i hci0 cmd 08 0008 ' + MouldKingCrypt.CreateTelegramForHCITool(MouldKing_6.ManufacturerID, rawdata)

    if platform.system() == 'Linux':
        subprocess.run(hcitool_args + ' &> /dev/null', shell=True, executable="/bin/bash")
    else:
        print('Unsupported OS or debug mode, this is the command that should be run :')

    if (debug or platform.system() != 'Linux'):
        print(str(hcitool_args) + '\n')

    return

def test_hub(hubId=0):
    print("HUB "+ str(hubId) +" connecting")
    mkconnect(hubId)
    time.sleep(1)

    for index in range(6):
        _automate(hubId, index) # start channel 0 = A
        print("Channel change requested")
        time.sleep(1)

print("\nReady to execute commands\n")
print("If run on windows, commands are shown but not executed (hcitool dependency)\n")
print("For connecting: mkconnect(hubId) ex: mkconnect(0) or mkconnect(1) for the second hub\n")
print("Available commands: \n mkconnect(hubId, debug) ***Initiate hub control***\n mkstop(hubId, debug) ***Stop ALL motors***\n mkcontrol(deviceId, channel, powerAndDirection, debug) ***Control a specific hub, channel, power and motor direction***\n test_hub(hubId) ***run automated tests on each channels***\n")
print("ex: test_hub(0), mkcontrol(0, 0, 0.5); mkcontrol(0, 1, -1, True)\n the minus sign - indicate reverse motor direction\n")

[J0EK3R]

@elclaudio
I did a little fix and some testing...
...and added the new function mkbtstop() to consoletest.py.

Look at the updated readme...
(Spoiler: to swicth the hub's address press the button...)

Currently we cannot address several hubs (I have three MK6.0) the same time because we didn't figuered out yet how to advertise different telegrams the same time.

[elclaudio]

@elclaudio I did a little fix and some testing... ...and added the new function mkbtstop() to consoletest.py.

Look at the updated readme... (Spoiler: to swicth the hub's address press the button...)

Currently we cannot address several hubs (I have three MK6.0) the same time because we didn't figuered out yet how to advertise different telegrams the same time.

that's funny, great find :-)
You can update the consoletest.py with my latest addition from my previous post

I've ordered both arduino nano ESP32 and a pico W for further testing. Both can run µPython or C program. I'm confident that this could work wyth thoses devices because micropython have a BT library that can send advertisements

https://docs.micropython.org/en/latest/library/bluetooth.html

Look at the function : BLE.gap_advertise(interval_us, adv_data=None, *, resp_data=None, connectable=True)

Strangely enough, so far, I haven't found a library for regular python that have such possibility, but it doesn't mean it doesn't exists

[J0EK3R]

@elclaudio
I have merged your version of consoletest.py into the repo and did some cleanup.

Remember:
You don't need to set a hubId when calling mkconnect() - there is only a device-independent telegram to switch hubs in bluetooth mode.

And here is a new understanding:

For connecting:
Power MK6.0 Hubs on - Led is flashing green/blue.
Call mkconnect() to send the bluetooth connect telegram.
All hubs powered-on will switch to bluetooth mode: Leds are flashing blue.
By short-pressing the button on MK6.0 Hubs you can choose the hubId.
hubId=0 - one Led flash
hubId=1 - two Led flashs
hubId=2 - three Led flashs

[J0EK3R]

@elclaudio
I have spend a lot of time trying to send Bluetooth Advertisements from my Radpberry Pi using some Python-Bluetooth-Frameworks that are accepted by the MK6.0 Hubs - without success.

We have two problems:
To address three MK6.0 Hubs we have to sent three different Advertisement telegrams "the same time" (the telegrams have to alternate within the hubs timeout) - so python libs have to offer the possibility to register, change and unregister Advertisements.
On the Raspberry Pi there are such libs and I think I could handle that...
The second problem is that the f* MK hubs are accepting only telegrams matching a special pattern: within the Advertisement they need first "Flags" and second the "Manufacturer specific data". The libs I've tried are generating telegrams with first "Manufacturer Data" and second the "Flags"...

[elclaudio]

@elclaudio I have spend a lot of time trying to send Bluetooth Advertisements from my Radpberry Pi using some Python-Bluetooth-Frameworks that are accepted by the MK6.0 Hubs - without success.

We have two problems: To address three MK6.0 Hubs we have to sent three different Advertisement telegrams "the same time" (the telegrams have to alternate within the hubs timeout) - so python libs have to offer the possibility to register, change and unregister Advertisements. On the Raspberry Pi there are such libs and I think I could handle that... The second problem is that the f* MK hubs are accepting only telegrams matching a special pattern: within the Advertisement they need first "Flags" and second the "Manufacturer specific data". The libs I've tried are generating telegrams with first "Manufacturer Data" and second the "Flags"...

thank for your feedback and for the time passed on your tests. I also spend many time on this, but, it's fun and addictive ;-)

the micropython bt lib seems to be what we need, there also aioble which is an "object-oriented, asyncio-based wrapper for MicroPython's bluetooth API." So much possibilities here.

I received today my pico W and my arduino nano ESP32. Since, they just arrived, I just have enough time to test standard bt advertisement on the pico (which works). I don't know if you have some early code to share for mkconnect yet, or may just use the ouput of crypt()

It seem's just a bytearray with hex data is required, the intervall is send as BLE.gap_advertise function's 1st parameter

bt advertising works with this simple example :

import bluetooth

def main():
    BLE = bluetooth.BLE()
    BLE.active(True)
	
	#Advertise
    name = bytes("ESP32BLEmPy", 'UTF-8')
    adv_data = bytearray(b'\x02\x01\x02') + bytearray((len(name) + 1, 0x09)) + name
    BLE.gap_advertise(100, adv_data)

edit: I successfully sent connect telegram from pico W without hcitool ! more later

edit 2:
It seem's that I need to add 2 byte (0x00, 0xFF) before appending the result of the crypt function in order to connect. I've seen thoses 2 bytes when connecting with the mould king app on android.

I added this function in a rush but you know better what should be done:

    @staticmethod
    def CreateTelegramForPicoW(rawDataArray):
        cryptedArray = MouldKingCrypt.Crypt(rawDataArray)
        cryptedArrayLen = len(cryptedArray)

        resultArray = bytearray(2 + cryptedArrayLen)
        resultArray[0] = 0x00
        resultArray[1] = 0xFF
        for index in range(cryptedArrayLen):
            resultArray[index + 2] = cryptedArray[index]

        return resultArray

Also, for BT advertising, do you thinks that we should send connect advertisements all the time ? It appear that it's not necessary when the hub is linked.

Stopping the advertisement is just a matter of (tested ok):

def mkStopAdvertise():
    ble.gap_advertise(None)

Since this works out of the box on a raspi pico w (I have yet to test it on the arduino ESP32 nano), for now, I think we should concentrate efforts on thoses platforms, it's also better suited for such a project (a microcontroller is enough)

2Vid_20240524_013056-1.mp4

[J0EK3R]

Wow, great :)
We are going step by step...

Also, for BT advertising, do you thinks that we should send connect advertisements all the time ?

No. The connect telegram ist only needed to switch the hubs to bt-mode.

I don't know if you have some early code to share for mkconnect yet, or may just use the ouput of crypt()

The output of crypt() is what has to be advertised, that's all. But the used bt-lib has to generate the correct telegram with the expected order of the parts. I mentioned the "Flags" part that has to be the first part. Second is the crypted data.

Btw.: I'm using Wireshark with a special USB-BT-Dongle to sniff the BT-Traffic - it's much easier with this tool to compare the telegrams...

It seem's that I need to add 2 byte (0x00, 0xFF) before appending the result of the crypt function in order to connect. I've seen thoses 2 bytes when connecting with the mould king app on android.

Possibility this are the "Flags" - I have to test this.

And: you are able to modify the files of the repo on your own and start a pull request so I can just merge your changes...

[J0EK3R]

@elclaudio
Now I have a pico w with Micropython installed.
Can you give me your code?

[elclaudio]

@elclaudio Now I have a pico w with Micropython installed. Can you give me your code?

Hi JOEK3R

I didn't had time to explore much further since my last message, I was very busy on other stuff.

This is what I tested :

added method in MouldKingCrypt.py :

    @staticmethod
    def CreateTelegramForPicoW(rawDataArray):
        cryptedArray = MouldKingCrypt.Crypt(rawDataArray)
        cryptedArrayLen = len(cryptedArray)

        resultArray = bytearray(2 + cryptedArrayLen)
        resultArray[0] = 0x00
        resultArray[1] = 0xFF
        for index in range(cryptedArrayLen):
            resultArray[index + 2] = cryptedArray[index]

        return resultArray

added in consoletest.py:

try:
    import bluetooth
except ImportError as err:
    print(err)

# Instantiate bluetooth object
ble = bluetooth.BLE()

# Activate bluetooth
ble.active(True)
	
	
def mkconnect2():
    hub = _getHubId(0)
    rawdata = hub.Connect()
    btdata = MouldKingCrypt.CreateTelegramForPicoW(rawData)
    btcrypteddata = bytearray(b'\x02\x01\x02') + bytearray((len(btdata) + 1, 0xFF)) + btdata
    ble.gap_advertise(100, btcrypteddata)
    return
	
def mkStopAdvertise():
    ble.gap_advertise(None)
    return

[J0EK3R]

Hi @elclaudio :)
Thank you for the code...

...but now I know why bluetooth is not working on my pico... ...I don't have the pico w - it's just a pico :|
I ordered a new one.

In the meantime I have reworked all ;) - it's more object-oriented now

just look inside consoletest.py

# uncomment to choose advertiser
from Advertiser.AdvertiserHCITool import AdvertiserHCITool as Advertiser
#from Advertiser.AdvertiserBluez import AdvertiserBluez as Advertiser
#from Advertiser.AdvertiserDBus import AdvertiserDBus as Advertiser
#from Advertiser.AdvertiserMicroPython import AdvertiserMicroPython as Advertiser

On your pico w you should uncomment AdvertiserMicroPython.

And AdvertiserHCITool now supports multiple connected hubs.
You can do things like

mkcontrol(0, 0, 0.5)  # hub0
mkcontrol(1, 0, -0.5) # hub1

...now my daily computertime is over ;)