Décupler la puissance des nodes
Avant-propos
YAML.SublimeText3.
CtrlNode.shake(), dès que le compteur dépasse la valeur 5, la fonction interne shortcircuit() est appelée, mais ...shortcircuit() continue d'être appelée, et des edges supplémentaires sont créés.raz_shake() dès que le compteur dépasse 5.CtrlNode.shake() (Voir la ligne 'Correction de bug') :
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 not self.l_shortables:
return
self.dt.delay(raz_shakes, delay=100)
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.
raz_shakes() # <- Correction de bug.
shortcircuit()
""" Mémorisation. """
self.x, self.y = x, y
self.dx, self.dy = dx, dy
UiEdge.paint() :
def paint(self, painter, QStyleOptionGraphicsItem, widget=None):
""" Extrémités : points P0 et P3. """
t_ends = self.o_edge.end_points # QPointF(), QPointF()
point_from, point_to = t_ends[0], t_ends[1] # Point de départ (P0), Point d'arrivée (P3)
path = QPainterPath(point_from)
""" Points intermédiaires : P1 et P2. """
delta_x = max(40, abs(point_from.x() - point_to.x())//2)
p1x, p1y = point_from.x() + delta_x, point_from.y() # P1
p2x, p2y = point_to.x() - delta_x, point_to.y() # P2
path.cubicTo(p1x, p1y, p2x, p2y, point_to.x(), point_to.y()) # Courbe de Bézier.
painter.setPen(self.state_pen if self in self.o_edge.o_scene.o_grscene.items() else QColor(Qt.transparent))
painter.drawPath(self.path())
self.setPath(path)
plots.py > Node.my_params() :
palette = ['ff0000', 'aa007f', '55aa00', '0000ff', 'ff5500', 'aa557f', '00aa00', 'ff00ff', '55aaff', '00ff7f']
Robot/nodes.
Les sous-dossier /aleatoire, /histos, /macd, etc ont été ajoutés, les fichiers y ont été déplacés.
GroupNodes.get_datas().GroupNodes.populate_tab()./pc/dock_nodes.py :
from PyQt5.QtWidgets import QListWidget, QListWidgetItem, QListView
from PyQt5.QtCore import QSize, Qt, QByteArray, QDataStream, QIODevice, QMimeData, QPoint
from PyQt5.QtGui import QPixmap, QIcon, QDrag
import os
class GroupNodes(QListWidget):
""" Une instance par onglet du dockable des nodes. """
def __init__(self, dir_name):
"""
:param dir_name: Nom du sous-dossier sur disque dur.
"""
super().__init__()
self.dir_name = dir_name
self.path_name = ''
self.setup()
ld_datas = self.get_datas()
self.populate_tab(ld_datas)
self.set_flags()
def setup(self):
self.path_name = f"..{os.sep}nodes{os.sep}{self.dir_name}{os.sep}"
os.makedirs(self.path_name, exist_ok=True) # Création dossier. Si ce dossier existe, pas d'erreur renvoyée.
# noinspection PyUnresolvedReferences
def get_datas(self):
""" 1 - Liste de tous les fichiers .py du sous-dossier => py_files
2 - Seuls les fichiers qui possèdent un dictionnaire 'd_datas' sont candidats.
3 - Création d'une liste de ces dictionnaires => ld_datas
:return: Liste de dictionnaires de datas 'ld_datas'
"""
ld_datas = list() # ld_ = Liste de dictionnaires.
if not os.path.isdir(self.path_name):
return list() # Si le dossier n'existe pas => onglet vide.
l_folders = next(os.walk(self.path_name))[1]
py_files = list()
for folder in l_folders:
py_file = os.path.abspath(self.path_name + folder + '/' + folder + '.py')
if os.path.isfile(py_file):
py_files.append(folder)
for py_file in py_files:
code = f"from nodes.{self.dir_name}.{py_file}.{py_file} import d_datas"
try:
""" Compilation dynamique. """
eval(compile(code, '<string>', 'exec'), globals(), globals())
d_datas['path'] = f'nodes.{self.dir_name}.{py_file}.{py_file}'
d_datas['folder'] = py_file
ld_datas.append(d_datas) # Si souligné rouge -> Mark all unresolved attributes ...
except (Exception,):
continue
return ld_datas
def set_flags(self):
""" Ces flags permettent :
- De placer la liste en mode 'icones'.
- D'adapter l'affichage à la largeur du dockable (retour à la ligne).
- Dans une grille de 64 x 64 px.
- De refuser les drops.
:return: NA
"""
self.setViewMode(QListView.IconMode) # Mode 'icones'. Permet le drag.
self.setResizeMode(QListView.Adjust) # Adapte l'affichage à la largeur du dockable (Retour à la ligne).
self.setGridSize(QSize(64, 64)) # Grille de 64 x 64 px.
self.setAcceptDrops(False) # Empêche le drop (pas de doublons).
def populate_tab(self, ld_datas):
"""
Affichage des icones et de leur libellé à partir de la liste 'self.ld_datas'.
:param ld_datas: Liste de dictionnaires : un par node, contenant le nom de l'icone et de son libellé.
:return: NA
"""
for d_datas in ld_datas:
""" Chaque item contient des informations : icône, libellé, flags, datas """
name = d_datas['name']
folder = d_datas['folder']
icon = self.path_name + folder + os.sep + d_datas['icon']
item = QListWidgetItem(name, self)
icon = QPixmap(icon)
item.setIcon(QIcon(icon))
""" Ajout d'informations, pour le drag. """
item.setData(Qt.UserRole, icon) # Affichage de l'icone sous la souris pendant le drag.
item.setData(Qt.UserRole + 1, d_datas['path']) # Nécéssaire pour la compilation, lancée par le drop.
def startDrag(self, *args, **kwargs):
""" Appelée automatiquement dès qu'un drag est amorcé dans la dock-nodes.
- Les infos ajoutées dans populate_tab() seront récupérés par le destinaire du drop.
"""
mime_data = QMimeData() # Enregistre les données au format voulu (ici : 'my_robot')
item = self.currentItem()
item_data = QByteArray()
""" Datas """
icon = item.data(Qt.UserRole)
path = item.data(Qt.UserRole + 1)
data_stream = QDataStream(item_data, QIODevice.WriteOnly) # Sérialisation.
data_stream << icon # Sérialisation de l'image.
data_stream.writeQString(path)
mime_data.setData('my_robot', item_data)
drag = QDrag(self) # Prend en charge le transport des données {mime_data}.
drag.setMimeData(mime_data)
drag.setHotSpot(QPoint(icon.width() // 2, icon.height() // 2)) # Curseur centré sur l'icone.
drag.setPixmap(icon)
drag.exec_(Qt.MoveAction)
super().startDrag(*args, **kwargs)
.pkl des graphes du dossier backups, car leur mapping est devenu faux.Description
Paramètres avancés dans le dockable des paramètres de chaque node.
Bouton des paramètres avancés.
/backups/{nom du graphe}/node{id}.Signaux d'un graphe dont le nom est également 'Signaux'.

show_matplotlib.py.Signaux.Plots.Signaux. Il doit être vide puisque nous avons supprimé tous les .pkl.
CtrlNode.get_default() :
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.
}
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
""" 2) Ajout de paramètres fixes spécifiques au type de node. """
d_default.update(self.fixed_params()) # méthode fixed_params()
""" 3) Ajout des paramètres dynamiques, qui dépendent des signaux entrants. """
d_default.update(self.dynamic_params) # propriété dynamic_params
""" Dictionnaire complet. Paramètres : fixes communs + fixes spécifiques + dynamiques. """
return {self.main_key: d_default}
Parameters.dic2params() :
def dic2params(self):
r""" La valeur de retour est une liste formatée pour satisfaire aux conditions imposées par pyqtgraph.Parameter.
Cette liste contient un ou plusieurs dictionnaires, contenant à leur tour :
- Des paramètres (nom, type, valeur par défaut, valeur réelle, paramètres optionnels)
- Des 'children' de type 'group' qui contiennent à leur tour une liste au même format. <-- list()
L'ensemble des paramètres constitue donc une arborescence (donc hiérarchique) assez complexe.
Voir exemple : D:\anaconda\envs\robot\Lib\site-packages\pyqtgraph\examples\parametertree.py
"""
def _compute_val(l_keys, key, val):
""" Cette méthode produit un dictionnaire (5 clés ou plus) selon les exemples ci-après :
Exemple1 :
- Avec l_keys = ["Paramètres du graphe 'Londres'"]
key = "Nom de l'onglet"
val = "Choisir un nom"
- On obtient : dic = {
'name': "Nom de l'onglet",
'default': 'Choisir un nom', # Provient des valeurs par défaut (od_default).
'value': 'Stratégie 01', # Provient des valeurs réelles (od_real).
'l_keys': ["Paramètres du graphe 'Londres'", "Nom de l'onglet"],
'type': 'str'
}
Exemple2 :
- Avec l_keys = ["Paramètres du graphe 'Londres'", 'Épaisseurs']
key = "Traits"
val = [1.1, {'step': 0.1, 'limits': (0.1, 3.0)}]
- On obtient : dic = {
'step': 0.1,
'limits': (0.1, 3.0),
'name': 'Traits',
'default': 1.1,
'value': 1.0,
'l_keys': ["Paramètres du graphe 'Londres'", 'Épaisseurs', 'Traits'],
'type': 'float'
}
"""
dic = dict()
if isinstance(val, list):
if len(val) != 2 or not isinstance(val[1], dict):
return False
dic = val[1]
val = val[0]
if 'values' in dic:
dic['type'] = 'list' # Liste de choix.
if key[0] == '$':
dic['type'] = 'text' # Champ texte multilignes.
key = key[1:]
elif key[0] == '*':
dic['type'] = 'action' # Un champ de type action est un bouton.
key = key[1:]
dic['name'] = key
dic['default'] = val
dic['value'] = self.od_real.read([self.o_parent.s_id] + l_keys + [key], dic['default'])
dic['l_keys'] = l_keys + [key]
if 'type' not in dic: # Si ce n'est pas une liste ...
if isinstance(val, bool):
dic['type'] = 'bool'
elif isinstance(val, int):
dic['type'] = 'int'
elif isinstance(val, float):
dic['type'] = 'float'
elif isinstance(val, str):
if val and val[0] == '#':
dic['type'] = 'color'
dic['default'] = pg.mkColor(dic['default'])
dic['value'] = pg.mkColor(dic['value'])
else:
dic['type'] = 'str' # ... c'est un str.
return dic
def _build_params(default_in, list_out, l_keys):
""" Fonction récursive.
:param default_in: Sous-dictionnaire des valeurs par défaut en entrée.
:param list_out: Liste en sortie, formatée pour ParameterTree. Modifiée 'in place'.
:param l_keys: clés 'à plat' (liste) des dictionnaires od_default et od_real.
:return: list_out, modifiée 'in place'.
"""
for key, val in default_in.items():
if isinstance(val, dict):
branche = {'name': key, 'type': 'group', 'children': []}
list_out.append(branche)
l_keys.append(key) # Ajout d'une clé pour traiter la descendance.
_build_params(val, branche['children'], l_keys)
del(l_keys[-1]) # Descendance traitée => suppression de la dernière clé.
else:
""" Traitement déporté pour simplifier la compréhension du code. """
dic = _compute_val(l_keys, key, val) # Retourne un dictionnaire, sauf si erreur.
if isinstance(dic, dict):
list_out.append(dic)
l_params = list()
_build_params(copy.deepcopy(self.od_default), l_params, [])
return l_params
La prise en compte du type bouton a été ajouté à la fonction interne _compute_val().
refresh() de CtrlNode.CtrlNode.yaml() est alors appelée ...get_yaml_files() pour récupérer les fichiers yaml.refresh() par 3 méthodes.CtrlNode.refresh() par ce code:
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 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()
shutil doit être importé.
Plots est spécial : son fichier de paramètres étendus n'est pas plots.yaml.
matplotlib.yaml, pyqtgraph.yaml, bokeh.yaml, etc.get_yaml_files() doit être surchargée dans plots.py.plots.py > Node.get_yaml_files() :
def get_yaml_files(self):
py_file = os.path.basename(__file__)
seeder_name = f"{self.get_param(['Fenêtre', 'Interface']).lower()}"
seeder = __file__.replace(py_file, seeder_name) + '_seeder.yaml'
path_node = f"{os.path.dirname(__file__).split('nodes')[0]}backups/{self.o_scene.graph_name}/node{self.id}"
extended_params = os.path.abspath(f'{path_node}/{self.id}-{seeder_name}.yaml')
return extended_params, seeder
.yaml.SublimeText3.yaml est reconnu par SublimeText. Il est donc inutile d'ajouter un plugin.
Yaml à SublimeText..yaml, n'importe où dans votre disque dur.SublimeText :
[
{ "keys": ["ctrl+s"], "command": "save_all" },
{ "keys": ["ctrl+q"], "command": "save" },
{ "keys": ["ctrl+keypad_divide"], "command": "toggle_comment", "args": { "block": false } }
]
yaml avec des paramètres par défaut, ce qui permet de ne pas partir de zéro.Yaml./nodes/generateurs/signaux/signaux_seeder.yaml :
Sinus:
Amplitude: 20
Points par période: 100
Cosinus:
Amplitude: 20
Points par période: 100
Carré:
Amplitude: 20
Points par période: 100
Triangle:
Amplitude: 20
Points par période: 100
Dent de scie montante:
Amplitude: 20
Points par période: 100
Dent de scie descendante:
Amplitude: 20
Points par période: 100
/nodes/afficheurs/plots/matplotlib_seeder.yaml :
Fenêtre:
# Redémarrer pour la prise en compte des paramètres de la fenêtre.
# Styles disponibles, faire : [print(style) for style in plt.style.available]
# ['seaborn', 'Solarize_Light2', '_classic_test_patch', 'bmh', 'classic', 'dark_background', 'fast', ...]
style: seaborn
# Pour voir tous les paramètres disponibles, faire: Dictionary(rcParams).print()
# (Vérifier l'import : from matplotlib import rcParams)
rcParams:
toolbar: 'toolbar2' # 'None', 'toolbar2', 'toolmanager'
# Types de graphiques -> plt.{x} : plot, hist, pie, bar, scatter
# Exemples d'attributs : https://python.doctor/page-creer-graphiques-scientifiques-python-apprendre
# - plt.grid(True)
# - plt.text(150, 6.5, r'Danger')
# - plt.xlabel('Vitesse')
# - plt.ylabel('Temps')
# - plt.legend()
# - plt.annotate('Limite', xy=(150, 7), xytext=(165, 5.5), arrowprops={'facecolor':'black', 'shrink':0.05})
Entrée 0:
fill between:
lines: # Seuls les 2 premiers éléments de la liste sont pris en compte.
- Signaux1-Carré
- 5 # Nom de signal ou nombre.
# - Signaux1-Triangle
- Signaux1-Cosinus
color: C0 # C0 C1 C2 C3 C4 C5 C6 C7 C8 C9 - https://matplotlib.org/2.0.2/users/colors.html
Signaux1-Cosinus: # Lines2D : https://matplotlib.org/stable/api/_as_gen/matplotlib.lines.Line2D.html
alpha: .4
Entrée 1:
Entrée 2:
Entrée 3:
Entrée 4:
Entrée 5:
Entrée 6:
Entrée 7:
Ces fichiers ne seront utilisés que lorsqu'on clique le bouton des paramètres étendus d'un nouveau node.
/nodes/.../backups/ ...Vérifier :
a1*x1 + a2*x2 + ... + an*xn, soit Σai*xi pour i de 1 à n.ai sont des coefficients de pondération, et pourraient être des paramètres yaml.classe-mère pour les traitements factorisés.
yaml_parent.py > YamlParent .../nodes/ → /nodes/yaml_parent.py :
from functions.utils import Dictionary
import yaml
class YamlParent:
def __init__(self, yaml_file):
self.yaml_file = yaml_file
self.od_params = Dictionary()
self.od_yaml = Dictionary()
self.set_yaml()
def set_yaml(self, od_params=None):
""" Affectation de 2 membres : od_params et od_yaml. """
self.od_params = Dictionary() if od_params is None else od_params
try:
with open(self.yaml_file, 'r', encoding='utf-8') as yml:
self.od_yaml = Dictionary(yaml.safe_load(yml))
except FileNotFoundError:
pass
Le package PyYAML doit être installé.
signaux_yaml.py > YamlParams et matplotlib_yaml.py > YamlParams .../nodes/generateurs/signaux/signaux_yaml.py :
from nodes.yaml_parent import YamlParent
class YamlParams(YamlParent):
def __init__(self, yaml_file):
super().__init__(yaml_file)
def get_params(self, signal_name):
default_period = self.od_params.read('Points par période')
default_amplitude = self.od_params.read('Amplitude')
period = self.od_yaml.read([signal_name, 'Points par période'], default_period)
amplitude = self.od_yaml.read([signal_name, 'Amplitude'], default_amplitude)
return period, amplitude
/nodes/afficheurs/plots/matplotlib_yaml.py :
from nodes.yaml_parent import YamlParent
import numpy as np
class YamlParams(YamlParent):
def __init__(self, yaml_file):
super().__init__(yaml_file)
def get_rcparams(self):
rc_params = self.od_yaml.read(['Fenêtre', 'rcParams'])
if not isinstance(rc_params, dict):
rc_params = dict()
return rc_params
def get_style(self):
return self.od_yaml.read(['Fenêtre', 'style'], 'seaborn')
def line_params(self, o_line):
"""
:param o_line: objet Line2D, auquel on a ajouté un membre :
o_line.od = super-dictionnaire des paramètres du dockable.
:return: NA.
On a ici 2 super-dictionnaires exploitables :
- o_line.od = dockable des paramètres du PC : seulement la grappe concernant cette courbe (line).
- self.od_yaml = paramètres du fichier yaml éditable manuellement.
Code libre, à compléter au fur et à mesure des besoins du yaml.
"""
l_key_line = [o_line.od.read('Entrée'), o_line.name]
""" alpha (transparence) """
alpha = self.od_yaml.read(l_key_line + ['alpha'], None)
if alpha is not None:
o_line.set_alpha(alpha)
def subplot_params(self, o_axe):
"""
:param o_axe: objet subplot, auquel on a ajouté des membres :
o_axe.o_calc = objet 'Calcul'
o_axe.d_columns['Entrée'] = num entrée
o_axe.d_columns[{nom courbe}] = n° de colonne dans tableau numpy.
:return:
"""
name_input = f"Entrée {o_axe.d_columns['Entrée']}"
try:
o_axe.filled.remove()
except (Exception,):
pass
fill_between = self.od_yaml.read([name_input, 'fill between'], None)
if fill_between is not None:
try:
line0 = fill_between['lines'][0]
line1 = fill_between['lines'][1]
y = o_axe.o_calc.np_array
x = np.arange(y.shape[0])
color = fill_between['color']
y0 = y[:, o_axe.d_columns[line0]] if isinstance(line0, str) else line0
y1 = y[:, o_axe.d_columns[line1]] if isinstance(line1, str) else line1
o_axe.filled = o_axe.fill_between(x, y0, y1, color=color)
except (Exception,):
pass
show_matplotlib.py@property ShowMatPlotLib.l_watched()ShowMatPlotLib.setup()ShowMatPlotLib.show_anim()ShowMatPlotLib.watch()__init__() par les 5 méthodes suivantes : __init__(), setup(), show_anim(), watcher() et watch().ShowMatPlotLib.__init__() :
def __init__(self):
super().__init__()
self.graph_name = sys.argv[1]
self.node_id = int(sys.argv[2])
self.parent_pid = int(sys.argv[3])
self.figure_title = ''
self.final_path = ''
self.lod_calc = list() # lod_calc = Liste des modèles de calcul (un par entrée) affectés à ce node (des pkl).
self.o_yaml = None # Instance de la classe Yaml de matplotlib
self.l_files_ante = list()
self.od_config = None
self.od_config_ante = Dictionary()
self.lo_calc = [None for _ in range(8)] # Liste d'objets 'Calcul' des nodes directement connectés (8 entrées)
self.ut = None
self.k = 0
self.fig = None
self.ax = [None for _ in range(8)] # Liste de 8 valeurs (les axes, ou graphiques : 1 par entrée)
self.gridspec = None
self.line = None
self.style = None
self.geometry = None
self.paused = False
self.b_busy = False
self.nb_points = 350
self.delay = 50
self.listen = QTimer()
self.anim = QTimer()
self.setup()
# noinspection PyUnresolvedReferences
def setup(self):
""" Chemin complet du node final 'Plots' contenant les modèles de calculs. """
bk_path = os.path.dirname(__file__).replace('show', 'backups') # Dossier de backup.
self.final_path = os.path.normpath(os.path.join(bk_path, self.graph_name, f'node{self.node_id}')) # 'Plots'
self.ut = Utils(self.final_path)
""" Fichier de configuration, éditable manuellement en yaml. """
# [print(style) for style in plt.style.available] # Décommenter pour voir la liste des styles disponibles.
# Dictionary(rcParams).print() # Décommenter pour voir la liste des paramètres disponibles.
yaml_file = os.path.normpath(f'{self.final_path}/{self.node_id}-matplotlib.yaml')
self.o_yaml = YamlParams(yaml_file)
rc_params = self.o_yaml.get_rcparams()
for key, val in rc_params.items():
rcParams[key] = val
plt.style.use(self.o_yaml.get_style())
""" Initialisation de matplotlib.pyplot. """
self.fig = plt.figure(constrained_layout=True)
self.ax[0] = self.fig.add_subplot()
""" Attribut 'pointer' ajouté à l'objet self.ax[0]. """
self.ax[0].pointer = 0
""" Restauration de l'état de la fenêtre. """
self.ut.restore_state(self.mgr.window)
""" Surveillance des fichiers '.pkl'. """
self.watcher()
""" Événements. """
self.listen.timeout.connect(self.listener)
self.anim.timeout.connect(self.show_anim)
self.fig.canvas.mpl_connect('key_press_event', self.key_event)
""" Boucles. """
self.listen.start(1000)
self.anim.start(self.delay)
plt.show()
# noinspection PyUnresolvedReferences
def show_anim(self):
""" Une entrée (num_input = 0) donc un seul graphique occupant tout l'espace de self.fig (de la fenêtre). """
""" Ini. """
num_input = 0
o_calc = self.lo_calc[num_input]
ax = self.ax[num_input]
if o_calc is None or o_calc.b_busy or ax is None or self.b_busy:
return
self.b_busy = True # Occupé. Évite une réentrance.
""" Lecture des données à afficher : entre {pointer} et {pointer+nb_points}. """
x = np.arange(ax.pointer, ax.pointer + self.nb_points)
y = o_calc.np_array[ax.pointer: ax.pointer + self.nb_points]
if y.shape[1] == 0: # Nombre de courbes.
self.b_busy = False # Libre.
return
""" Limites x et y (abscisses et ordonnées). """
ax.set_xlim(ax.pointer, ax.pointer + self.nb_points) # Abscisses.
y_min, y_max = np.min(y), np.max(y)
y_padding = (y_max - y_min) * 0.05 # Marges top et bottom du graphique.
y_min, y_max = y_min - y_padding, y_max + y_padding
ax.set_ylim(y_min, y_max) # Ordonnées.
""" Attribution des valeurs. """
for o_line in ax.lines:
if hasattr(o_line, 'attr'):
num_col = int(o_line.attr['num_col'])
o_line.set_xdata(x)
o_line.set_ydata(y[:, num_col])
""" Avancement du pointeur. """
if len(ax.lines) > 0: # Immobile si aucune courbe n'est affichée.
ax.pointer += 1
if ax.pointer > 90_000: # Bouclage provisoire.
ax.pointer = 0
""" Affichage. """
plt.draw()
self.b_busy = False # Libre.
def watcher(self):
""" Liste des éléments à surveiller (1 dossier + fichiers .pkl + fichiers .yaml).
- Le dossier est self.final_path, dans les backups.
- Les fichiers calc_*.pkl : un par entrée active de ce node d'affichage.
- Les fichiers .yaml :
A partir des fichiers calc_*.pkl présents dans le dossier self.final_path """
l_pkl = [self.final_path+os.sep+calc for calc in os.listdir(self.final_path) if calc.startswith('calc_')]
s_yaml = set() # set() au lieu de list() pour s'affranchir des doublons.
try:
for pkl_file in l_pkl:
with open(pkl_file, 'rb') as pk:
od_calc = Dictionary(pickle.load(pk))
for key, val in od_calc.items():
if isinstance(val, dict) and key.startswith('Node'):
yaml_file = val['Yaml file']
if os.path.isfile(yaml_file):
s_yaml.add(yaml_file)
except (Exception,):
pass
graph_path = f"{os.path.dirname(__file__).replace('show', 'backups')}{os.sep}{self.graph_name}"
l_nodes = [f'{graph_path}{os.sep}{node}' for node in os.listdir(graph_path) if str(node).startswith('node')]
l_files = [graph_path] + l_nodes + l_pkl + list(s_yaml) # Concaténation de 4 listes.
if self.l_files_ante != l_files:
self.l_files_ante = self.ut.deepcopy(l_files)
self.ut.watch_file(l_files, self.watch)
self.watch(l_pkl + list(s_yaml))
def watch(self, l_files=None):
b_has_files = False
for folder in l_files:
if os.path.isfile(folder):
b_has_files = True
break
if not b_has_files:
self.watcher()
return
""" Lecture de tous les calc_*.pkl (un par entrée) et écriture de leur contenu dans une liste. """
l_pkl = [self.final_path+os.sep+calc for calc in os.listdir(self.final_path) if calc.startswith('calc_')]
lod_calc = list()
try:
for pkl_file in l_pkl:
with open(pkl_file, 'rb') as pk:
lod_calc.append(Dictionary(pickle.load(pk)))
except (Exception,) as e:
pass
""" Ajout d'un drapeau dans un node si son .yaml a été modifié. """
for file in l_files:
if file[-5:].lower() == '.yaml':
parts = file.split(os.sep)
node_name = parts[-2].capitalize()
for od_calc in lod_calc:
if node_name in od_calc:
# print(node_name)
od_calc.write([node_name, 'yaml updated'], True)
""" La méthode self.update_figure() est chargée des mises à jour des calculs et des affichages. """
if lod_calc != self.lod_calc:
self.lod_calc = self.ut.deepcopy(lod_calc)
self.update_figure()
Les modules YamlParams de Matplotlib et matplotlib.rcParams doivent être importés.
Dans l'état actuel du code, seuls 2 paramètres de la fenêtre générale de matplotlib sont opérationnels : la barre d'outils et le style.
Vérifier - Effectuer les actions suivantes.
Lancer le poste de contrôle (run main).
Cliquer sur Voir pour afficher le graphique matplotlib : par défaut, la fenêtre a une barre d'outils et le style 'seaborn'.
Cliquer sur le bouton des paramètres étendus de Plots : La fiche 0-matplotlib.yaml apparaît dans SublimeText.
Modifier le style, par exemple dark_background (Guillemets facultatifs).
Affecter à rcParams/toolbar, la valeur 'None' (Guillemets ou apostrophes obligatoires)
Enregistrer (Ctrl + s).
Comme indiqué au début du fichier yaml, pour les paramètres de la fenêtre (et seulement ceux-là), il faut relancer l'afficheur.
Par conséquent, cliquer sur Fermer puis sur Voir.
Il faut implémenter l'objet yaml dans la classe Calcul des nodes. A titre d'exemple, nous traiterons ici les nodes de type Signaux et de type Plots.
SignauxYaml dans CtrlNode.update_dcalc() CtrlNode.update_dcalc() :
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)
get_server_node().Yaml.ShowMatPlotLib.get_server_node() : ← Lignes modifiées : d_context = ... et code = ...
def get_server_node(self, od_calc):
"""
:param od_calc: Dictionnaire de l'entrée appelante.
:return: Tuple (Nom du node-serveur, Num entrée de moi-même (Node 'Plots')).
"""
s_edges = od_calc.read('edges', None) # Préfixe 's_' car les edges sont groupés dans un set().
""" Recherche du socket_out emetteur (directement connecté au socket_in de ce node 'Plots'). """
socket_from, socket_to = None, None
for edge in s_edges:
socket_to = edge[1]
if socket_to[0] == self.node_id:
socket_from = edge[0]
break
if socket_from is None or socket_to is None:
return None, None
node_sid = f'Node{socket_from[0]}'
num_input = socket_to[1]
if node_sid not in od_calc:
return None, None
if self.lo_calc[num_input] is not None:
""" Retour si l'objet o_calc existe déjà. """
return node_sid, num_input
""" Création de l'objet o_calc par compilation dynamique."""
compile_from = od_calc.read([node_sid, 'Compile from'], None)
d_context = {'yaml_file': od_calc.read([node_sid, 'Yaml file'])}
code = f'from {compile_from} import Calcul; o_calcul = Calcul(yaml_file)'
try:
code = compile(code, '<string>', 'exec')
eval(code, d_context)
o_calcul = d_context['o_calcul']
self.lo_calc[num_input] = o_calcul
return node_sid, num_input
except (Exception,):
print(f'Erreur de compilation dans {compile_from}')
return None, num_input
Calcul des nodes doit être modifiée, pour traiter ce fichier yaml./nodes/generateurs/signaux/signaux.py > Calcul :
class Calcul(CtrlCalcul):
""" ********** Le code ci-dessous ne concerne pas le poste de contrôle, mais seulement les calculs. ********** """
def __init__(self, yaml_file):
super().__init__()
self.default_period = 0
self.default_amplitude = 0
self.len_buffer = 0
self.o_yaml = YamlParams(yaml_file)
def create_values(self, key, od_params):
if self.b_busy:
return True
self.b_busy = True
""" Grappe de od_params, concernant uniquement ce node (signaux). """
od_self = self.new_od(od_params.read(key))
self.o_yaml.set_yaml(od_self) # Update du super-dictionnaire des paramètres (self.o_yaml.od_yaml).
self.default_period = od_self.read('Points par période')
self.default_amplitude = od_self.read('Amplitude')
self.len_buffer = od_self.read('Taille buffer')
""" Mise à jour des paramètres : dockable + yaml. """
self.o_yaml.set_yaml(od_self)
""" Liste de tous les signaux, actifs ou non : [Sinus, Cosinus, ...]. """
l_sig_names = [signal_name for signal_name, val in od_self.items() if isinstance(val, bool)]
for signal_name in l_sig_names:
if signal_name == 'Checked':
self.od_config.write(['Checked'], od_self.read('Checked', False))
else:
self.od_config.write([signal_name, 'actif'], od_self.read(signal_name, False)) # True ou False.
if od_self.read('Sinus'): # Case à cocher du dockable.
self.d_signals['Sinus'] = self.get_sinus_wave(od_self)
if od_self.read('Cosinus'):
self.d_signals['Cosinus'] = self.get_cosinus_wave(od_self)
if od_self.read('Carré'):
pass # à coder ...
if od_self.read('Triangle'):
pass # à coder ...
if od_self.read('Dent de scie montante'):
pass # à coder ...
if od_self.read('Dent de scie descendante'):
pass # à coder ...
""" Création, si inexistant, du tableau de sortie. """
if self.np_array is None:
self.np_array = np.empty((self.len_buffer, 0))
for key, val in self.d_signals.items():
if not self.od_config.read([key, 'actif'], False):
continue # Non traité si inactif.
num_col = self.od_config.read([key, 'num_col'], None)
if num_col is None:
num_col = self.np_array.shape[1]
self.np_array = np.hstack((self.np_array, val))
else:
""" Surcharge de la colonne num_col. """
self.np_array[:, num_col:num_col+1] = val
self.od_config.write([key, 'num_col'], num_col)
self.b_busy = False
return True
def get_sinus_wave(self, od_self):
period, amplitude = self.o_yaml.get_params('Sinus')
nb_iter = self.len_buffer // (2 * period)
x = np.linspace(0, 2 * np.pi, period + 1)[:-1]
y = np.reshape(amplitude * np.sin(x), (period, 1))
y = np.concatenate((y, y) * nb_iter)
np_array = np.full((self.len_buffer, 1), np.nan)
np_array[:y.shape[0]] = y # Les dernières valeurs sont des nan.
return np_array
def get_cosinus_wave(self, od_self):
period, amplitude = self.o_yaml.get_params('Cosinus')
nb_iter = self.len_buffer // (2 * period)
x = np.linspace(0, 2 * np.pi, period + 1)[:-1]
y = np.reshape(amplitude * np.cos(x), (period, 1))
y = np.concatenate((y, y) * nb_iter)
np_array = np.full((self.len_buffer, 1), np.nan)
np_array[:y.shape[0]] = y # Les dernières valeurs sont des nan.
return np_array
La classe YamlParams doit être importée.
Plotsyaml soient prises en compte, il faut ajouter le chemin du fichier Yaml dans le dictionnaire de calculs du node Plots. On ajoute donc la clé Yaml file au dictionnaire.plots.py > Node.recalculate() :
def recalculate(self, num_input):
d_calc = {'edges': set()}
o_socket_in = self.lo_sockets_in[num_input]
o_socket_in.update_dcalc(d_calc)
od_params = self.new_od(self.o_params.od_params[self.main_key])
od_calc = self.new_od()
od_calc.write('Compute', self.type)
od_calc.write('Yaml file', self.get_yaml_files()[0])
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 :' or key[-1] == "Nombre d'entrées":
continue
b_add = True
if key[0].startswith('Entrée'):
t_socket = self.id, int(key[0][7:])
b_add = False
for edge in d_calc['edges']:
if t_socket in edge:
b_add = True
break
if b_add:
od_calc.write(key, od_params.read(key))
d_calc[self.s_id] = dict(od_calc)
""" On n'enregistre le pkl que s'il a été modifié. """
if d_calc != self.ld_calc[num_input]:
fic_pkl = os.path.abspath(f'{self.path_node}/calc_{num_input}.pkl')
try:
with open(fic_pkl, 'wb') as pk:
pickle.dump(d_calc, pk, pickle.HIGHEST_PROTOCOL)
self.ld_calc[num_input] = self.deepcopy(d_calc) # Mémorisation.
except (Exception,):
pass
La clé Yaml file a été ajoutée.
Plots.Plots provoquera des erreurs.yaml ne donne un exemple d'application que sur l'entrée 0.alpha = 0. à 1.YamlParams de matplotlib traite déjà cet exemple d'application.
update_figure() de la classe ShowMatPlotLib.
... bon coding et bon courage !
Vérification

Snippets
Bonjour les codeurs !