Standardiser la création de nodes
Avant-propos
ld_inputs
et ld_outputs
, qui listent les sockets et les séparateurs.o_socket_in
et o_socket_out
) qui ont été modifiées et qui n'ont plus les séparateurs.show_edges()
peut donc être simplifiée : 4 vérifications au lieu de 6.CtrlScene.show_edges()
:
def show_edges(self):
""" - La clé 'edges' contient un set de descriptions d'edges (un set évite les doublons).
- Pour chaque description, on vérifie sa validité.
- Si elle est valide, on instancie l'edge, sinon on l'ignore.
- L'instanciation prend en charge l'affichage dans la scène.
(node_from, socket_out) from |------------ edge ---------------| to (node_to, socket_in)
"""
def s_id2o_node(s_id):
""" Recheche le node ayant pour clé-pkl la valeur s_id.
:param s_id: Nom de la grappe dans le pkl.
:return: objet CtrlNode ou None si non trouvé.
"""
for o_grnode in l_grnodes:
if o_grnode.s_id == s_id:
return o_grnode.o_node
return None # Non trouvé.
edges = self.o_pkl.od_pkl.read('edges', {}) # Lecture dans le pkl.
l_grnodes = self.get_grnodes() # Liste de tous les nodes présents dans la scène.
for edge in edges: # Exemple : ((0, 1), (3, 1))
""" La description de l'edge est-elle valide ? 4 vérifications. """
""" 1) Node from. """
s_id_node = f'Node{edge[0][0]}' # Exemple : 0
o_node_from = s_id2o_node(s_id_node)
if o_node_from is None:
continue
""" 2) Socket_out du node_from. """
id_socket = edge[0][1] # Exemple : 1
if not (0 <= id_socket < len(o_node_from.lo_sockets_out)):
continue
o_socket_out = o_node_from.lo_sockets_out[id_socket]
""" 3) Node to. """
s_id_node = f'Node{edge[1][0]}' # Exemple : 3
o_node_to = s_id2o_node(s_id_node)
if o_node_to is None:
continue
""" 4) Socket_in du node_to. """
id_socket = edge[1][1] # Exemple : 1
if not (0 <= id_socket < len(o_node_to.lo_sockets_in)):
continue
o_socket_in = o_node_to.lo_sockets_in[id_socket]
""" 5) Ici tout est ok : les 2 nodes existent, les sockets aussi et ce ne sont pas des séparateurs. """
CtrlEdge(self, o_socket_out, o_socket_in)
socket_in
dès la suppression de l'edge, dans remove_items()
.
gredge.o_edge.o_socket_in.to_update()
.CtrlScene.remove_items()
:
def remove_items(self):
""" Les items sont les nodes, les liens et autres widgets de la scène. """
selected = self.o_grscene.selectedItems()
l_grnodes, s_gredges = list(), set() # Le set permet d'éviter les doublons.
for sel in selected:
if sel.__class__.__name__ == 'UiNode':
l_grnodes.append(sel)
elif sel.__class__.__name__ == 'UiEdge':
s_gredges.add(sel)
""" On ajoute à s_gredges les edges accrochés aux nodes sélectionnés. """
for grnode in l_grnodes:
for o_socket in grnode.o_node.lo_sockets_in + grnode.o_node.lo_sockets_out:
if o_socket.__class__.__name__ == 'CtrlSocket':
s_gredges.update(o_socket.get_gredges())
""" On fait les comptes. """
nb_nodes = len(l_grnodes)
nb_edges = len(s_gredges)
if nb_nodes + nb_edges == 0:
return
""" Construction du message. """
s_nodes = f'{nb_nodes} node' + ('s' if nb_nodes > 1 else '') if nb_nodes > 0 else ''
s_edges = f'{nb_edges} edge' + ('s' if nb_edges > 1 else '') if nb_edges > 0 else ''
s_and = '' if nb_nodes * nb_edges == 0 else ' et '
msg = f"Vous êtes sur le point de supprimer {s_nodes}{s_and}{s_edges}.\nAcceptez-vous ?"
if ut.msg_yesno('Confirmation', msg, self):
""" Suppression des edges (et mise à jour du super-dictionnaire pkl). """
for gredge in s_gredges:
self.o_grscene.removeItem(gredge)
gredge.o_edge.o_socket_in.to_update()
self.save_edges()
""" Suppression des nodes (et mise à jour du super-dictionnaire pkl). """
for o_grnode in l_grnodes:
self.remove_node(o_grnode)
""" Enregistrement pkl. """
self.o_ur.b_action = True # Ajout dans l'historique du "Undo-Redo".
self.o_pkl.backup()
""" Affichage des paramètres de la scène. """
self.show_params()
Il en va de même lorsqu'on croise 2 edges, pour la même raison. Ajout de 2 lignes d'update pour les socket_in
permutés.
CtrlScene.cross_edges()
:
def cross_edges(self):
""" Croisement de 2 edges. Conditions :
- 2 edges sélectionnés.
- Appui sur la touche 'Tab'.
- Les extrémités d'un edge à créer appartiennent à des nodes différents.
- Si un cas est non autorisé (nb d'edges sélectionnés différent de 2 ou bouclage sur un même node) :
|_ On emet un bip (importer la classe windsound). """
lo_gredges = list()
for gredge in self.o_grscene.selectedItems():
if gredge.__class__.__name__ == 'UiEdge':
lo_gredges.append(gredge)
if len(lo_gredges) == 2:
""" Les sockets, extrémités des edges. """
o_socket_out0, o_socket_in0 = lo_gredges[0].o_edge.o_socket_out, lo_gredges[0].o_edge.o_socket_in
o_socket_out1, o_socket_in1 = lo_gredges[1].o_edge.o_socket_out, lo_gredges[1].o_edge.o_socket_in
""" Sortie reliée à une entrée d'un même node non autorisé. """
if o_socket_out0.o_node != o_socket_in1.o_node and o_socket_out1.o_node != o_socket_in0.o_node:
""" Conditions réunies => Croisement. """
""" Suppression des 2 edges sélectionnés. """
self.o_grscene.removeItem(lo_gredges[0])
self.o_grscene.removeItem(lo_gredges[1])
""" Création et sélection de 2 edges. """
CtrlEdge(self, o_socket_out0, o_socket_in1).o_gredge.setSelected(True)
CtrlEdge(self, o_socket_out1, o_socket_in0).o_gredge.setSelected(True)
o_socket_in0.to_update() # Paramètres des nodes en aval à updater.
o_socket_in1.to_update() # Paramètres des nodes en aval à updater.
""" Enregistrement + persistance. """
self.save_edges(backup=True)
else:
""" Optionnel. """
winsound.MessageBeep()
elif len(lo_gredges) > 0:
winsound.MessageBeep()
boundingRect()
des edges.UiEdge.boundingRect()
:
def boundingRect(self):
""" Méthode nécessaire, sans quoi les affichages hors champ n'ont pas lieu. """
if self.o_edge.o_socket_in.__class__.__name__ == 'QPointF':
""" Correction de bug. Il arrive parfois que :
- Les traits de bâti des edges en construction persistent à l'écran.
- Un plantage se produise, sans message d'erreur. """
return QRectF()
t_ends = self.o_edge.end_points
point_from, point_to = t_ends[0], t_ends[1] # Point de départ (P0), Point d'arrivée (P3)
x, y = min(point_from.x(), point_to.x()), min(point_from.y(), point_to.y())
w, h = max(point_from.x(), point_to.x()) - x, max(point_from.y(), point_to.y()) - y
return QRectF(x, y, w, h).normalized()
Description
Voici un graphe dont les paramètres sont ci-dessous :
La propagation des affichages (titre et provenance) est effectuée comme ci-dessous :
typ_id
, signal_ante
, signal_source
, signal_now
.dynamic_params()
.
my_params()
de la classe dérivée autant de fois qu'il y a de signaux.get_signals()
.
typ_id
, signal_ante
, signal_source
, signal_now
sont produites ...typ_id_from
, signal_ante_from
, signal_source_from
, signal_now_from
.my_signals()
.need_update()
permet de savoir, lorsqu'un paramètre de node est modifié, si les nodes en aval doivent être recalculés.
refresh()
.imports
, d_datas
et la classe Node
.__init__()
. Peut accueillir des attributs spécifiques.setup()
. Liste des court-circuits. Peut effectuer des traitements spécifiques avant et/ou après l'appel du setup()
de la classe mère.@property ld_inputs
. Liste de dictionnaires et de séparateurs.@property ld_outputs
. Liste de dictionnaires et de séparateurs.fixed_params()
. Paramètres fixes affichables dans le dockable, avant les paramètres dynamiques.my_params()
. Paramètres dynamiques affichables dans le dockable. Code appelant : CtrlNode.dynamic_params()
.my_signals()
. Description du traitement, à partir des signaux aux entrées et des paramètres (figure ci-dessus). Code appelant CtrlNode.get_signals()
.refresh()
. Propagation des modificactions de certains paramètres aux nodes en aval. Peut faire appel à la méthode need_update()
.CtrlNode.my_params()
a été modifiée :
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 {}
CtrlNode.need_update()
a été ajoutée :
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)
@property ld_inputs
car ce node n'a pas d'entrées.my_params()
car ce node n'a pas d'entrées.refresh()
, seules les modifications des cases à cocher (bool
) ont une incidence sur les nodes en aval.@property ld_inputs
car ce node n'a pas d'entrées.my_params()
car ce node n'a pas d'entrées.refresh()
, seules les modifications des cases à cocher (bool
) ont une incidence sur les nodes en aval ...
/nodes/generateurs/histos.py
:
from pc.ctrl_node import CtrlNode
d_datas = {
'name': 'Histos', # Label affiché sous l'icone.
'icon': 'histos.png', # Icone affichée.
}
class Node(CtrlNode):
def __init__(self, o_scene, s_id, pos):
super().__init__(o_scene, s_id, pos)
self.type = 'Histos'
self.setup()
def setup(self, child_file=__file__):
super().setup(child_file)
@property
def ld_outputs(self):
return [{
'label': 'Sortie',
'label_pos': (-38, -10)
}]
def fixed_params(self):
return {
'Instrument': ['EUR/USD', {'values': ["EUR/USD", "EUR/GBP", "GER30", "FRA40"]}],
'Période': ['Les derniers', {'values': ['Les premiers', 'Les derniers', 'Intervalle']}],
'Nombres(s) ou dates(s)': ['100 000',
{'tip': "Les nombres supportent des espaces pour les séparateurs de milliers."
"\nLes dates doivent être écrites comme ceci : '2016-5-14'."
"\n\tSi premiers → une date de fin."
"\n\tSi derniers → une date de début."
"\n\tSi intervalle → 2 dates, séparées par une virgule."}],
'Tick': False, # à chaque variation du signal.
'1 mn': False, # 1 minute.
'5 mn': False, # 5 minutes.
'15 mn': False, # 15 minutes.
'30 mn': False, # 30 minutes.
'1 h': False, # 1 heure.
'4 h': False, # 4 heures.
'1 j': False, # 1 jour.
'1 s': False, # 1 semaine.
'1 m': False, # 1 mois.
'Maille 1': False, # Renko, maille de 1 pip.
'Maille 2': False, # 2 pips.
'Maille 3': False, # 3 pips.
'Maille 4': False, # 4 pips.
'Maille 5': True, # 5 pips.
'Maille 7': False, # 7 pips.
'Maille 10': False, # 10 pips.
'Maille 14': False, # 14 pips.
'Maille 20': False, # 20 pips.
}
def my_signals(self, l_signals_in, num_socket_out):
l_signals = list()
for signal_key in self.fixed_params().keys():
b_signal = self.get_param(signal_key) # Booléen.
if isinstance(b_signal, bool) and b_signal:
typ_id, signal_ante, signal_now, signal_source = f'{self.type}{self.id}', '', signal_key, ''
l_signals.append((typ_id, signal_ante, signal_now, signal_source))
return l_signals
def refresh(self, l_keys):
# if self.need_update(l_keys): # Condition généraliste.
if isinstance(self.get_param(l_keys[-1]), bool): # Condition spécifique.
self.lo_sockets_out[0].to_update()
super().refresh(l_keys)
Notez le tooltip dans le champ 'Nombre(s) ou date(s)'.
@property ld_inputs
car ce node n'a pas d'entrées.my_params()
car ce node n'a pas d'entrées.refresh()
, seule la modification du type
a une incidence sur les nodes en aval./nodes/generateurs/aleatoire.py
:
from pc.ctrl_node import CtrlNode
d_datas = {
'name': 'Aléatoire',
'icon': 'aleatoire.png',
}
class Node(CtrlNode):
def __init__(self, o_scene, s_id, pos):
super().__init__(o_scene, s_id, pos)
self.type = 'Aléatoire'
self.setup()
def setup(self, child_file=__file__):
super().setup(child_file)
@property
def ld_outputs(self):
return [{
'label': 'Sortie',
'label_pos': (-38, -10)
}]
def fixed_params(self):
return {
'Type': ['Normal', {'values': ['Normal', 'Poisson', 'Binomial', 'Logistic', 'Émulation histos']}],
'Graine (-1=None)': [-1, {'tip': "Une valeur > 0 crée une série pseudo-aléatoire,"
"\nUne valeur ≤ 0 crée une série aléatoire."}],
}
def my_signals(self, l_signals_in, num_socket_out):
typ_id, signal_ante, signal_now, signal_source = f'{self.type}{self.id}', '', self.get_param('Type'), ''
return [(typ_id, signal_ante, signal_now, signal_source)]
def refresh(self, l_keys):
if l_keys[-1] == 'Type':
self.lo_sockets_out[0].to_update()
super().refresh(l_keys)
/nodes/indicateurs/moy_mobile.py
:
from pc.ctrl_node import CtrlNode
d_datas = {
'name': 'MMobile', # Label affiché sous l'icone.
'icon': 'moy_mobile.png', # Icone affichée.
}
class Node(CtrlNode):
def __init__(self, o_scene, s_id, pos):
super().__init__(o_scene, s_id, pos)
self.type = 'MM'
self.setup()
def setup(self, child_file=__file__):
self.l_short_circuits = [(0, 0)] # Entrée N°0 'court-circuitable' avec la sortie N°0.
super().setup(child_file)
@property
def ld_inputs(self):
return [{
'label': '',
'label_pos': (6, -10)
}]
@property
def ld_outputs(self):
return [{
'label': 'Sortie',
'label_pos': (-38, -10)
}]
def my_params(self, context):
return {
'Type de MA': ['SMA', {'values': ['SMA', 'EMA', 'SMMA', 'LWMA']}],
"Périodes (sep=',')": '14',
}
def my_signals(self, l_signals_in, num_socket_out):
""" Faisceau de signaux (l_signals) délivré à la sortie. """
l_signals = list()
for signals_in in l_signals_in[0]:
typ_id_from, signal_ante_from, signal_now_from, signal_source_from, \
_, signal_title, typ_id, signal_ante, signal_source = signals_in
periods = [int(p) for p in self.get_param([signal_title, "Périodes (sep=',')"]).split(',')]
for period in periods:
signal_now = f"{self.get_param([signal_title, 'Type de MA'], 'SMA')}{period}"
signal_source = self.join(signal_source_from, self.join(typ_id_from, signal_now_from), sep='\n')
typ_id, signal_ante = f'{self.type}{self.id}', self.join(signal_ante_from, signal_now_from)
l_signals.append((typ_id, signal_ante, signal_now, signal_source))
return l_signals
def refresh(self, l_keys):
""" Code spécifique. """
if self.need_update(l_keys):
self.lo_sockets_out[0].to_update() # Une seule sortie, N° 0.
super().refresh(l_keys)
/nodes/indicateurs/macd.py
:
from pc.ctrl_node import CtrlNode
d_datas = {
'name': 'MACD', # Label affiché sous l'icone.
'icon': 'macd.png', # Icone affichée.
}
class Node(CtrlNode):
def __init__(self, o_scene, s_id, pos):
super().__init__(o_scene, s_id, pos)
self.type = 'MACD'
self.setup()
def setup(self, child_file=__file__):
""" Listes des dictionnaires de base attribués aux sockets. """
self.l_short_circuits = [(0, 0), (0, 1)] # e0 -> s0 et e0 -> s1
super().setup(child_file)
@property
def ld_inputs(self):
return [
{ # Liste de 1 dictionnaire.
'label': 'Entrée',
'label_pos': (6, -10)
}
]
@property
def ld_outputs(self):
return [ # Liste de 2 dictionnaires.
{
'label': 'Sortie',
'label_pos': (-38, -10)
},
'sep', # Séparateur.
{
'label': 'Oscillateur',
'label_pos': (-59, -10)
}
]
def my_params(self, context):
""" Code appelant : CtrlNode.dynamic_params() """
return {
'Type de MA longue': ['SMA', {'values': ['SMA', 'EMA', 'SMMA', 'LWMA']}],
'Moyenne longue': 26,
'Type de MA courte': ['SMA', {'values': ['SMA', 'EMA', 'SMMA', 'LWMA']}],
'Moyenne courte': 12,
'Type de MA Signal': ['SMA', {'values': ['SMA', 'EMA', 'SMMA', 'LWMA']}],
'Signal': 9,
}
def my_signals(self, l_signals_in, num_socket_out):
""" Retourne un faisceau de signaux (l_signals) délivré à la sortie N° num_socket_out. """
l_signals = [[], []] # 2 sorties.
for signals_in in l_signals_in[0]:
typ_id_from, signal_ante_from, signal_now_from, signal_source_from, \
_, signal_title, typ_id, signal_ante, signal_source = signals_in
signal_source = self.join(signal_source_from, self.join(typ_id_from, signal_now_from), sep='\n')
typ_id, signal_ante = f'{self.type}{self.id}', self.join(signal_ante_from, signal_now_from)
t_short = self.get_param([signal_title, 'Moyenne courte'], 12) # Période de la moy courte.
t_long = self.get_param([signal_title, 'Moyenne longue'], 26) # Période de la moy longue.
t_signal = self.get_param([signal_title, 'Signal'], 9) # Période du signal.
""" Sortie 0 : 2 signaux. """
signal_now = f"Longue.{self.get_param([signal_title, 'Type de MA longue'], 'SMA')}{t_long}"
l_signals[0].append((typ_id, signal_ante, signal_now, signal_source))
signal_now = f"Courte.{self.get_param([signal_title, 'Type de MA courte'], 'SMA')}{t_short}"
l_signals[0].append((typ_id, signal_ante, signal_now, signal_source))
""" Sortie 1 : 3 signaux. """
signal_now = '(Courte-Longue)'
l_signals[1].append((typ_id, signal_ante, signal_now, signal_source))
signal_now = f"Signal.{self.get_param([signal_title, 'Type de MA Signal'], 'SMA')}{t_signal}"
l_signals[1].append((typ_id, signal_ante, signal_now, signal_source))
signal_now = f'MACD.{t_long}.{t_short}.{t_signal}'
l_signals[1].append((typ_id, signal_ante, signal_now, signal_source))
return l_signals[num_socket_out]
def refresh(self, l_keys):
""" Ex : l_keys = ["Paramètres du node 'Node0'", 'Signaux-1 : Sinus', 'Signal actif'] """
if self.need_update(l_keys):
""" Update (si l'état actif a changé) OU (si actif ET un des paramètres a changé). """
self.lo_sockets_out[0].to_update()
self.lo_sockets_out[1].to_update()
super().refresh(l_keys)
@property ld_outputs
car ce node n'a pas de sorties.my_signals()
car ce node n'a pas de sorties.refresh()
, car il n'y a pas de nodes en aval.@property ld_outputs
car ce node n'a pas de sorties.my_signals()
car ce node n'a pas de sorties.refresh()
, car il n'y a pas de nodes en aval./nodes/afficheurs/plots.py
:
from pc.ctrl_node import CtrlNode
d_datas = {
'name': 'Plots',
'icon': 'plots.png',
}
class Node(CtrlNode):
def __init__(self, o_scene, s_id, pos):
super().__init__(o_scene, s_id, pos)
self.type = 'Plots'
self.setup()
def setup(self, child_file=__file__):
super().setup(child_file)
@property
def ld_inputs(self):
return [ # Liste de 4 dictionnaires.
{
'label': 'Entrée 0',
'label_pos': (6, -10)
},
{
'label': 'Entrée 1',
'label_pos': (6, -10)
},
{
'label': 'Entrée 2',
'label_pos': (6, -10)
},
{
'label': 'Entrée 3',
'label_pos': (6, -10)
}
]
def my_params(self, context):
""" Voir documentation dans la classe-mère : CtrlNode.my_params(). """
palette = ['FFBFBF', 'FFE6BF', 'F2FFBF', 'CCFFBF', 'BFFFD9', 'BFFFFF', 'BFD9FF', 'CCBFFF', 'F2BFFF', 'FFBFE6']
legend = self.join(context[1], context[2]) # {Nom du signal}-{traitement}. Ex : 'Cosinus-SMA18'.
indx = context[4] % len(palette) # Ordre du signal dans le faisceau entrant.
return {
'Légende': legend,
'Couleur': '#' + palette[indx],
'Épaisseur': [1., {'step': .1, 'limits': (.1, 7.)}],
'Style': ['__________', {
'values': ['__________', '- - - - - - - - -', '. . . . . . . . .', '_ . _ . _ . _ .',
'_ . . _ . . _ . .']}]
}
collapsed_state()
de la classe Parameters
.show_params()
.collapsed_state()
(point N°6) dans Parameters.show_params()
:
def show_params(self, b_multi=False):
"""
Code appelant : CtrlScene.show_params() - Affichage dans le dockable 'Params'.
:param b_multi: Conditionne l'affichage d'une image.
:return: NA.
"""
""" 1 - Nettoyage de l'arbre des paramètres dans le dockable 'params'. """
self.clear_params()
""" 2 - Multi-sélection. """
if b_multi:
self.multi_selection()
return
""" 3 - Liste des paramètres 'l_params' créée à partir de od_real et od_default. """
l_params = self.dic2params()
try:
p = Parameter.create(name='params', type='group', children=l_params) # GroupParameter
except (Exception,):
print("Parameters.show_params() :\nLa liste 'l_params' n'est pas conforme.")
return
""" 4 - Mise en place de la liste des paramètres dans un widget 'ParameterTree'. """
pt = ParameterTree(showHeader=False) # ParameterTree
pt.setParameters(p, showTop=False) # Titre 'params' masqué.
""" 5 - Affichage à l'écran, en plaçant l'arbre de paramètres dans un layout propre. """
self.layout.addWidget(pt)
""" 6 - Option : Persistance de l'état des branches 'collapsées'. """
self.collapsed_state(pt)
""" 7 - Événement : change() est appelée à chaque modification. """
p.sigTreeStateChanged.connect(self.change)
Parameters.collapsed_state()
:
def collapsed_state(self, pt):
""" Ce code est exécuté à chaque affichage des paramètres d'un node ou de la scène.
- Lecture et application du dernier état mémorisé.
- Déclaration d'événements pour détecter les actions 'expanded-colapsed' produites par l'utilisateur.
|_ Ces événements appellent une fonction qui mémorise dans le pkl l'état de l'arbre des paramètres.
"""
# à coder ...
Bon coding et bon courage !
Snippets
Bonjour les codeurs !