jumeaux-numeriques/portail_coulissant/porcou_lib.py

import bge # Bibliothèque Blender Game Engine (UPBGE)
import subprocess # Multiprocessus
from twin_threading import thread_cmd_start, thread_cmd_stop, thread_cmd_end # Multithreading (multitâches)
from twin_serial import jumeau, jumeau_stop, serial_close # Liaison série
import os
import sys
import time

###############################################################################
# porcou_lib.py
# @title: Bibliothèque utilisateur du portail coulissant
# @project: Blender-EduTech
# @lang: fr
# @authors: Philippe Roy <philippe.roy@ac-grenoble.fr>
# @copyright: Copyright (C) 2020-2023 Philippe Roy
# @license: GNU GPL
###############################################################################

# Récupérer la scène UPBGE
scene = bge.logic.getCurrentScene()

# UPBGE constants
JUST_ACTIVATED = bge.logic.KX_INPUT_JUST_ACTIVATED
JUST_RELEASED = bge.logic.KX_INPUT_JUST_RELEASED
ACTIVATE = bge.logic.KX_INPUT_ACTIVE
# JUST_DEACTIVATED = bge.logic.KX_SENSOR_JUST_DEACTIVATED

###############################################################################
# Gyrophare
###############################################################################

# Ordre pour allumer le gyrophare
def gyr (order):
    scene.objects['Led']['activated']=order

###############################################################################
# Actionneurs
###############################################################################
   
# Ordre pour le moteur phase ouvrir
def mot_o (order):
    scene.objects['Moteur']['open']=order

# Ordre pour le moteur phase fermer
def mot_f (order):
    scene.objects['Moteur']['close']=order

# Ordre pour le capteur barrage IR
def ir_emet(order):
    scene.objects['Emetteur IR']['active']=order

###############################################################################
# Capteurs
###############################################################################

# Compte-rendu du capteur fin de course portail ouvert
def fdc_o ():
    if scene.objects['Microrupteur fdc ouvert']['activated'] or scene.objects['Microrupteur fdc ouvert']['activated_real']:
        return True
    else:
        return False

# Compte-rendu du capteur fin de course portail fermé
def fdc_f ():
    if scene.objects['Microrupteur fdc ferme']['activated'] or scene.objects['Microrupteur fdc ferme']['activated_real']:
        return True
    else:
        return False

# Compte-rendu du capteur barrage IR
# Absence d'obstacle -> True, présence d'obstacle -> False
def ir_recep ():
    if scene.objects['Recepteur IR']['activated'] or scene.objects['Recepteur IR']['activated_real']==False:
       return False
    else:
        return True

###############################################################################
# Boutons poussoirs
###############################################################################

# Compte-rendu du bouton pousssoir coté rue
def bp_ext ():
    if scene.objects['Bp cote rue']['activated'] or scene.objects['Bp cote rue']['activated_real']:
        return True
    else:
        return False

# Compte-rendu du bouton pousssoir coté cour
def bp_int ():
    if scene.objects['Bp cote cour']['activated'] or scene.objects['Bp cote cour']['activated_real']:
        return True
    else:
        return False

###############################################################################
# Cycle
###############################################################################

# Temporisation
def tempo (duree):
    time.sleep(duree)

# Arrêt
def stop():
    if scene.objects['System']['twins']:
        serial_close(scene.objects['System']['board'])
        time.sleep(1)
    scene.objects['System']['plot_draw']=False
    thread_cmd_stop()

# Fin naturelle
def end():
    if scene.objects['System']['twins']:
        serial_close(scene.objects['System']['board'])
        time.sleep(1)
    scene.objects['System']['plot_draw']=False
    thread_cmd_end()

def fin():
    end()

def quit():
    end()

###############################################################################
# Visualisation des données
###############################################################################

# Lancement du grapheur
def plot(data):
    # subprocess.run([sys.executable, os.path.join(os.getcwd(), "twin_plot.py")], , stdin=subprocess.PIPE) # Process bloquant

    # Terminer le processus précédent
    if scene.objects['System']['plot_proc'] is not None:
        if scene.objects['System']['plot_proc'].poll()==None:
            scene.objects['System']['plot_proc'].terminate()
    scene.objects['System']['plot_draw'] = True
    scene.objects['System']['plot_time'] = 0
    scene.objects['System']['plot_data'] = data # FIXME : conversion mémonique -> 3D 
    scene.objects['System']['plot_proc'] = subprocess.Popen([sys.executable, os.path.join(os.getcwd(), "twin_plot.py")], stdin=subprocess.PIPE, encoding = 'utf8')
    scene.objects['System']['plot']=True

# Communication avec Qt5
def plot_maj(cont):
    if cont.sensors['Plot'].positive :

        # Affichage du graphique
        if scene.objects['System']['plot_draw']:

            # Préparation du message
            # FIXME : ajouter les valeurs réelles et valeurs numériques ('activated_real')
            msg=str(scene.objects['System']['plot_time'])
            for obj in scene.objects['System']['plot_data']:
                if scene.objects[obj]['activated']:
                    msg = msg+",1"
                else:
                    msg = msg+",0"
            msg = msg+"\n"

            # Envoi (Pipe)
            if scene.objects['System']['plot_proc'].poll()==None:
                # scene.objects['System']['plot_proc'].communicate(input=time_send.encode())[0] # Communication bloquante
                scene.objects['System']['plot_proc'].stdin.write(msg)
            else:
                print ("Stop")
                scene.objects['System']['plot']=False
                scene.objects['System']['plot_draw'] =False
                scene.objects['System']['plot_proc'].terminate()
            
        # Arret de l'affichage du graphique
        else :
            if scene.objects['System']['plot_proc'].poll()==None:
                pass
            else:
                print ("Stop")
                scene.objects['System']['plot']=False
                scene.objects['System']['plot_draw'] =False
                scene.objects['System']['plot_proc'].terminate()