Adaptation de la classe-mère CtrlCalcul et de certaines méthodes de CtrlNode.
☐ Fichier complet /pc/ctrl_node.py :
# Imports externes
import copy
import shutil
import os
import hashlib
import numpy as np
import pickle
# Imports internes
from functions.utils import Dictionary, DateTime, Utils
from pc.ui_node import UiNode
from pc.parameters import Parameters
from pc.ctrl_socket import CtrlSocket
from pc.ctrl_edge import CtrlEdge
class Helper:
@staticmethod
def deepcopy(reference):
return copy.deepcopy(reference)
@staticmethod
def new_od(dic=None):
return Dictionary(dic)
@staticmethod
def join(*l_variants, sep='-'):
""" Exemple d'appel 1 : y = self.join(a, b, c, d, ...)
Exemple d'appel 2 : y = self.join(*l_x) <-- Ne pas oublier '*'
Concatène les éléments de l_strings en une chaine. Les chaînes vides sont ignorées. le séparateur est sep.
:param l_variants: Nombre quelconque d'éléments de tout type.
:param sep: '-' par défaut. Peut être '' (vide) ou \n (retour à la ligne) ... ou autre.
:return: Chaîne concaténée.
Exemple l_strings = ['Paris', 'Lille', '', 'Bruxelles', 'Prague'] => return "Paris-Lille-Bruxelles-Prague"
"""
string_txt = ''
for string in l_variants:
if string != '':
string_txt += f'{sep}{string}' # Accepte tout type de variable pour string.
return string_txt[len(sep):] # Suppression du 1er sep.
class CtrlNode(Helper):
def __init__(self, o_scene, s_id, pos):
self.o_scene = o_scene
self.s_id = s_id # Clé d'entrée dans le super-dictionnaire pkl.
self.od_pkl = self.o_scene.o_pkl.od_pkl
self.o_grnode = None
self.o_params = None
self.pos = pos # pos = (x, y)
self.width = 96
self.height = 60
self.type = ''
self.ut = Utils()
self.main_key = f"Paramètres du node '{self.s_id}'"
self.b_chk = bool(self.od_pkl.read([self.s_id, 'b_chk'], True))
self.b_on = self.b_chk
self.child_file = ''
""" Sockets. """
self.lo_sockets_in = list() # Liste d'objets 'sockets in' instanciés.
self.lo_sockets_out = list() # Liste d'objets 'sockets out' instanciés.
self.default_color = '#88bbff' # Couleur par défaut des sockets de sortie.
self.l_sep_inputs = [] # Emplacement des séparateurs (après les entrées indiquées).
self.l_sep_outputs = [] # Emplacement des séparateurs (après les sorties indiquées).
""" Contenu spécifique. """
self.o_grcontent = None
""" Secousses => Court-circuits. """
self.dt = DateTime()
self.l_short_circuits = list() # Surchargé par les classes dérivées.
self.l_shortables = list()
self.b_removable = True # Les edges d'entrée peuvent être détruits en cas de court-circuit.
self.k_shake = 0 # Nombre de secousses.
self.b_shaking = False # Secousse en cours.
self.x, self.y = 0, 0 # Position antérieure (x, y), mémorisée.
self.dx, self.dy = 0, 0 # Deltas (variations) antérieurs, mémorisés.
def setup(self, child_file):
""" Code appelant : classe dérivée, on connait donc ses attributs. """
""" Paramètres. """
self.o_params = Parameters(self)
""" Hauteur automatique du node : elle dépend du nombre de sockets en entrée ou en sortie (le plus grand). """
h_title = self.d_display['geometry']['h_title']
nb_sockets_max = max(len(self.ld_inputs), len(self.ld_outputs)) # En fait : Sockets + séparateurs.
first_y = 16 * (1 + (h_title + 4) // 16) # Multiple de 16.
self.height = first_y + nb_sockets_max * 16 - 4
""" Nouveau node par drag & drop. """
self.child_file = child_file
b_new = self.pos != (0, 0) # Nouveau node (node dropé).
if b_new:
""" - Cas d'un nouveau node (dropé).
- Son centre apparaît exactement à la position du curseur de la souris.
- On mémorise son path et sa position. """
self.pos = self.pos[0] - self.width // 2, self.pos[1] - self.height // 2
path = 'nodes' + os.path.relpath(child_file).split('nodes')[1][:-3].replace(os.sep, '.')
self.od_pkl.write([self.s_id, 'path'], path)
else:
self.pos = self.od_pkl.read([self.s_id, 'position'], (0, 0))
""" Instanciation de l'UI. """
self.o_grnode = UiNode(self) # Gestion de l'UI (Interface utilisateur) : dessin couleurs, ...
if b_new:
self.o_grnode.save_pos() # Important ! à la fin (contient o_pkl.backup()).
self.o_scene.o_grscene.addItem(self.o_grnode) # Incorporation dans la scène.
""" Sockets. """
num_socket = 0
for i, d_input in enumerate(self.ld_inputs):
""" On complète les dictionnaires de base. """
if isinstance(d_input, dict): # Si ce n'est pas un dictionnaire, c'est un séparateur.
d_input['pos'] = True, i # On ajoute l'emplacement. Tuple (input: True, num_position).
d_input['id'] = self.id, num_socket # On ajoute l'identifiant. Tuple (id_node, num_socket)
self.lo_sockets_in.append(CtrlSocket(self, d_input)) # Socket instancié et ajouté à la liste.
num_socket += 1
num_socket = 0
for i, d_output in enumerate(self.ld_outputs):
""" On complète les dictionnaires de base. """
if isinstance(d_output, dict): # Si ce n'est pas un dictionnaire, c'est un séparateur.
d_output['pos'] = False, i # On ajoute l'emplacement. Tuple (input: False, num_position).
d_output['id'] = self.id, num_socket # On ajoute l'identifiant. Tuple (id_node, num_socket)
self.lo_sockets_out.append(CtrlSocket(self, d_output)) # Socket instancié et ajouté à la liste.
num_socket += 1
self.short_circuit_check()
@property
def id(self):
""" Ex : 'Node 14' renvoie 14. """
return int(self.s_id[4:])
@property
def d_display(self):
""" Attributs par défaut pouvant être surchargés dans les classes dérivées. """
return {
'geometry': {'h_title': 32, 'round': 6},
'col_pen': {'select': '#ffa637', 'hover': '#888', 'on': "#000", 'off': '#aaaa00'}, # ordre = priorité.
'col_brush': {'on': "#8888bbcc", 'off': '#cccccccc'},
'thick_pen': {'hover': 4., 'leave': 2.}
}
def get_default(self):
""" Titre du node et couleurs des 'socket_out' pour chaque type de node. """
""" Complétée par les classes dérivées avec leurs paramètres statiques et dynamiques. """
""" 1) Paramètres communs : Titre, bouton et autant de couleurs que de sorties. """
d_default = {
'Titre du node': 'Choisir un titre',
'*🔆': '' # Un astérisque devant indique qu'il s'agit d'un bouton.
}
""" 2) Ajout de paramètres fixes spécifiques au type de node, AVANT les couleurs. """
d_default.update(self.fixed0_params()) # méthode fixed_params()
""" 3) Couleurs. """
d_colors = dict()
for d_output in self.ld_outputs:
if isinstance(d_output, dict):
d_colors[d_output['label']] = self.default_color
if d_colors:
d_default['Couleurs'] = d_colors
""" 4) Ajout de paramètres fixes spécifiques au type de node, APRES les couleurs. """
d_default.update(self.fixed_params()) # méthode fixed_params()
""" 5) Ajout des paramètres dynamiques, qui dépendent des signaux entrants. """
d_default.update(self.dynamic_params) # propriété dynamic_params
""" 6) Dictionnaire complet. Paramètres : fixes communs + fixes spécifiques + dynamiques. """
return {self.main_key: d_default}
def get_param(self, l_key, v_default=None):
if self.o_params is None:
return v_default
od_params = Dictionary(self.o_params.od_params[self.main_key])
return od_params.read(l_key, v_default)
def set_checked(self, val):
self.b_chk = val
""" Update my_signals() """
for o_socket_out in self.lo_sockets_out:
o_socket_out.to_update()
""" Infrastructure. """
self.o_scene.infrastructure()
""" Enregistrement. """
self.od_pkl.write([self.s_id, 'b_chk'], self.b_chk)
self.o_scene.o_ur.b_action = True # Ajout dans l'historique du "Undo-Redo".
self.o_scene.o_pkl.backup()
""" ************************* Méthodes à surcharger ************************** """
@property
def ld_inputs(self):
return list()
@property
def ld_outputs(self):
return list()
def fixed0_params(self):
""" - Renvoie un dictionnaire de paramètres, indépendants des connexions.
- Placé AVANT les couleurs.
- Exemple : le node de type 'Labo' renvoie 'Expérience', etc. """
return {}
def fixed_params(self):
""" - Renvoie un dictionnaire de paramètres, indépendants des connexions.
- Placé APRES les couleurs.
- Exemple : le node de type 'Signaux' renvoie 'Sinus', 'Cosinus', etc. """
return {}
def bundle_params(self):
""" - Renvoie un dictionnaire de paramètres pour chaque entrée du node.
- Une entrée reçoit un faisceau de signaux (bundle). """
return {}
def my_params(self, context):
"""
:param context: Liste de 7 valeurs, pouvant être utilisée pour construire le dictionnaire de sortie :
- context[0] typ_id_from = Type + id du node en amont, producteur du signal. Ex : 'MM12'.
- context[1] signal_ante_from = Nom du signal créé par le node en amont du node producteur. Ex : 'Cosinus'.
- context[2] signal_now_from = Nom du signal créé par le node en amont. Ex : 'SMA18'.
- context[3] signal_source_from = 'Provenance du signal' affiché dans le node en amont.
- context[4] num_signal = Ordre du signal dans le faisceau entrant.
- context[5] signal_title = Titre du signal dans le dockable des paramètres.
- context[6] signal_source = 'Provenance du signal' -> texte avec retours à la ligne.
:return: Dictionnaire formaté pour les paramètres.
|_ Exemple {'Type de MA': ['SMA', {'values': ['SMA', 'EMA', 'SMMA', 'LWMA']}], "Périodes (sep=',')": '14'}
"""
return {}
def my_signals(self, l_signals_in, num_socket_out):
""" Description des signaux délivrés à la sortie N° num_socket_out.
Comment ça marche ?
- Cette fonction prend en entrée tous les signaux entrants ainsi que les paramètres pkl.
- Un traitement spécifique est décrit, qui produit plusieurs signaux.
- Ces signaux sont ensuite distribués aux différentes sorties du node.
:param l_signals_in: Signaux d'entrée. Cette liste contient autant d'éléments que d'entrées connectées.
- Chaque élément est un tuple : (typ_id_from, signal_ante_from, signal_now_from, signal_source_from, \
num_signal, signal_title, typ_id, signal_ante, signal_source)
(Les variables suffixées de '_from' proviennent du node en amont).
:param num_socket_out: N° de la sortie.
:return: Liste de signaux exclusivement destinés à la sortie N° num_socket_out.
Chaque signal est décrit par un tuple à 4 valeurs : (typ_id, signal_ante, signal_now, signal_source).
"""
raise SystemExit(self.child_file + '\nCtrlNode.my_signals() : Cette méthode doit être surchargée.')
""" *********************** Fin - Méthodes à surcharger ********************** """
@property
def dynamic_params(self):
""" Code appelant : self.get_default().
Ces paramètres sont dynamiques car ils dépendent des signaux connectés aux entrées.
- Ce node demande la liste des signaux à chaque socket d'entrée : o_socket_in.l_signals
- Cette liste est formatée de la même façon pour tous les types de node.
- C'est une liste de tuples, chaque tuple a 4 valeurs et caractérise un signal.
- Il est rappelé qu'un edge véhicule non pas un signal, mais un faisceau de signaux (un faisceau).
- Le format d'un tuple est le suivant :
- typ_id_from = Type + id du node en amont, producteur du signal. Ex : 'MM12'.
- signal_ante_from = Nom du signal créé par le node en amont du node producteur. Ex : 'Cosinus'.
- signal_now_from = Nom du signal créé par le node producteur. Ex : 'SMA18'.
- signal_source_from = Texte destiné à être affiché dans le dockable, dans 'Provenance du signal :'
- Node
|_ Socket_in
|_ Faisceau de signaux
|_ Signal
|_ Paramètre dynamique.
"""
if not self.lo_sockets_in:
""" Tant que l'initialisation n'est pas terminée, lo_sockets_in est une liste vide."""
return {}
d_params = dict()
nb_inputs = len(self.lo_sockets_in)
for k in range(nb_inputs):
d_input = self.bundle_params() # Paramètres pour cette entrée.
""" Récupération en cascade : Socket_in -> Edge -> Socket_out -> Node -> Socket_in -> etc. """
for num_signal, t_signal in enumerate(self.lo_sockets_in[k].l_signals):
typ_id_from, signal_ante_from, signal_now_from, signal_source_from = t_signal[:4] # tuple à 4 valeurs.
signal_title = self.join(typ_id_from, signal_ante_from, signal_now_from)
signal_source = self.join(signal_source_from, self.join(typ_id_from, signal_now_from), sep='\n')
""" Paramètres communs à tous les signaux. """
d_input[signal_title] = {
'$Provenance du signal :': [signal_source, {'readonly': True}],
'Signal actif': True,
}
l_args = [num_signal, signal_title, signal_source] + list(t_signal)
d_input[signal_title].update(self.my_params(l_args))
d_params[f'Entrée {k}'] = d_input
if nb_inputs == 1 and 'Entrée 0' in d_params:
""" S'il n'y a qu'une entrée, inutile d'écrire, dans le titre des paramètres, de laquelle il s'agit. """
d_params = d_params['Entrée 0']
return d_params
def get_signals(self, num_socket):
""" Code appelant : oSocket_out N° num_socket.
- Ce node crée des signaux et les fournit aux nodes en aval par l'intermédiaire des sockets et des edges.
- Il crée autant de faisceaux de signaux (vecteurs) qu'il a de sorties.
- Pour chaque sortie, la liste fournie est formatée de la même façon pour tout type de node.
- Voir détails ci-dessus, dans dynamic_params().
"""
if not (self.b_chk and self.lo_sockets_out): # Setup en cours
return []
""" Update od_params. """
self.o_params.set_params()
nb_inputs = len(self.lo_sockets_in)
l_all_signals_in = list()
for k in range(nb_inputs):
""" Le nom de l'entrée se sera pas écrit dans le titre des paramètres, s'il n'y en a qu'une. """
l_name_input = [f'Entrée {k}'] if nb_inputs > 1 else []
l_signals = list() # Liste des signaux actifs de l'entrée N° k.
""" self.lo_sockets_in[k].l_signals = Faisceau de signaux arrivant sur l'entrée k. """
for num_signal, t_signal in enumerate(self.lo_sockets_in[k].l_signals):
typ_id_from, signal_ante_from, signal_now_from, signal_source_from = t_signal[:4] # tuple à 4 valeurs.
signal_title = self.join(typ_id_from, signal_ante_from, signal_now_from)
if self.get_param(l_name_input + [signal_title, 'Signal actif'], False):
typ_id, signal_ante = f'{self.type}{self.id}', self.join(signal_ante_from, signal_now_from)
signal_source = self.join(signal_source_from, self.join(typ_id_from, signal_now_from), sep='\n')
l_signals.append(t_signal[:4] + (num_signal, signal_title, typ_id, signal_ante, signal_source))
l_all_signals_in.append(l_signals)
return self.my_signals(l_all_signals_in, num_socket)
""" ******************************* Secousses ******************************* """
def short_circuit_check(self):
""" Passage unique dans cette méthode, au démarrage. Fin programme si erreur.
- Peut-on supprimer les edges d'entrée en cas de court-circuit ?
- Oui, à condition que TOUTES les sorties existent dans self.l_short_circuits. """
l_indx_outputs = list() # Liste des index de sortie.
d_counter = dict()
for short_circuit in self.l_short_circuits: # Exemple -> self.l_short_circuits = [(0, 0), (0, 2)]
num_input = short_circuit[0]
num_output = short_circuit[1]
if num_input >= len(self.ld_inputs):
""" Fin programme. """
raise SystemExit(f"La liste des court-circuits du type de node '{self.type}' est erronée."
f"\nL'entrée N°{num_input} est hors limites."
f"\nRevoir le contenu de self.l_short_circuits dans le setup de '{self.type}'.")
if num_output >= len(self.ld_outputs):
""" Fin programme. """
raise SystemExit(f"La liste des court-circuits du type de node '{self.type}' est erronée."
f"\nLa sortie N°{num_output} est hors limites."
f"\nRevoir le contenu de self.l_short_circuits dans le setup de '{self.type}'.")
l_indx_outputs.append(num_output) # Dans cet exemple -> ( 0 , 2 )
if num_output in d_counter:
d_counter[num_output] += 1
else:
d_counter[num_output] = 1
for i, d_output in enumerate(self.ld_outputs):
if isinstance(d_output, dict):
if i not in l_indx_outputs:
""" Condition non satisfaite. """
self.b_removable = False # b_removable est à True dans __init__().
break
""" Contrainte technologique : Plusieurs entrées sur une même sortie est interdit ! """
for nb_inputs in d_counter.values():
if nb_inputs > 1:
""" Fin programme. """
raise SystemExit("Plusieurs entrées sont court-circuitables sur une même sortie."
f"\nRevoir le contenu de self.l_short_circuits dans le setup de '{self.type}'.")
def set_shortables(self):
""" Méthode appelée lorsque le bouton gauche de la souris est appuyé (sur ce node).
- Chaque type de node possède sa liste de court-circuits faisables.
- Retourne une liste de tuples.
- Chaque tuple est composé des 2 gr_edges pouvant être réunis pour n'en faire qu'un. """
self.l_shortables = list()
for short_circuit in self.l_short_circuits: # Exemple -> self.l_short_circuits = [(0, 0), (0, 2)]
lo_gredges_in = list(self.lo_sockets_in[short_circuit[0]].get_gredges())
lo_gredges_out = list(self.lo_sockets_out[short_circuit[1]].get_gredges())
if lo_gredges_in: # Cette liste contient 0 ou 1 edge.
for o_gredge_out in lo_gredges_out:
self.l_shortables.append((lo_gredges_in[0], o_gredge_out))
def shake(self):
def shortcircuit():
for to_gredge in self.l_shortables: # Tuples de gredges.
o_socket_out = to_gredge[0].o_edge.o_socket_out # From, depuis.
o_socket_in = to_gredge[1].o_edge.o_socket_in # To, jusqu'à.
""" Suppression des 2 edges du tuple. """
self.o_scene.o_grscene.removeItem(to_gredge[1]) # Edge de sortie.
if self.b_removable:
self.o_scene.o_grscene.removeItem(to_gredge[0]) # Edge d'entrée.
""" Création du nouvel edge de court-circuit. """
CtrlEdge(self.o_scene, o_socket_out, o_socket_in)
o_socket_in.to_update()
""" Persistance. """
self.o_scene.save_edges(backup=True)
""" Infrastructure. """
self.o_scene.infrastructure()
def raz_shakes():
""" Après cette raz, un nouveau court-circuit est autorisé. """
self.k_shake = 0
self.b_shaking = False
if self.b_shaking or not self.l_shortables:
return
self.dt.delay(raz_shakes, delay=1000)
x, y = self.o_grnode.pos().x(), self.o_grnode.pos().y() # Positions.
dx, dy = x - self.x, y - self.y # Déplacements (delta x, delta y).
if (dx * self.dx < 0) or (dy * self.dy < 0):
""" Le mouvement a changé de sens => incrémentation du compteur. """
self.k_shake += 1
if self.k_shake > 5: # Valeur à ajuster.
self.b_shaking = True
self.k_shake = 0
self.dt.delay(shortcircuit, delay=50)
""" Mémorisation. """
self.x, self.y = x, y
self.dx, self.dy = dx, dy
""" ***************************** Fin secousses ***************************** """
def set_colors(self, l_keys):
""" - La couleur est appliquée au socket_out concerné.
- Elle est également appliquée à tous ses éventuels edges connectés.
- A leur tour, ceux-ci propagent cette couleur à leur socket d'arrivée. """
for o_socket_out in self.lo_sockets_out:
if o_socket_out.__class__.__name__ == 'CtrlSocket':
if o_socket_out.label == l_keys[-1]:
""" Affectation de la couleur au socket_out concerné. """
o_socket_out.color = self.get_param(l_keys, '#666')
""" Récupération de tous les edgess partant de ce socket. """
l_gredges = list(o_socket_out.get_gredges())
for o_gredge in l_gredges:
""" Affectation de la même couleur à chaque edge. """
o_gredge.o_edge.color = o_socket_out.color
""" Affectation de la même couleur à chaque socket d'arrivée. """
o_gredge.o_edge.o_socket_in.color = o_socket_out.color
""" Rafraîchissement de l'affichage. """
self.o_grnode.update()
def need_update(self, l_keys):
"""
:param l_keys: Clé, dans od_params, du paramètre modifié (liste).
:return: True si les nodes en aval nécessitent d'être recalculés, False sinon.
"""
l_param_names = list(self.my_params('')) + list(self.fixed_params()) # Les paramètres du dockable.
b_actif = self.get_param([l_keys[1], 'Signal actif'], True) # True 'Signal actif' n'existe pas (générateurs).
""" Update nécessaire (si l'état actif a changé) OU (si actif ET un des paramètres a changé). """
return (l_keys[-1] == 'Signal actif') or (b_actif and l_keys[-1] in l_param_names)
def get_state(self):
""" Chaque node, selon ses spécificités, retourne un booléen de son état on/off (True/False). """
b_on = False
if self.b_chk:
for key in self.o_params.od_params.key_list():
if key[-1] == 'Signal actif':
if self.o_params.od_params.read(key, False):
b_on = True
break
return b_on
def recalculate(self, num_input):
""" - Propagation vers l'aval. Les nodes finaux sont des afficheurs ou des actionneurs.
- Après cela, les nodes finaux construisent leur dict de calcul par propagation vers l'amont. """
for o_socket_out in self.lo_sockets_out:
o_socket_out.recalculate()
def update_dcalc(self, d_calc):
if self.b_on:
""" Propagation amont. """
for o_socket_in in self.lo_sockets_in:
o_socket_in.update_dcalc(d_calc)
od_params = Dictionary(self.o_params.od_params[self.main_key])
""" Construction du dictionnaire pour les calculs, à partir du od_params. """
od_calc = Dictionary()
path = 'nodes' + os.path.relpath(self.child_file).split('nodes')[1][:-3].replace(os.sep, '.')
od_calc.write('Compute', self.type)
od_calc.write('Yaml file', self.get_yaml_files()[0])
od_calc.write('Compile from', path)
od_calc.write('Checked', self.b_chk)
for key in od_params.key_list():
if key[0] == 'Titre du node' or key[0] == 'Couleurs' or key[0] == '*🔆':
continue
if key[-1] == '$Provenance du signal :':
continue
od_calc.write(key, od_params.read(key))
""" Modification 'en place'. """
d_calc[self.s_id] = dict(od_calc)
def get_yaml_files(self):
py_file = os.path.basename(self.child_file)
seeder = self.child_file.replace(py_file, f'{py_file[:-3]}_seeder.yaml')
path_node = f"{__file__.split('pc')[0]}backups/{self.o_scene.graph_name}/node{self.id}"
extended_params = os.path.abspath(f'{path_node}/{self.id}-{py_file[:-3]}.yaml')
return extended_params, seeder
def yaml(self):
""" Méthode appelée lorsqu'on clique sur le bouton '🔆'. Lancement de l'appli associée à '.yaml'. """
extended_params, seeder_file = self.get_yaml_files()
path_node = os.path.dirname(extended_params)
os.makedirs(path_node, exist_ok=True)
if not os.path.exists(extended_params):
""" Création du fichier de paramètress étendus en yaml : copie du seeder si possible, sinon vide. """
if os.path.exists(seeder_file):
shutil.copyfile(seeder_file, extended_params)
else:
with open(extended_params, 'w'):
pass
if not os.path.exists(extended_params):
return
""" Lancement du fichier avec application associée. """
os.startfile(extended_params)
def rebuild(self, b_input=True):
""" Reconstruction du node suite à un changement dynamique du nombre d'entrées/sorties. """
""" Suppression des (sockets + edges) de ce node. """
lo_sockets = self.lo_sockets_in if b_input else self.lo_sockets_out
for o_socket in lo_sockets:
""" Suppression des éventuels edges connectés à ce socket. """
for o_gredge in list(o_socket.get_gredges()):
self.o_scene.o_grscene.removeItem(o_gredge)
""" Suppression du socket. """
self.o_scene.o_grscene.removeItem(o_socket.o_grsocket)
o_socket.o_grsocket = None # Suppression de l'objet.
""" Nettoyage de la liste des entrées/sorties. """
lo_sockets.clear()
""" Redessine le node avec sa nouvelle hauteur. """
ld_inputs, ld_outputs = self.deepcopy(self.ld_inputs), self.deepcopy(self.ld_outputs)
nb_inputs, nb_outputs = len(ld_inputs), len(ld_outputs)
h_title = self.d_display['geometry']['h_title']
first_y = 16 * (1 + (h_title + 4) // 16)
self.height = first_y + max(nb_inputs, nb_outputs) * 16 - 4 # Nouvelle hauteur.
self.o_grnode.set_outline()
""" Régénération de sockets. """
for pos, d_input in enumerate(ld_inputs):
""" On complète les dictionnaires de base. Pas de séparateurs. """
if isinstance(d_input, dict):
indx = len(self.lo_sockets_in)
d_input['pos'] = True, pos # On ajoute la pos. Tuple (input: True, num_position).
d_input['id'] = self.id, indx # On ajoute l'identifiant. Tuple (id_node, num_socket)
self.lo_sockets_in.append(CtrlSocket(self, d_input)) # Socket instancié et ajouté à la liste.
for pos, d_output in enumerate(ld_outputs):
""" On complète les dictionnaires de base. Pas de séparateurs. """
if isinstance(d_output, dict):
indx = len(self.lo_sockets_out)
d_output['pos'] = False, pos # On ajoute la pos. Tuple (output: False, num_position).
d_output['id'] = self.id, indx # On ajoute l'identifiant. Tuple (id_node, num_socket)
self.lo_sockets_out.append(CtrlSocket(self, d_output)) # Socket instancié et ajouté à la liste.
""" Régénération des edges. """
self.o_scene.show_edges()
self.o_scene.save_edges()
""" Dockable des paramètres : le nombre d'entrées a changé. """
self.o_params.set_params()
self.dt.delay(self.o_scene.show_params, 2000) # delay : Permet de continuer la saisie dans le dockable.
def refresh(self, l_keys):
if l_keys[-1] == '🔆':
self.yaml()
elif l_keys[-1] == 'Titre du node':
""" Titre du node. """
self.o_grnode.set_title()
elif l_keys[-2] == 'Couleurs':
""" Couleur des sockets et des edges. """
self.set_colors(l_keys[1:])
else:
""" Fin de refresh - Appel conditionnel à infrastructure(). """
if self.b_chk and (l_keys[0] == 'infrastructure' or l_keys[-1] == 'Signal actif'):
""" Les nodes de type générateur doivent insérer le mot 'infrastructure'
en 1ère place dans la liste l_keys[], car ils n'ont pas 'Signal actif' dans leurs paramètres. """
self.o_scene.infrastructure()
""" Tous les paramètres, autres que le titre et les couleurs, influencent les calculs. """
self.o_scene.recalculate()
class CtrlCalcul(Helper):
""" Attributs de classe. """
d_servers = dict() # Dictionnaire de classe.
od_calcs = Dictionary()
def __init__(self):
self.id = None
self.s_id = None
self.compute = None
self.o_yaml = None
self.l_mykey = list()
self.np_array = None
self.b_generator = False
self.od_descr = Dictionary()
self.od_mydic = Dictionary()
self.len_buffer = 10_000 # 300_000
self.no_params = ['Compute', 'Yaml file', 'Compile from', 'Checked']
def setup_server(self, id_server):
self.id = id_server
self.l_mykey = self.get_keys_from_idnode(id_server) # Clé dans self.od_calcs.
self.od_mydic = Dictionary(self.od_calcs.read(self.l_mykey))
self.o_yaml.yaml_file = self.od_mydic.read(['Yaml file'])
self.compute = self.od_mydic.read(['Compute'])
self.s_id = f"{self.compute}{self.id}"
self.o_yaml.update_odyaml() # Update du super-dictionnaire des paramètres (self.o_yaml.od_yaml).
@property
def dt_servers_in(self):
""" - Le dictionnaire retourné contient autant de clés que d'entrées connectées et actives.
- Chaque valeur est un tuple : (o_server, signature, edge). """
""" 1 - Récupération de la liste de tous les edges dans self.od_calcs. """
s_edges = set()
ll_keys = self.od_calcs.key_list()
for l_keys in ll_keys:
if l_keys[-1] == 'edges':
s_edges.update(self.od_calcs.read(l_keys))
""" 2 - Instance et signature des nodes-serveurs en amont connectés aux entrées. """
d_servers = dict()
for edge in list(s_edges):
if edge[1][0] == self.id:
""" edge = Edge connecté à l'entrée de self. """
o_server, signature = self.get_server(edge[0][0])
d_servers[f'e{edge[1][1]}'] = o_server, signature, edge
return d_servers
@classmethod
def get_oserver(cls, id_server, compile_from):
""" Ce code est appelé par un node-client. Il aura un serveur par entrée.
- Chaque node-serveur est associé à une instance de classe Calcul (dérivée de CtrlCalcul).
- Au premier passage concernant ce node-serveur, l'objet o_server n'existe pas.
- Dans ce cas, on le crée par compilation dynamique et on l'enregistre dans un dictionnaire.
- Le dictionnaire est un dictionnaire de classe : cls.d_servers.
- Cela permet d'éviter de créer une nouvelle instance si celle-ci existe déjà.
"""
server_sid = f"{compile_from.split('.')[-1]}{id_server}" # Clé du dictionnaire des serveurs. ex : macd3
if server_sid not in cls.d_servers:
""" Compilation dynamique. Si l'objet o_server n'existe pas, on le crée."""
d_context = {}
code = f"from {compile_from} import Calcul; o_server=Calcul()"
try:
compiled = compile(code, '<string>', 'exec')
eval(compiled, d_context)
cls.d_servers[server_sid] = d_context['o_server']
except (Exception,) as err:
print(server_sid, ': Erreur de compilation dans', compile_from, '\n', err)
return None
return cls.d_servers[server_sid] # Objet o_server.
def get_keys_from_idnode(self, id_node):
"""
@param id_node: ex : 1
@return: Renvoie la clé d'accès au node 'node_sid', dans le dictionnaire général self.od_calcs.
"""
node_sid = f'Node{id_node}' # Ex : 'Node1'
for l_key in self.od_calcs.key_list():
if l_key[1] == node_sid:
return l_key[:2] # ['calc_*', 'Node*'] Clé du node-serveur dans le dictionnaire général self.od_calcs.
def get_server(self, id_server):
l_key_node = self.get_keys_from_idnode(id_server)
compile_from = self.od_calcs.read(l_key_node + ['Compile from'])
if compile_from is None:
return None
child = self.od_calcs.read(l_key_node + ['Expérience'], '')
compile_from += f'0{child}'[-2:] if isinstance(child, int) else ''
o_server = self.get_oserver(id_server, compile_from)
if o_server is None:
return None, None
o_server.setup_server(id_server)
return o_server, o_server.get_signature()
def get_signature(self):
""" La valeur de retour est un hash de od_yaml, od_mydic, liste des signatures des nodes-serveurs. """
l_signs_in = list()
for server_in in self.dt_servers_in.values():
""" server_in = tuple (instance, signature) du node-serveur. """
l_signs_in.append(server_in[1]) # server_in[1] = signature.
""" Assemblage des constituants de la signature. """
params = (self.o_yaml.od_yaml, self.od_mydic, l_signs_in)
return f'{self.id}-{hashlib.sha256(str(params).encode("utf-8")).hexdigest()}'
def calculation(self):
""" Ici node-serveur. Première méthode appelée par le node-client en aval, avant de commencer ses calculs. """
""" 1 - Chargement des derniers datas mémorisés en *.calc (pickle) : self.od_descr et self.np_array. """
self.load_last_datas() # Super-dictionnaire et tableau numpy vides si *.calc inexistant ou erroné.
""" 2 - Création du dict. de description actualisé od_descr, aux fins de comparaison avec self.od_descr. """
od_descr = self.setup_od_descr()
""" 3 - Comparaison sélective des descriptifs now et ante. Si aucune différence, return (calculs inutiles). """
if self.now_equal_ante(od_descr):
return
self.od_descr = od_descr
""" 4 - Propagation des calculs en amont. """
for o_server, _, edge in self.dt_servers_in.values():
""" Bien que ce soit cette même méthode, ce n'est pas une récursivité, mais une propagation amont."""
o_server.calculation()
""" 5 - self.np_array vide. """
nb_cols = len([key for key in list(self.od_descr.keys()) if key.startswith(f'{self.s_id}-')])
self.np_array = np.full((self.len_buffer, nb_cols + 1), fill_value=np.nan, dtype=np.float32) # nan
self.np_array[:, 0] = 0 # col0 = status (0 ou 1)
""" 6 - Traitement des calculs. """
self.process()
""" 7 - Persistance en *.calc (pickle). """
self.create_out_calc()
def load_last_datas(self):
""" Ici, node-serveur. Récupération des derniers datas depuis son fichier out_*.calc. Les datas sont :
- Un dictionnaire descriptif des consignes : 'self.od_descr'.
- Un tableau numpy : 'self.np_array'.
- Pour certains nodes, les datas ne sont pas dans out_*.cal, mais dans une base de données.
- Le cas échéant, cette méthode sera surchargée. """
server_root = os.path.dirname(self.o_yaml.yaml_file)
os.makedirs(server_root, exist_ok=True)
out_file = os.path.normpath(f'{server_root}/out.calc')
try:
with open(out_file, 'rb') as calc:
self.od_descr, self.np_array = pickle.load(calc)
if not self.od_descr or self.np_array.size == 0 or self.np_array.shape[0] != self.len_buffer:
raise () # On force le passage par except.
except (Exception,):
""" Initialisation. """
self.od_descr, self.np_array = self.new_od(), np.empty((self.len_buffer, 0))
if not self.dt_servers_in:
self.od_mydic.write(['self_server', 'Signal actif'], True)
def setup_od_descr(self):
""" Création du super-dictionnaire descriptif des calculs : od_descr.
- Une clé pour chaque signal, actif ou pas, ayant exactement le même libellé que celui
qui est affiché dans le dockable du node-client. Exemple pour un node 'Signaux' : 'Signaux1-Sinus'
- La valeur de cette clé est un dictionnaire ayant pour clés obligatoires :
- Diverses valeurs, spécifiques au node, nécessaires aux calculs.
- num_col : Entier. Défini au dernier passage, sinon -1, destinée aux nodes-clients.
- A la racine du dictionnaire, outre les clés de signaux, il y a :
- error : Booléen. Erreur générale. Par exemple, ici, on vérifie que le tableau numpy
a bien le nombre de colonnes indiqué dans le dictionnaire de description.
- signatures : liste de signatures sha256. Une signature par entrée. Pour le node 'Signaux', 0.
Ces signatures nous permettent de savoir si les signaux entrants ont changé.
- Valeur de retour : od_descr.
"""
od_descr = Dictionary()
""" Signature = ma signature + signatures des nodes-serveurs. """
od_descr['signature'] = self.get_signature()
num_col = 0
for l_keys in self.od_mydic.key_list():
if l_keys[-1] != 'Signal actif':
continue
my_sid = self.s_id
if l_keys[0].startswith('Entrée '):
key = l_keys[:2]
my_sid += '.' + l_keys[0][-1]
else:
key = l_keys[:1]
val = self.od_mydic.read(key)
root_key = self.join(*[my_sid] + key[-1].split('-')[1:])
self.descr_signal(od_descr, val, root_key)
for key_dock in od_descr.keys():
if key_dock.startswith(root_key):
num_col += 1
od_descr.write([key_dock, 'num_col'], num_col)
if key_dock.endswith('-Original'):
try:
od_descr.write([key_dock, 'actif'], val['Original'])
except (Exception,):
pass
elif od_descr.read([key_dock, 'actif']) is None:
od_descr.write([key_dock, 'actif'], val['Signal actif'])
return od_descr
def now_equal_ante(self, od_descr):
l_no_compare = ['actif']
for key_dock in od_descr.key_list():
if key_dock[-1] in l_no_compare:
continue
if od_descr.read(key_dock) != self.od_descr.read(key_dock):
return False
else:
for o_server, _, _ in self.dt_servers_in.values():
if o_server.np_array is None:
return False
else:
return True
def descr_signal(self, od_descr, val, root_key):
""" Méthode OBLIGATOIREMENT surchargée par les classes dérivées. """
""" - Création du super-dictionnaire de description 'od_descr'.
- Les clés de 'od_descr' devront contenir les arguments nécessaires aux calculs.
- Les calculs seront effectués par process(), ou, en mode 'générateur', par get_matrix().
- {key_dock} doit avoir EXACTEMENT la même orthographe que
la clé affichée dans le dockable des paramètres du node CLIENT
- Sources EXCLUSIVES : 'self.od_mydic' et 'self.o_yaml'.
- Dynamique - Cette méthode est appelée en boucle par la classe-mère :
- Une fois par signal aux entrées (actif ou pas).
- La construction du super-dictionnaire se fait donc par couches successives.
- Cela ne signifie pas qu'il y a autant de signaux en sortie qu'en entrée !
- Ces 2 nombres sont INDÉPENDANTS !
- En effet, plusieurs signaux d'entrée peuvent contribuer pour un seul signal de sortie ...
- ... et/ou, inversement, un signal d'entrée peut concerner plusieurs signaux de sortie.
************************* IMPORTANT *************************
- Le coding se fera par itérations successives :
- Commencer à coder les calculs (process() ou get_matrix()), qui parcourra ce dictionnaire 'od_descr'.
- Si les calculs nécessitent une information absente dans 'od_descr', on revient ici pour l'ajouter.
- On prolonge le coding des calculs, puis on boucle ici ...
- ... jusqu'à la fin.
"""
raise SystemExit(f"{self.s_id} : ctrl_node.py > CtrlCalcul."
"\nLa méthode descr_signal() doit obligatoirement être surchargée.")
def process(self):
""" Méthode surchargée par les classes 'générateur' dérivées : Signaux, Aléatoire, Histos, ... """
""" Cette méthode peut également être surchargée par d'autres classes dérivées, mais attention :
- La classe dérivée doit utiliser EXCLUSIVEMENT la méthode process() OU la méthode get_matrix() :
- process() : Le tableau numpy va être affecté à 100% ...
|_ ... ce qui risque d'être chronophage pour certains calculs.
- get_matrix() : Juste une toute petite partie du tableau numpy va être remplie en live, ...
|_ ... selon les besoins des nodes-clients, grâce à l'utilisation de générateurs python.
|_ Dans ce cas, la réactivité sera excellente, même si les calculs sont complexes.
Méthode process() :
===================
- La récupération des paramètres se fait exclusivement depuis le super-dictionnaire self.od_descr.
- Le tableau numpy existe avec la bonne shape, mais est entièrement vide :
- Autant de colonnes que de signaux, plus une : la colonne de status, en position 0.
- Les valeurs sont des np.nan ...
- ... sauf la colonne 0 qui est remplie de '0'.
- Les valeurs sont produites :
- Par calculs, qui utilisent des fonctions "prêtes à l'emploi", du module 'indicators'.
- Par lecture d'une base de données.
"""
pass
def get_matrix(self, pointer, nb_lines=0):
"""
@param pointer: Pointeur "Depuis", si pointer<0 , la totalité du tableau est retournée.
@param nb_lines: "Jusqu'a" = pointer + nb_lignes.
- Pour certains nodes, les datas ne sont pas dans 'self.np_array', mais en base de données.
- Le cas échéant, cette méthode sera surchargée.
- En mode générateur-python, cette méthode est également surchargée.
- Le traitement est court, ponctuel et partiel.
- A l'issue du traitement, une toute petite partie de tableau numpy 'self.np_array' a été remplie.
- La colonne 0 contient des flags (0 ou 1), levés lorsque les lignes correspondantes ont été traitées.
- Le retour est une matrice ayant le même nombre de colonnes que np_array et {nb_lines} lignes.
"""
if self.np_array.__class__.__name__ == 'ndarray':
return self.np_array if pointer < 0 else self.np_array[pointer: pointer + nb_lines] # Tout ou slice.
def create_out_calc(self):
""" Création du fichier-résultat out_*.calc = (od_descr, liste de tableaux numpy). """
""" 1 - Contrôle - Le tableau nd_array doit avoir une shape à 2 dimensions. """
""" Correction nécessaire si le tableau est un vecteur. """
if len(self.np_array.shape) == 1: # <-- C'est un vecteur, on reshape.
self.np_array = self.np_array.reshape(self.np_array.shape[0], 1)
""" 2 - Persistance. """
server_root = os.path.dirname(self.od_mydic['Yaml file'])
out_file = os.path.normpath(f'{server_root}/out.calc')
try:
with open(out_file, 'wb') as calc:
pickle.dump((self.od_descr, self.np_array), calc, pickle.HIGHEST_PROTOCOL)
except (Exception,) as err:
print(err)
return False
return True
def add_vector(self, key_dock, num_col, vector):
if vector is None:
return
vector = vector.reshape(vector.shape[0], 1)
add_nan = np.full((self.len_buffer - vector.shape[0], 1), np.nan)
vector = np.vstack((add_nan, vector))
if num_col is None or num_col < 0:
""" Nouvelle colonne d'indice num_col (égal au nombre actuel de colonnes). """
num_col = self.np_array.shape[1]
self.np_array = np.hstack((self.np_array, vector))
else:
""" Surcharge de la colonne num_col. """
self.np_array[:, num_col:num_col + 1] = vector
self.od_descr.write([key_dock, 'num_col'], num_col)
def get_num_col(self, client_key): # client_key = MM4-Normal-Original-Original-Original
sliced_key = self.join(*client_key.split('-')[1: -1]) # Normal-Original-Original (extrémités retirées).
for l_key in self.od_descr.key_list(): # [Aléatoire1-Normal, 'num_col']
if l_key[-1] == 'num_col' and l_key[-2].endswith(sliced_key):
return self.od_descr.read(l_key)
def get_vector_in(self, client_key): # client_key = MM4-Normal-Original-Original-Original
return self.np_array[:, self.get_num_col(client_key)]
← 2 entrées par défaut, 1 sortie.
