Creazione di un semplice plug-in per la generazione di canali smartCORE con
Il plug-in remoto più semplice (e pratico) che si possa scrivere per smartCORE è uno che genera semplicemente i propri dati, senza influenze dall'esterno. In questo esempio creiamo un plug-in che genera tre diversi segnali di funzione: un'onda sinusoidale, un'onda quadra e un'onda a dente di sega.
Configurazione
Per prima cosa creiamo una configurazione per il modulo smartCORE che definisce i segnali da noi generati. Per emettere segnali verso smartCORE dobbiamo aggiungere alla nostra configurazione la chiave "producerChannels". Essa contiene un elenco di canali con nome, tipo di dati e unità fisica configurati. Per i nomi dei nostri segnali scegliamo "remote.test.[sine,square,sharktooth]" e come tipo di dati scegliamo "float". Qui potremmo usare anche "int" o "double", a seconda di ciò che richiedono i dati generati. Non indichiamo alcuna unità fisica, poiché questi canali generano numeri arbitrari per i nostri scopi di 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"
}
]
}
}
Implementazione
Nel nostro plug-in dobbiamo prepararci a eseguire il ciclo principale. Per prima cosa stabiliamo una connessione con smartCORE e verifichiamo che sia in esecuzione.
# 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")
Successivamente richiediamo l'elenco dei canali a cui abbiamo accesso. Dovrebbe contenere i canali che abbiamo indicato nella nostra configurazione.
# 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']
Inoltre creiamo un dict che associa i nomi dei canali ai rispettivi indici. Lo utilizziamo per inviare i nostri dati ai canali smartCORE corretti.
# map channel names to their indices
indexDict = {}
for channel in channels:
indexDict[channel['n']] = channel['i']
print(indexDict)
Ora che abbiamo messo da parte l'impalcatura di base, arriva la parte divertente. Creiamo un ciclo infinito e lo racchiudiamo in un'istruzione try-catch, così da poter uscire dal ciclo con un comando da tastiera (CTRL + C). Nel ciclo creiamo tre variabili per i nostri tre canali. Sine, che prende il seno del tempo attuale in ms, Square, che ogni due secondi commuta tra 6 e 0, e Sharktooth, che prende il modulo del tempo attuale.
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
Non appena i nostri dati sono pronti, iniziamo a creare un pacchetto WriteSamplesRequest, composto dall'header WriteSamplesRequest e da un payload con i nostri dati. Il payload è costituito da un campo "acknowledge" "a" e da un elenco di canali "c". I canali stessi sono composti da un indice "i", un valore "v" e un timestamp "t". Riempiamo questi campi con i rispettivi indici, i valori correnti e l'ora attuale.
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
Dopo aver inviato il pacchetto e ricevuto il "pacchetto Ack", dovremmo poter vedere i nostri segnali in optiCONTROL o optiCLOUD.