Modifications depuis le PC appliquées en direct
Avant-propos
Étudiez ou révisez Numpy : nombreux cours et tutos sur internet.
1
.
'NoneType' object is not iterable
, dans CtrlScene.recalculate()
.self.o_pkl.od_pkl.read('edges')
qui devrait retourner une liste d'edges, mais qui retourne None
.None
.
""" 3 - Suppression des fichiers ... """
:l_edges = [edge[1] for edge in self.o_pkl.od_pkl.read('edges') if edge[1][0] in l_ends]
CtrlScene.recalculate()
:
l_edges = [edge[1] for edge in self.o_pkl.od_pkl.read('edges', []) if edge[1][0] in l_ends]
plots.py > Node.recalculate()
, supprimer le bloc marqué """ MAP à supprimer. """
.CtrlNode.__init__()
, remplacer cette ligne ...self.b_chk = self.od_pkl.read([self.s_id, 'b_chk'], True)
self.b_chk = bool(self.od_pkl.read([self.s_id, 'b_chk'], True))
Description
générateur de signaux
connecté à un node de type Plots
, à 1 entrée.Signaux
, nous pouvons :
Plots
, nous pouvons :
watch()
.update_figure()
: live-update du titre de la fenêtre.Calcul
.update_figure()
: update des calculs.update_figure()
: live-update de l'affichage.Signaux
dans les backups
.node0
dans ce dossier Signaux
.calc_0.pkl
dans ce dossier node0
.
0
et 1
soient répercutées en live.calc_0.pkl
.show_matplotlib.py
, le code d'appel afin que le graphe appelé soit Signaux
, et que le node Plots soit d'ID 0
.show_matplotlib.py
:
if __name__ == '__main__':
sys.argv.append('Signaux') # Nom du graphe.
sys.argv.append('0') # Plots id. (La ligne de commande ne doit contenir que du str).
sys.argv.append('-1') # Gestion de la fermeture automatique. Poste de contrôle PID : -1 en local.
mpl = ShowMatPlotLib()
show_matplotlib.py
: tout fonctionne comme avant.deepcopy
à la fin de la classe Utils
./functions/utils/Utils.deepcopy()
:
# **********************************************************
# Divers *
# **********************************************************
@staticmethod
def deepcopy(reference):
return copy.deepcopy(reference)
Importer la classe copy
.
Le dictionnaire pour les calculs a précédemment été créé par le poste de contrôle.
Or nous aurons besoin de 2 clés supplémentaires pour les calculs : Compile from
et Checked
.
La méthode CtrlNode.update_dcalc()
doit donc être modifiée :
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(['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':
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)
/node0
) est modifié, supprimé ou créé.
calc_*.pkl
pourront être modifiés de l'extérieur : par le poste de contrôle (PC) ou par d'autres programmes externes.calc_*.pkl
pourront être supprimés (par exemple par le PC lorsqu'une entrée de Plots est supprimée ou invalidée).calc_*.pkl
pourront être créés (par exemple par le PC lorsqu'une entrée de Plots est ajoutée ou validée).watch()
est automatiquement appelée, afin de mettre à jour les calculs et les affichages.watch()
:
.pkl
de calculs modifiés ou créés, et seulement ceux-la, produisant pour chacun un super-dictionnaire de calculs.lod_calc
de ces super-dictionnaires de calcul.Plots
.update_figure()
qui actualise les calculs et les graphiques, à partir de lod_calc
.update_figure()
utilise cette liste lod_calc
des super-dictionnaires. Pour chacun d'eux, il va :
o_calc
, instance de la classe Calcul
du node producteur de signaux (ici, du node Signaux-1
)Plots
(ici, pour l'instant, il n'y en a qu'un).o_calc
de faire ou d'actualiser les calculs, en appelant sa méthode o_calc.create_values()
.o_calc.create_value()
crée un tableau numpy
avec autant de colonnes que de signaux à produire.
create_value()
, mais leur code est spécifique.Signaux-1
et doit pouvoir fournir des séries de valeurs, qui, une fois affichées, représenteront une sinusoïde, un signal carré, triangualire, etc.moyenne mobile
ou un MACD
, les calculs seront évidemment différents.show_anim()
: Enfin, en marge de tout cela, nous avons la méthode show_admin()
que nous connaissons déjà, et qui est lancée de manière récurrente par timer.
numpy
produit par l'objet o_calc
: il s'agit du tableau o_calc.np_array
.Utils
.self.watcher
dans /functions/utils.py > Utils.__init__()
:
self.watcher = None
watch_file()
.Utils.watch_files()
:
# **********************************************************
# Watching - Surveillance de fichier *
# **********************************************************
def watch_file(self, l_files, callback):
"""
QFileSystemWatcher peut renvoyer 2 signaux coup sur coup au lieu d'un seul :
|_ Le premier parce que le fichier à surveiller a été supprimé.
|_ Le deuxième parce qu'il a été re-créé.
Pour éviter ce défaut et un double callback, on met en place un délai de 100 ms ...
|_ ... qui prend en charge le callback.
:param l_files: Liste des fichiers et dossiers à surveiller (chemins complets)
:param callback: procédure à exécuter à chaque shoot du watcher.
:return: callback(<fic>) avec le fichier modifié en argument.
"""
def file_changed(fic):
if fic in status[0]:
return
status[0].append(fic) # current_file
self.o_dt.delay(send_file, 100)
def send_file(): # Fin du monostable.
callback(status[0])
status[0] = list() # Reset.
status = [list()] # [[fichiers modifiés]] <- Liste dans liste.
self.watcher = QFileSystemWatcher(l_files)
self.watcher.directoryChanged.connect(file_changed)
self.watcher.fileChanged.connect(file_changed)
# |_ self.watcher = Mémorisation du QFileSystemWatcher obligatoire à cause du garbage collector de Python.
QFileSystemWatcher
doit être importé depuis PyQt5.Qtcore.
watcher
: Il a lieu au démarrage de l'appli show_matplotlib.py
, depuis le setup()
.""" Surveillance des fichiers yaml et pkl. """
de ShowMatPlotLib.setup()
:
""" Surveillance des fichiers '.pkl'. """
self.watch()
La méthode self.watch()
n'existe pas, nous la créerons au point suivant.
@property
self.l_watched
.ShowMatPlotLib.l_watched
:
@property
def l_watched(self):
return [self.final_path+os.sep+calc for calc in os.listdir(self.final_path) if calc.startswith('calc_')]
final_path
devient un attribut, self.final_path
. Comme elle a déjà été créée et utilisée au début du setup()
, dans le bloc """ Chemin complet du node final ... """
, il va falloir corriger ce code et la déclarer dans __init__()
.""" Chemin complet du node final ... """
de ShowMatPlotLib.setup()
:
""" 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)
__init__()
, le voici complet.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 = dict() # lod_calc = Liste des modèles de calcul (un par entrée) affectés à ce node (des pkl).
self.lod_calc_ante = dict()
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()
Dictionary
doit être importé.
Oui mais ... comme expliqué ci-dessus au point 2 - Principe de fonctionnement, la surveillance des fichiers appelle automatiquement la méthode watch()
.
Celle-ci n'existe pas, créons-la.
watch()
watch()
est bien appelée lorsqu'un fichier du dossier /backups/Signaux/node0
est modifié, créé ou supprimé.ShowMatPlotLib.watch()
:
def watch(self, l_files=None):
if l_files is None:
self.ut.watch_file([self.final_path] + self.l_watched, self.watch) # Rafraîchissement de la surveillance.
else:
print('Watch', l_files)
main.py > Run
.Voir
, mais depuis Pycharm
, lancer un deuxième process : show_matplotlib.py > Run
.Run
, vérifier que show_matplotlib
est sélectionné, afin de voir les messages du print()
.Run
de Pycharm
..pkl
modifié, le dictionnaire qu'il contient est converti en super-dictionnaire puis ajouté à une liste.print()
de vérification et en ajoutant ce code.ShowMatPlotLib.watch()
devient :
def watch(self, l_files=None):
""" 1er passage lors de l'initialisation. Passages suivants lors de modifications de fichiers '.pkl'. """
l_files = self.l_watched if l_files is None else l_files
if len(l_files) < 2:
""" Les fichiers .pkl ont été supprimés, l_files ne contient plus que le chemin du dossier.
Ceci se produit lors de modifications de l'infrastructure : edges ou nodes supprimés, ajoutés, etc. """
l_files = self.l_watched
""" Lorsque des fichiers surveillés on été supprimes, ils ne sont plus surveillés lorsqu'ils sont recréés.
D'où la nécessité de rafraîchier la surveillance. """
self.ut.watch_file([self.final_path] + self.l_watched, self.watch) # Rafraîchissement de la surveillance.
self.lod_calc = list() # lod_calc = Liste des modèles de calcul (od = ordered dic) affectés à ce node.
for file in l_files:
""" Paramètres de calcul des graphiques et de leur fenêtre. """
if os.path.basename(file)[:5] + file[-4:] == 'calc_.pkl':
""" Paramètres de calculs. """
try:
with open(file, 'rb') as pk:
self.lod_calc.append(Dictionary(pickle.load(pk)))
except (Exception,) as e:
""" Entrée devenue inactive => Le fichier pkl a été supprimé. """
pass
""" La méthode self.update_figure() est chargée des mises à jour des calculs et des affichages. """
if self.lod_calc != self.lod_calc_ante:
self.lod_calc_ante = self.ut.deepcopy(self.lod_calc)
self.update_figure()
watch()
délègue la mise à jour des calculs et des affichages à la méthode update_figure()
(en dernière ligne).update_figure()
: live-update du titre de la fenêtre.ShowMatPlotLib.update_figure()
:
def update_figure(self):
""" Update suite aux modifications pkl. """
""" 1 - Titre de la fenêtre. """
l_keys = [f'Node{self.node_id}']
if self.lod_calc and 'edges' in self.lod_calc[0]:
local = ' - Mode local' if len(sys.argv) < 6 else ''
figure_title = self.lod_calc[0].read(l_keys + ['Fenêtre', 'Titre'], 'Vide')
self.mgr.set_window_title(figure_title + local)
Plots-0
. Assurez-vous que l'interface sélectionnée est Matplotlib
.Pycharm
, show_matplotlib > Run ...
Voir
du node Plots-0
.Paramètres du node 'Node0' > Fenêtre > Titre
.Matplotlib
, à savoir :
Calcul
du node serveur de signaux, qui est simplement le node connecté au Plot-0
.POO
: nécessite un peu plus de préparation, mais offre des possibilités bien supérieures.pyplot
, nommé plt
, qui fournit la figure : figure = plt.figure()
, c'est à dire la fenêtre d'affichage elle-même.figure
peut contenir plusieurs graphiques
, nommés subplots
ou axes
.
Plots
a jusqu'à 8 entrées, on se limitera donc à 8.""" Initialisations. """
du setup()
doit donc être modifié.ShowMatPlotLib.setup()
:
""" Initialisation de matplotlib.pyplot. """
plt.style.use('seaborn') # Pour voir les styles disponibles, faire print(plt.style.available).
self.fig = plt.figure(constrained_layout=True)
self.ax[0] = self.fig.add_subplot()
self.ax[0].plot(np.empty(0)) # Ajout d'un objet 'line' (une courbe) dans le tableau self.ax[0].lines[]
self.ax[0].lines[0].name = 'aleatoire'
""" Attributs 'pointer' et 'datas' ajoutés dans l'objet self.ax[0]. """
self.ax[0].pointer = 0
self.ax[0].datas = np.cumsum(np.random.randn(100_000)) # Vecteur de 100 000 valeurs.
On recode la méthode show_anim()
, qui doit être adaptée à la lecture de grandes matrices numpy
avec pointeur glissant.
ShowMatPlotLib.show_anim()
:
# noinspection PyUnresolvedReferences
def show_anim(self):
""" Pointeur et objet o_line = 1 courbe. """
ax = self.ax[0]
ax.pointer += 1
o_line = ax.lines[0]
""" Abscisse. """
x = np.arange(ax.pointer, ax.pointer + self.nb_points)
ax.set_xlim(ax.pointer, ax.pointer + self.nb_points)
o_line.set_xdata(x)
""" Ordonnée. """
y = ax.datas[ax.pointer: ax.pointer + self.nb_points]
y_min, y_max = np.min(y), np.max(y)
ax.set_ylim(y_min, y_max)
o_line.set_ydata(y)
""" Affichage """
plt.draw()
Plots
) possède propre sa classe Calcul
, localisée dans le fichier {node}.py
.CtrlCalcul
, qui contient le code factorisé et qui est localisée dans le fichier ctrl_node.py
.CtrlNode
, nous allons procéder à un refactoring :
Helper
, au début du fichier ctrl_node.py
, après les imports.ctrl_node.py > Helper
:
class Helper:
@staticmethod
def deepcopy(reference):
return copy.deepcopy(reference)
@staticmethod
def new_od(dic=None):
return Dictionary(dic)
CtrlNode
, en modifiant son en-tête :
class CtrlNode(Helper):
Rechercher et supprimer les méthodes deepcopy()
et new_od()
de cette classe CtrlNode
.
CtrlCalcul
, à la fin du fichier ctrl_node.py
(Remarquez qu'elle hérite également du helper).ctrl_node.py > CtrlCalcul
:
class CtrlCalcul(Helper):
def __init__(self):
self.d_signals = dict()
self.od_config = self.new_od()
self.np_array = None
self.b_busy = False
/nodes/generateurs/signaux.py
. A la fin de ce fichier, copier cette classe.node /nodes/generateurs/signaux.py > Calcul
:
class Calcul(CtrlCalcul):
def __init__(self):
super().__init__()
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))
""" Création, si inexistant, du tableau de sortie. """
if self.np_array is None:
self.np_array = np.empty((od_self.read('Taille buffer'), 0))
""" 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'):
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é'):
# à coder ...
#
# if od_self.read('Triangle'):
# à coder ...
#
# if od_self.read('Dent de scie montante'):
# à coder ...
#
# if od_self.read('Dent de scie descendante'):
# à coder ...
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
@staticmethod
def get_sinus_wave(od_self):
period = od_self.read('Points par période')
if period == 0:
return None
amplitude = od_self.read('Amplitude')
len_buffer = od_self.read('Taille buffer')
nb_iter = len_buffer // (2 * period)
x = np.linspace(0, 2 * np.pi, period + 1)[:-1]
y = np.reshape(amplitude * np.sin(x), (period, 1))
return np.concatenate((y, y) * nb_iter)
@staticmethod
def get_cosinus_wave(od_self):
period = od_self.read('Points par période')
if period == 0:
return None
amplitude = od_self.read('Amplitude')
len_buffer = od_self.read('Taille buffer')
nb_iter = len_buffer // (2 * period)
x = np.linspace(0, 2 * np.pi, period + 1)[:-1]
y = np.reshape(amplitude * np.cos(x), (period, 1))
return np.concatenate((y, y) * nb_iter)
Importer CtrlCalcul
et numpy
(alias np
).
update_figure()
: update des calculs.update_figure()
, appelée à chaque modification depuis le poste de contrôle, est ainsi structurée, dans l'ordre :
Calcul
, o_calc
, par compilation dynamique.self.ax[0].lines[]
.o_calc
, méthode create_values()
.update_figure()
, codée de 1 à 5.ShowMatPlotLib.update_figure()
:
def update_figure(self):
""" Update suite aux modifications pkl. """
""" 1 - Titre de la fenêtre. """
if self.lod_calc and 'edges' in self.lod_calc[0]:
local = ' - Mode local' if len(sys.argv) < 6 else ''
figure_title = self.lod_calc[0].read([f'Node{self.node_id}', 'Fenêtre', 'Titre'], 'Vide')
self.mgr.set_window_title(figure_title + local)
""" 2 - od_calc est le super-dictionnaire de l'entrée 0. """
if len(self.lod_calc) == 0:
""" En cas d'absence de dictionnaire de calcul, on efface les courbes affichées et on sort. """
l_lines = self.ax[0].lines[:]
for o_line in l_lines:
self.ax[0].lines.remove(o_line)
return
od_calc = self.lod_calc[0]
""" 3 - Serveur de signaux : Il s'agit du node connecté à cette entrée.
Création de l'objet Calcul par compilation dynamique. """
node_sid, num_input = self.get_server_node(od_calc)
if node_sid is None:
return
""" 4 - Update des courbes : Actualisation de la liste des courbes self.ax[0].lines. """
l_keys = [f'Node{self.node_id}']
self.update_lines(num_input, od_calc.read(l_keys))
""" 5 - Calculs : Délégation à l'objet o_calc. """
o_calc = self.lo_calc[0]
od_calc.write([node_sid, 'Taille buffer'], 100_000)
o_calc.create_values(node_sid, od_calc)
od_lines = Dictionary(od_calc.read(l_keys))
l_key_lines = [key for (key, val) in od_lines.items() if isinstance(val, dict) and 'Signal actif' in val]
indx = -1
for key, val in o_calc.od_config.items():
if isinstance(val, dict) and val['actif'] and 'num_col' in val:
indx += 1
od_lines.write([l_key_lines[indx], 'num_col'], val['num_col'])
o_line = self.get_line(num_input, l_key_lines[indx])
if o_line is not None:
o_line.attr = od_lines.read(l_key_lines[indx])
""" 6 - Affichage - Pour chaque subplot : update des courbes avant leur traitement. """
# ...
Oui mais ... les méthodes get_server_node()
et update_lines()
n'existent pas, il faut les créer.
ShowMatPlotLib.get_server_node()
:
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 = {}
code = f'from {compile_from} import Calcul; o_calcul = Calcul()'
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
ShowMatPlotLib.update_lines()
:
def update_lines(self, num_input, d_plots):
""" Actualisation de la liste de courbes self.ax[num_input].lines[].
On compare les listes de coubes nécessaires et de courbes existantes.
On en déduit celles qu'il faut ajouter, celles qu'il faut supprimer. """
""" 1) Courbes nécessaires : l_needed. """
l_needed = list()
for key, val in d_plots.items():
if isinstance(val, dict) and 'Signal actif' in val and val['Signal actif']:
l_needed.append(key)
""" 2) Courbes existantes : l_exist. """
l_exist = self.line_exist(num_input)
""" 3) Ajout de courbes. """
for needed in l_needed:
if needed not in l_exist:
o_line, = self.ax[num_input].plot(np.empty(0)) # Élément 0 de la liste.
o_line.name = needed
""" 4) Suppression de courbes. """
for exist in l_exist:
if exist not in l_needed:
o_line = self.get_line(num_input, exist)
if o_line is not None:
self.ax[num_input].lines.remove(o_line)
Oui mais ... les méthodes line_exist()
et get_line()
, nécessaires dans update_lines()
n'existent pas. Créons-les.
ShowMatPlotLib.line_exist()
:
def line_exist(self, num_input):
l_lines = list()
for line in self.ax[num_input].lines:
l_lines.append(line.name)
return l_lines
def get_line(self, num_input, line_name):
for o_line in self.ax[num_input].lines:
if o_line.name == line_name:
return o_line
return None
show_anim()
est provisoire, et nous a permis de faire la MAP (Mise au point) de notre code, avec un signal de test.numpy
contenant jusqu'à 6 colonnes (sinus, cosinus, ...) et 100 000 lignes.Signaux-1
.ShowMatPlotLib.show_anim()
:
# 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 = '... à coder ...'
y = '... à coder ...'
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:
# à coder...
""" Avancement du pointeur. """
if len(ax.lines) > 0: # Immobile si aucune courbe n'est affichée.
ax.pointer += 1
""" Affichage. """
plt.draw()
self.b_busy = False # Libre.
update_figure()
: live update de l'affichage.ShowMatPlotLib.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)
""" Nom complet du fichier de configuration, éditable manuellement en yaml. """
# à compléter plus tard.
""" Initialisation de matplotlib.pyplot. """
plt.style.use('seaborn') # Pour voir les styles disponibles, faire print(plt.style.available).
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.watch()
""" É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()
ShowMatPlotLib.update_figure()
:
def update_figure(self):
""" Update suite aux modifications pkl. """
""" 1 - Titre de la fenêtre. """
l_keys = [f'Node{self.node_id}']
if self.lod_calc and 'edges' in self.lod_calc[0]:
local = ' - Mode local' if len(sys.argv) < 6 else ''
figure_title = self.lod_calc[0].read(l_keys + ['Fenêtre', 'Titre'], 'Vide')
self.mgr.set_window_title(figure_title + local)
""" 2 - od_calc est le super-dictionnaire de l'entrée 0. """
if len(self.lod_calc) == 0:
""" En cas d'absence de dictionnaire de calcul, on efface les courbes affichées et on sort. """
l_lines = self.ax[0].lines[:]
for o_line in l_lines:
self.ax[0].lines.remove(o_line)
return
od_calc = self.lod_calc[0]
""" 3 - Serveur de signaux : Il s'agit du node connecté à cette entrée.
Création de l'objet Calcul par compilation dynamique. """
node_sid, num_input = self.get_server_node(od_calc)
if node_sid is None:
return
""" 4 - Update des courbes : Actualisation de la liste des courbes self.ax[0].lines. """
self.update_lines(num_input, od_calc.read(l_keys))
""" 5 - Calculs : Délégation à l'objet o_calc. """
o_calc = self.lo_calc[0]
od_calc.write([node_sid, 'Taille buffer'], 100_000)
o_calc.create_values(node_sid, od_calc)
od_lines = Dictionary(od_calc.read(l_keys))
l_key_lines = [key for (key, val) in od_lines.items() if isinstance(val, dict) and 'Signal actif' in val]
indx = -1
for key, val in o_calc.od_config.items():
if isinstance(val, dict) and val['actif'] and 'num_col' in val:
indx += 1
od_lines.write([l_key_lines[indx], 'num_col'], val['num_col'])
o_line = self.get_line(num_input, l_key_lines[indx])
if o_line is not None:
o_line.attr = od_lines.read(l_key_lines[indx])
""" 6 - Affichage - Pour chaque subplot : update des courbes avant leur traitement. """
""" Parcours des courbes de cette entrée de node. """
for o_line in self.ax[0].lines:
od_line = Dictionary(od_calc.read(l_keys + [o_line.name]))
if not od_line:
continue
""" Affectation des attributs : couleur, épaisseur, style, légende. """
# https://matplotlib.org/stable/api/_as_gen/matplotlib.lines.Line2D.html
o_line.set_color(od_line.read('Couleur'))
o_line.set_linewidth(od_line.read('Épaisseur'))
o_line.set_linestyle(self.get_linestyle(od_line.read('Style')))
o_line.set_label(od_line.read('Légende'))
""" Fenêtre : titre et légendes. """
# https://matplotlib.org/stable/api/legend_api.html
title = od_calc.read(l_keys + ['Titre'])
self.ax[0].title.set_text(title)
self.ax[0].legend(loc='upper left', ncol=2, framealpha=.8, facecolor='#ffe').set_frame_on(True)
plt.draw() # Nécessaire lorsque l'affichage est en pause.
ShowMatPlotLib.get_linestyle()
:
@staticmethod
def get_linestyle(param_style):
if param_style.startswith('___'):
return '-'
elif param_style.startswith('- -'):
return '--'
elif param_style.startswith('. .'):
return ':'
return '-.'
Vérification
Voir
du poste de contrôle :
Si le lancement a lieu depuis le PC, Mode local n'apparaît plus dans le titre.
Snippets
Bonjour les codeurs !