Création d'un plug-in simple pour générer des canaux smartCORE avec
Le plug-in distant (remote) le plus simple (et pratique) que l'on puisse écrire pour smartCORE est un plug-in qui génère simplement ses propres données, sans influence extérieure. Dans cet exemple, nous créons un plug-in qui génère trois signaux de fonction différents : une onde sinusoïdale, une onde carrée et une onde en dents de requin.
Configuration
Nous commençons par créer une configuration pour le module smartCORE, qui définit les signaux que nous générons. Pour sortir des signaux vers smartCORE, nous devons ajouter la clé "producerChannels" à notre configuration. Elle contient une liste de canaux avec nom, type de données et unité physique configurés. Pour nos noms de signaux, nous choisissons "remote.test.[sine,square,sharktooth]" et pour le type de données, nous choisissons "float". Nous pourrions aussi utiliser ici "int" ou "double", selon ce qu'exigent les données générées. Nous n'indiquons pas d'unité physique, car ces canaux génèrent des nombres quelconques à des fins de test.
{
"factory": "remote",
"module": "remote",
"config": {
"port": 61616,
"localhost": false,
"producerChannels": [{
"name": "remote.test.sine",
"dataType": "float"
},
{
"name": "remote.test.square",
"dataType": "float"
},
{
"name": "remote.test.sharktooth",
"dataType": "float"
}
]
}
}
Implémentation
Dans notre plug-in, nous devons nous préparer à exécuter notre boucle principale. Nous établissons d'abord une connexion avec smartCORE et vérifions s'il est en cours d'exécution.
# Parse command line arguments
parser = argparse.ArgumentParser(description='Generates test signals and outputs them to smartCORE channels')
parser.add_argument('--port', dest='port', default=61616, type=int, required=False)
parser.add_argument('--addr', dest='addr', default='127.0.0.1', type=str, required=False)
args = parser.parse_args()
# establish connection to smartCORE
addr = (args.addr, args.port)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
sock.settimeout(2.0)
buffer = packetHeader(CommandType.LifeSignRequest)
buffer += msgpack.packb({})
sock.sendto(buffer, addr)
# wait to receive a reply
received = sock.recv(1500)
# needs try errorhandling
header = header_from_buffer(received[:HEADER_SIZE])
# check that reply has the correct header
if header.type != CommandType.LifeSignResponse.value:
# TODO errorhandling
raise NotImplementedError
buf = BytesIO()
buf.write(received[HEADER_SIZE:])
buf.seek(0)
unpacker = msgpack.Unpacker(buf, raw=False)
# ensure smartCORE is running
for unpacked in unpacker:
if unpacked["smartcore-state"] != "Running":
raise RuntimeError("smartcore is not running")
Nous demandons ensuite la liste des canaux auxquels nous avons accès. Elle doit contenir les canaux que nous avons indiqués dans notre configuration.
# Request channel list
buffer = packetHeader(CommandType.ChannelListRequest)
buffer += msgpack.packb({})
sock.sendto(buffer, addr)
# TODO: needs better read without cutoff
received = sock.recv(1500)
# TODO: needs try errorhandling
header = header_from_buffer(received[:HEADER_SIZE])
# check that reply has correct header
if header.type != CommandType.ChannelListResponse.value:
# TODO errorhandling
raise NotImplementedError
buf = BytesIO()
buf.write(received[HEADER_SIZE:])
buf.seek(0)
unpacker = msgpack.Unpacker(buf, raw=False)
for unpacked in unpacker:
channels = unpacked['c']
Nous créons en plus un dict qui associe les noms de canaux à leurs indices. Nous l'utilisons pour envoyer nos données vers les bons canaux smartCORE.
# map channel names to their indices
indexDict = {}
for channel in channels:
indexDict[channel['n']] = channel['i']
print(indexDict)
Maintenant que le squelette de base est en place, vient la partie amusante. Nous créons une boucle infinie et l'entourons d'une instruction try-catch, afin de pouvoir quitter la boucle par une saisie au clavier (Ctrl + C). Dans la boucle, nous créons trois variables pour nos trois canaux : Sine, qui prend le sinus du temps actuel en ms, Square, qui alterne toutes les deux secondes entre 6 et 0, et Sharktooth, qui prend le modulo du temps actuel.
while True:
# Update signal generators
sine = math.sin(int(time.time() * 1_000))
square = 6 if int(time.time()) % 4 >= 2 else 0
sharktooth = time.time() % 51
Dès que nos données sont prêtes, nous commençons à construire un paquet WriteSamplesRequest, composé de l'en-tête WriteSamplesRequest et d'une charge utile contenant nos données. La charge utile se compose d'un champ « acknowledge » "a" et d'une liste de canaux "c". Les canaux eux-mêmes se composent d'un index "i", d'une valeur "v" et d'un horodatage "t". Nous renseignons ces champs avec leurs indices, les valeurs actuelles et l'heure actuelle.
buffer = packetHeader(CommandType.WriteSamplesRequest)
payload = { "a": "some_placeholder",
"c": [
{
"i": indexDict["remote.test.sine"],
"v": sine,
"t": int(time.time() * 1_000)
},
{
"i": indexDict["remote.test.square"],
"v": square,
"t": int(time.time() * 1_000)
},
{
"i": indexDict["remote.test.sharktooth"],
"v": sharktooth,
"t": int(time.time() * 1_000)
}
]
}
# print(payload)
sys.stdout.flush()
buffer += msgpack.packb(payload)
sock.sendto(buffer, addr)
# TODO: needs better read without cutoff
Une fois le paquet envoyé et le « paquet Ack » reçu en retour, nous devrions pouvoir voir nos signaux dans optiCONTROL ou optiCLOUD.