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
.
Signaux
Yaml
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.
Plots
yaml
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 !