jumeaux-numeriques/twin_serial.py

import bge # Bibliothèque Blender Game Engine (UPBGE)
import importlib
import time
import serial # Liaison série
import pyfirmata # Protocole Firmata
from serial.tools.list_ports import comports # Détection du port automatique

###############################################################################
# twin_serial.py
# @title: Gestion de la liaison série
# @project: Blender-EduTech
# @lang: fr
# @authors: Philippe Roy <philippe.roy@ac-grenoble.fr>
# @copyright: Copyright (C) 2022-2023 Philippe Roy
# @license: GNU GPL
###############################################################################

# UPBGE scene
scene = bge.logic.getCurrentScene()

# Récupérer le brochage du jumeau réel
system=importlib.import_module(scene.objects['System']['system']) # Système
pin_config = system.get_public_vars()

###############################################################################
# Liaison série
###############################################################################

##
# Recherche automatique du port
##

def autoget_port():
    # USB Vendor ID,  USB Product ID
    board_dict={'microbit' :[3368, 516],
        'uno' :[9025, 67],
        'mega' :[9025, 66]}
    for com in comports(): # Arduino Uno
        if com.vid == board_dict["uno"][0] and com.pid == board_dict["uno"][1]:
            return [com.device,"Arduino Uno"]
    for com in comports(): # Arduino Mega
        if com.vid == board_dict["mega"][0] and com.pid == board_dict["mega"][1]:
            return [com.device,"Arduino Mega"]
    return [None,""]

##
# Création de l'objet carte (protocole Firmata)
##

def board_init(port):
    try:
        return pyfirmata.Arduino(port)
    except:
       return None

##
# Création de l'objet serial (communication série)
##

def serial_init(port,speed):
    try:
        return serial.Serial(port,speed)
    except:
       return None

##
# Affiche la liste des cartes (communication série)
##

def  devices():
    for com in comports():
        print ("Name : "+str(com.name)+"\n"
               +" Device : "+str(com.device)+"\n"
               +" Hardware ID : "+str(com.hwid)+"\n"
               +" USB Vendor ID : "+str(com.vid)+"\n"
               +" USB Product ID : "+str(com.pid)+"\n"
               +" USB device location : "+str(com.location)+"\n"
               +" USB manufacturer : "+str(com.manufacturer)+"\n"
               +" USB product : "+str(com.product)+"\n"
               +" Interface-specific : "+str(com.interface))

##
# Activation de la communication avec la carte de communication (Arduino, Micro:bit)
# Vitesse  : 115200 -> 7 fps, 38400 -> 6 fps,  9600 -> 2 fps
# pyserial : baudrate=115200
# pyfirmata : baudrate=57600
##

def  serial_open():

    # UI : étape 1
    scene.objects['Twins-text']['Text'] = "Connection en cours ..."

    # Mise en place de la carte
    speed = 57600
    # speed = 115200
    # speed = 9600
    [device, board_name] =autoget_port() # Recherche automatique du port
    if device is None:
        scene.objects['System']['twins'] = False
        scene.objects['Twins-text']['Text'] = "Aucune connection disponible : jumeau réel débranché."
        print ("Jumelage : aucune connection disponible, jumeau réel débranché.")
        return None, False
    board = board_init(device)
    if board is None:
        scene.objects['System']['twins'] = False
        scene.objects['Twins-text']['Text'] = "Aucune connection disponible : port "+device+" pas prêt."
        print ("Jumelage : aucune connection disponible : port "+device+" pas prêt.")
        return None, False
    scene.objects['System']['twins'] = True
    board_it = pyfirmata.util.Iterator(board) # Itérateur pour les entrées
    board_it.start()

    # UI : étape 2
    if board =="":
        scene.objects['Twins-text']['Text'] = "Connection ouverte : "+device+" - "+str(speed)+" baud."
    else:
        scene.objects['Twins-text']['Text'] = "Connection ouverte : "+board_name+" sur "+device+" à "+str(speed)+" baud."
    time.sleep(0.1)
    return board, True

##
# Fermeture de la communication série
##

def  serial_close(board):
    if scene.objects['System']['twins']:
        scene.objects['Twins-text']['Text'] = "Connection fermée."
        board.exit() # Fermer proprement la communication avec la carte
        time.sleep(0.1)
        scene.objects['System']['twins'] = False
        print ("Jumelage : connection fermée.")

# Configuration manuelle du port
# FIXME : plus tard
def  config(port, speed):
    pass
    # global board
    # global twins_serial
    # if scene.objects['System']['twins']:
    #     serial_msg1 = "CF\n"
    #     twins_serial.write(serial_msg1.encode())
    #     tempo (1)
    #     serial_msg2 = str(speed)+"\n"
    #     twins_serial.write(serial_msg2.encode())
    #     tempo (1)
    #     serial_msg3 = str(temps_avancer)+"\n"
    #     twins_serial.write(serial_msg3.encode())
    #     tempo (1)
    #     serial_msg4 = str(temps_tourner)+"\n"
    #     twins_serial.write(serial_msg4.encode())
    #     tempo (1)
    #     serial_msg5 = "FC\n"
    #     twins_serial.write(serial_msg5.encode())

###############################################################################
# Jumeau
###############################################################################

##
# Créer une broche
##

def jumeau_get_pin(board, name, brochage):
    for pin in brochage :
        if pin ==name:
            if len (brochage[pin])==3:
                return board.get_pin(brochage[pin][0]+':'+str(brochage[pin][1])+':'+brochage[pin][2])
            else:
                print ("Jumelage : adressage incomplet de l'objet 3D :", pin)
    return None

##
# Activer le jumelage
##

def jumeau (brochage=None):
    
    # Carte
    board, board_ok =serial_open()
    if board_ok == False:
        return None
    scene.objects['System']['board']=board
    print ("Jumelage :", scene.objects['System']['board'])

    # Brochage
    if brochage is not None:
        for pin in pin_config :
            if len (pin_config[pin][1])>0:
                scene.objects[pin_config[pin][0][0]][pin_config[pin][1][0]] = jumeau_get_pin(board, pin, brochage)
        return  board
    else:
        return None

##
# Désactiver le jumelage
##

def jumeau_stop ():
    serial_close(scene.objects['System']['board'])

##
# Définir les modes d'activation du jumelage pour le systeme
##

def jumeau_mode_system (input_objs, output_objs, input_real, input_digital, output_real, output_digital):

    # Entrées numériques
    if input_digital:
        for obj in input_objs:
            scene.objects[obj]['prior']=True
    else:
        for obj in input_objs:
            scene.objects[obj]['prior']=False

    # Entrées réelles
    if input_real:
        for obj in input_objs:
            scene.objects[obj]['prior_real']=True
    else:
        for obj in input_objs:
            scene.objects[obj]['prior_real']=False

    # Sorties numériques
    if output_digital:
        for obj in output_objs:
            scene.objects[obj]['prior']=True
    else:
        for obj in output_objs:
            scene.objects[obj]['prior']=False

    # Sorties réelles
    if output_real:
        for obj in output_objs:
            scene.objects[obj]['prior_real']=True
    else:
        for obj in output_objs:
            scene.objects[obj]['prior_real']=False

##
# Définir les modes d'activation du jumelage pour un objet 3D
##

def jumeau_mode_object (name, priority_real=True, priority_digital=True):
    scene.objects[pin_config[name][0][0]]['prior_real']=priority_real
    scene.objects[pin_config[name][0][0]]['prior']=priority_digital

###############################################################################
# Message
###############################################################################

##
# Envoi d'un message vers la communication série
# FIXME : plus tard
##

# def  serie_msg(text):
#     global twins_serial
#     text2= text+"\n"
#     scene.objects['Twins-text']['Text'] = "Communication : envoi message : "+text
#     twins_serial.write(text2.encode())

##
# Mise en écoute de jumeau numérique (figeage de la scène)
# FIXME : plus tard
##

# def  twins_listen(cont):
#     global twins_serial
#     if scene.objects['System']['twins']:
#         if scene.objects['System']['twins_readline'] != "":
#             scene.objects['Twins-text']['Text'] = "Écoute de la connection figeage de la scène.... Message reçu : "+scene.objects['System']['twins_readline']
#         else:
#             scene.objects['Twins-text']['Text'] = "Écoute de la connection  figeage de la scène..."
#         if cont.sensors['Property'].positive:
#             if scene.objects['System']['twins_listen'] :
#                 serial_msg = twins_serial.readline()
#                 if serial_msg is not None:
#                     scene.objects['System']['twins_readline'] = str(serial_msg)
#                     # scene.objects['Twins-text']['Text'] = "Message reçu : "+str(serial_msg)
#                     scene.objects['System']['twins_listen'] = False

##
# Réception d'un message de la communication série
# FIXME : plus tard
##

# def  serie_rcpt():
#     # scene.objects['Twins-text']['Text'] = "Écoute de la \nconnection\n figeage de \n la scène"
#     scene.objects['System']['twins_readline'] = ""
#     scene.objects['System']['twins_listen'] = True
#     while  scene.objects['System']['twins_readline'] == "":
#         if scene.objects['System']['twins_readline'] != "":
#             break
#     # scene.objects['Twins-text']['Text'] = "Connection\nouverte :\n"+scene.objects['System']['twins_port']+"\n"+str(scene.objects['System']['twins_speed'])+" baud"
#     return scene.objects['System']['twins_readline']