17272281697

Committed 27 Aug 2025 04:09PM UTC coverage: 97.727% (-0.04%) from 97.767%

Build # 17272281697

Build Type

push

github

Committed by

tonegas

Commit Message

minor chages

Run Details

8 of 12 new or added lines in 2 files covered. (66.67%)

14 existing lines in 4 files now uncovered.

12727 of 13023 relevant lines covered (97.73%)

0.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.67

/nnodely/visualizer/mplvisualizer.py

import subprocess, json, os, importlib
import numpy as np

from nnodely.visualizer.textvisualizer import TextVisualizer
from nnodely.layers.fuzzify import return_fuzzify
from nnodely.layers.parametricfunction import return_standard_inputs, return_function
from nnodely.support.utils import check
from nnodely.basic.modeldef import ModelDef

from nnodely.support.logger import logging, nnLogger
log = nnLogger(__name__, logging.INFO)

def get_library_path(library_name):
    spec = importlib.util.find_spec(library_name)
    if spec is None:
        raise ImportError(f"Library {library_name} not found")
    return os.path.dirname(spec.origin)

class MPLVisualizer(TextVisualizer):
    def __init__(self, verbose = 1):
        super().__init__(verbose)
        # Path to the data visualizer script
        import signal
        import sys
        get_library_path('nnodely')
        self.__training_visualizer_script = os.path.join(get_library_path('nnodely'),'visualizer','dynamicmpl','trainingplot.py')
        self.__time_series_visualizer_script = os.path.join(get_library_path('nnodely'),'visualizer','dynamicmpl','resultsplot.py')
        self.__fuzzy_visualizer_script = os.path.join(get_library_path('nnodely'),'visualizer','dynamicmpl','fuzzyplot.py')
        self.__function_visualizer_script = os.path.join(get_library_path('nnodely'),'visualizer','dynamicmpl','functionplot.py')
        self.__process_training = {}
        self.__process_results = {}
        self.__process_function = {}
        def signal_handler(sig, frame):
            for key in self.__process_training.keys():
                self.__process_training[key].terminate()
                self.__process_training[key].wait()
            for name_data in self.__process_results.keys():
                for key in self.__process_results[name_data].keys():
                    self.__process_results[name_data][key].terminate()
                    self.__process_results[name_data][key].wait()
            self.__process_results = {}
            for key in self.__process_function.keys():
                self.__process_function[key].terminate()
                self.__process_functios[key].wait()
            sys.exit()

        signal.signal(signal.SIGINT, signal_handler)

    def showStartTraining(self):
        pass

    def showTraining(self, epoch, train_losses, val_losses):
        if epoch == 0:
            for key in self.__process_training.keys():
                if self.__process_training[key].poll() is None:
                    self.__process_training[key].terminate()
                    self.__process_training[key].wait()
                self.__process_training[key] = {}

            self.__process_training = {}
            for key in self.modely._model_def['Minimizers'].keys():
                self.__process_training[key] = subprocess.Popen(['python', self.__training_visualizer_script], stdin=subprocess.PIPE, text=True)

        num_of_epochs = self.modely.running_parameters['num_of_epochs']
        train_tag = self.modely.running_parameters['train_tag']
        val_tag = self.modely.running_parameters['val_tag']
        if epoch+1 <= num_of_epochs:
            for key in self.modely._model_def['Minimizers'].keys():
                if val_losses:
                    val_loss = val_losses[key][epoch]
                else:
                    val_loss = []
                data = {"title":f"Training on {train_tag} and {val_tag}", "key": key, "last": num_of_epochs - (epoch + 1), "epoch": epoch,
                        "train_losses": train_losses[key][epoch], "val_losses": val_loss}
                try:
                    # Send data to the visualizer process
                    self.__process_training[key].stdin.write(f"{json.dumps(data)}\n")
                    self.__process_training[key].stdin.flush()
                except BrokenPipeError:
                    self.closeTraining()
                    log.warning("The visualizer process has been closed.")

        if epoch+1 == num_of_epochs:
            for key in self.modely._model_def['Minimizers'].keys():
                self.__process_training[key].stdin.close()

    def showResult(self, name_data):
        super().showResult(name_data)
        check(name_data in self.modely.performance, ValueError, f"Results not available for {name_data}.")
        if name_data in self.__process_results:
            for key in self.modely._model_def['Minimizers'].keys():
                if key in self.__process_results[name_data] and self.__process_results[name_data][key].poll() is None:
                    self.__process_results[name_data][key].terminate()
                    self.__process_results[name_data][key].wait()
                self.__process_results[name_data][key] = None
        self.__process_results[name_data] = {}

        for key in self.modely._model_def['Minimizers'].keys():
            # Start the data visualizer process
            self.__process_results[name_data][key] = subprocess.Popen(['python', self.__time_series_visualizer_script], stdin=subprocess.PIPE,
                                                    text=True)
            np_data_A = np.array(self.modely.prediction[name_data][key]['A'])
            if len(np_data_A.shape) > 3 and np_data_A.shape[1] > 20:
                np_data_B = np.array(self.modely.prediction[name_data][key]['B'])
                indices = np.linspace(0, np_data_A.shape[1] - 1, 20, dtype=int)
                data_A = np_data_A[:, indices, :, :].tolist()
                data_B = np_data_B[:, indices, :, :].tolist()
            else:
                data_A = self.modely.prediction[name_data][key]['A']
                data_B = self.modely.prediction[name_data][key]['B']

            data = {"name_data": name_data,
                    "key": key,
                    "performance": self.modely.performance[name_data][key],
                    "prediction_A": data_A,
                    "prediction_B": data_B,
                    "sample_time": self.modely._model_def['Info']["SampleTime"]}
            try:
                # Send data to the visualizer process
                self.__process_results[name_data][key].stdin.write(f"{json.dumps(data)}\n")
                self.__process_results[name_data][key].stdin.flush()
                self.__process_results[name_data][key].stdin.close()
            except BrokenPipeError:
                self.closeResult(self, name_data)
                log.warning(f"The visualizer {name_data} process has been closed.")

    def showWeights(self, weights = None):
        pass

    def showFunctions(self, functions = None, xlim = None, num_points = 1000):
        check(self.modely.neuralized, ValueError, "The model has not been neuralized.")
        for key, value in self.modely._model_def['Functions'].items():
            if key in functions:
                if key in self.__process_function and self.__process_function[key].poll() is None:
                    self.__process_function[key].terminate()
                    self.__process_function[key].wait()

                if 'functions' in self.modely._model_def['Functions'][key]:
                    x, activ_fun = return_fuzzify(value, xlim, num_points)
                    data = {"name": key,
                            "x": x,
                            "y": activ_fun,
                            "chan_centers": value['centers']}
                    # Start the data visualizer process
                    self.__process_function[key] = subprocess.Popen(['python', self.__fuzzy_visualizer_script],
                                                                  stdin=subprocess.PIPE,
                                                                  text=True)
                elif 'code':
                    model_def = ModelDef(self.modely._model_def)
                    model_def.updateParameters(self.modely._model)
                    function_inputs = return_standard_inputs(value, model_def, xlim, num_points)
                    function_output, function_input_list = return_function(value, function_inputs)

                    data = {"name": key}
                    if value['n_input'] == 2:
                        data['x0'] = function_inputs[0].reshape(num_points, num_points).tolist()
                        data['x1'] = function_inputs[1].reshape(num_points, num_points).tolist()
                        data['output'] = function_output.reshape(num_points, num_points).tolist()
                    else:
                        data['x0'] = function_inputs[0].reshape(num_points).tolist()
                        data['output'] = function_output.reshape(num_points).tolist()
                    data['params'] = []
                    for i, key in enumerate(value['params_and_consts']):
                        data['params'] += [function_inputs[i+value['n_input']].tolist()]
                    data['input_names'] = function_input_list

                    # Start the data visualizer process
                    self.__process_function[key] = subprocess.Popen(['python', self.__function_visualizer_script],
                                                                  stdin=subprocess.PIPE,
                                                                  text=True)
                try:
                    # Send data to the visualizer process
                    self.__process_function[key].stdin.write(f"{json.dumps(data)}\n")
                    self.__process_function[key].stdin.flush()
                    self.__process_function[key].stdin.close()
                except BrokenPipeError:
                    self.closeFunctions()
                    log.warning(f"The visualizer {functions} process has been closed.")

    def closeFunctions(self, functions = None):
        if functions is None:
            for key in self.__process_function.keys():
                self.__process_function[key].terminate()
                self.__process_function[key].wait()
            self.__process_function = {}
        else:
            for key in functions:
                self.__process_function[key].terminate()
                self.__process_function[key].wait()
                self.__process_function.pop(key)

    def closeTraining(self, minimizer = None):
        if minimizer is None:
            for key in self.modely._model_def['Minimizers'].keys():
                if key in self.__process_training and self.__process_training[key].poll() is None:
                    self.__process_training[key].terminate()
                    self.__process_training[key].wait()
                self.__process_training[key] = {}
        else:
            self.__process_training[minimizer].terminate()
            self.__process_training[minimizer].wait()
            self.__process_training.pop(minimizer)

    def closeResult(self, name_data = None, minimizer = None):
        if name_data is None:
            check(minimizer is None, ValueError, "If name_data is None, minimizer must be None.")
            for name_data in self.__process_results.keys():
                for key in self.__process_results[name_data].keys():
                    self.__process_results[name_data][key].terminate()
                    self.__process_results[name_data][key].wait()
            self.__process_results = {}
        else:
            if minimizer is None:
                for key in self.__process_results[name_data].keys():
                    self.__process_results[name_data][key].terminate()
                    self.__process_results[name_data][key].wait()
                self.__process_results[name_data] = {}
            else:
                self.__process_results[name_data][minimizer].terminate()
                self.__process_results[name_data][minimizer].wait()
                self.__process_results[name_data].pop(minimizer)






1	import subprocess, json, os, importlib	1✔
2	import numpy as np	1✔
3
4	from nnodely.visualizer.textvisualizer import TextVisualizer	1✔
5	from nnodely.layers.fuzzify import return_fuzzify	1✔
6	from nnodely.layers.parametricfunction import return_standard_inputs, return_function	1✔
7	from nnodely.support.utils import check	1✔
8	from nnodely.basic.modeldef import ModelDef	1✔
9
10	from nnodely.support.logger import logging, nnLogger	1✔
11	log = nnLogger(__name__, logging.INFO)	1✔
12
13	def get_library_path(library_name):	1✔
14	spec = importlib.util.find_spec(library_name)	1✔
15	if spec is None:	1✔
16	raise ImportError(f"Library {library_name} not found")	×
17	return os.path.dirname(spec.origin)	1✔
18
19	class MPLVisualizer(TextVisualizer):	1✔
20	def __init__(self, verbose = 1):	1✔
21	super().__init__(verbose)	1✔
22	# Path to the data visualizer script
23	import signal	1✔
24	import sys	1✔
25	get_library_path('nnodely')	1✔
26	self.__training_visualizer_script = os.path.join(get_library_path('nnodely'),'visualizer','dynamicmpl','trainingplot.py')	1✔
27	self.__time_series_visualizer_script = os.path.join(get_library_path('nnodely'),'visualizer','dynamicmpl','resultsplot.py')	1✔
28	self.__fuzzy_visualizer_script = os.path.join(get_library_path('nnodely'),'visualizer','dynamicmpl','fuzzyplot.py')	1✔
29	self.__function_visualizer_script = os.path.join(get_library_path('nnodely'),'visualizer','dynamicmpl','functionplot.py')	1✔
30	self.__process_training = {}	1✔
31	self.__process_results = {}	1✔
32	self.__process_function = {}	1✔
33	def signal_handler(sig, frame):	1✔
34	for key in self.__process_training.keys():	×
35	self.__process_training[key].terminate()	×
36	self.__process_training[key].wait()	×
37	for name_data in self.__process_results.keys():	×
38	for key in self.__process_results[name_data].keys():	×
39	self.__process_results[name_data][key].terminate()	×
40	self.__process_results[name_data][key].wait()	×
41	self.__process_results = {}	×
42	for key in self.__process_function.keys():	×
43	self.__process_function[key].terminate()	×
44	self.__process_functios[key].wait()	×
45	sys.exit()	×
46
47	signal.signal(signal.SIGINT, signal_handler)	1✔
48
49	def showStartTraining(self):	1✔
50	pass	1✔
51
52	def showTraining(self, epoch, train_losses, val_losses):	1✔
53	if epoch == 0:	1✔
54	for key in self.__process_training.keys():	1✔
55	if self.__process_training[key].poll() is None:	1✔
56	self.__process_training[key].terminate()	1✔
57	self.__process_training[key].wait()	1✔
58	self.__process_training[key] = {}	1✔
59
60	self.__process_training = {}	1✔
61	for key in self.modely._model_def['Minimizers'].keys():	1✔
62	self.__process_training[key] = subprocess.Popen(['python', self.__training_visualizer_script], stdin=subprocess.PIPE, text=True)	1✔
63
64	num_of_epochs = self.modely.running_parameters['num_of_epochs']	1✔
65	train_tag = self.modely.running_parameters['train_tag']	1✔
66	val_tag = self.modely.running_parameters['val_tag']	1✔
67	if epoch+1 <= num_of_epochs:	1✔
68	for key in self.modely._model_def['Minimizers'].keys():	1✔
69	if val_losses:	1✔
70	val_loss = val_losses[key][epoch]	×
71	else:
72	val_loss = []	1✔
73	data = {"title":f"Training on {train_tag} and {val_tag}", "key": key, "last": num_of_epochs - (epoch + 1), "epoch": epoch,	1✔
74	"train_losses": train_losses[key][epoch], "val_losses": val_loss}
75	try:	1✔
76	# Send data to the visualizer process
77	self.__process_training[key].stdin.write(f"{json.dumps(data)}\n")	1✔
78	self.__process_training[key].stdin.flush()	1✔
79	except BrokenPipeError:	×
80	self.closeTraining()	×
81	log.warning("The visualizer process has been closed.")	×
82
83	if epoch+1 == num_of_epochs:	1✔
84	for key in self.modely._model_def['Minimizers'].keys():	1✔
85	self.__process_training[key].stdin.close()	1✔
86
87	def showResult(self, name_data):	1✔
88	super().showResult(name_data)	1✔
89	check(name_data in self.modely.performance, ValueError, f"Results not available for {name_data}.")	1✔
90	if name_data in self.__process_results:	1✔
91	for key in self.modely._model_def['Minimizers'].keys():	1✔
92	if key in self.__process_results[name_data] and self.__process_results[name_data][key].poll() is None:	1✔
93	self.__process_results[name_data][key].terminate()	1✔
94	self.__process_results[name_data][key].wait()	1✔
95	self.__process_results[name_data][key] = None	1✔
96	self.__process_results[name_data] = {}	1✔
97
98	for key in self.modely._model_def['Minimizers'].keys():	1✔
99	# Start the data visualizer process
100	self.__process_results[name_data][key] = subprocess.Popen(['python', self.__time_series_visualizer_script], stdin=subprocess.PIPE,	1✔
101	text=True)
102	np_data_A = np.array(self.modely.prediction[name_data][key]['A'])	1✔
103	if len(np_data_A.shape) > 3 and np_data_A.shape[1] > 20:	1✔
NEW 104	np_data_B = np.array(self.modely.prediction[name_data][key]['B'])	×
NEW 105	indices = np.linspace(0, np_data_A.shape[1] - 1, 20, dtype=int)	×
NEW 106	data_A = np_data_A[:, indices, :, :].tolist()	×
NEW 107	data_B = np_data_B[:, indices, :, :].tolist()	×
108	else:
109	data_A = self.modely.prediction[name_data][key]['A']	1✔
110	data_B = self.modely.prediction[name_data][key]['B']	1✔
111
112	data = {"name_data": name_data,	1✔
113	"key": key,
114	"performance": self.modely.performance[name_data][key],
115	"prediction_A": data_A,
116	"prediction_B": data_B,
117	"sample_time": self.modely._model_def['Info']["SampleTime"]}
118	try:	1✔
119	# Send data to the visualizer process
120	self.__process_results[name_data][key].stdin.write(f"{json.dumps(data)}\n")	1✔
121	self.__process_results[name_data][key].stdin.flush()	1✔
122	self.__process_results[name_data][key].stdin.close()	1✔
123	except BrokenPipeError:	×
124	self.closeResult(self, name_data)	×
125	log.warning(f"The visualizer {name_data} process has been closed.")	×
126
127	def showWeights(self, weights = None):	1✔
128	pass	×
129
130	def showFunctions(self, functions = None, xlim = None, num_points = 1000):	1✔
131	check(self.modely.neuralized, ValueError, "The model has not been neuralized.")	1✔
132	for key, value in self.modely._model_def['Functions'].items():	1✔
133	if key in functions:	1✔
134	if key in self.__process_function and self.__process_function[key].poll() is None:	1✔
135	self.__process_function[key].terminate()	×
136	self.__process_function[key].wait()	×
137
138	if 'functions' in self.modely._model_def['Functions'][key]:	1✔
139	x, activ_fun = return_fuzzify(value, xlim, num_points)	1✔
140	data = {"name": key,	1✔
141	"x": x,
142	"y": activ_fun,
143	"chan_centers": value['centers']}
144	# Start the data visualizer process
145	self.__process_function[key] = subprocess.Popen(['python', self.__fuzzy_visualizer_script],	1✔
146	stdin=subprocess.PIPE,
147	text=True)
148	elif 'code':	1✔
149	model_def = ModelDef(self.modely._model_def)	1✔
150	model_def.updateParameters(self.modely._model)	1✔
151	function_inputs = return_standard_inputs(value, model_def, xlim, num_points)	1✔
152	function_output, function_input_list = return_function(value, function_inputs)	1✔
153
154	data = {"name": key}	1✔
155	if value['n_input'] == 2:	1✔
156	data['x0'] = function_inputs[0].reshape(num_points, num_points).tolist()	1✔
157	data['x1'] = function_inputs[1].reshape(num_points, num_points).tolist()	1✔
158	data['output'] = function_output.reshape(num_points, num_points).tolist()	1✔
159	else:
160	data['x0'] = function_inputs[0].reshape(num_points).tolist()	×
161	data['output'] = function_output.reshape(num_points).tolist()	×
162	data['params'] = []	1✔
163	for i, key in enumerate(value['params_and_consts']):	1✔
164	data['params'] += [function_inputs[i+value['n_input']].tolist()]	1✔
165	data['input_names'] = function_input_list	1✔
166
167	# Start the data visualizer process
168	self.__process_function[key] = subprocess.Popen(['python', self.__function_visualizer_script],	1✔
169	stdin=subprocess.PIPE,
170	text=True)
171	try:	1✔
172	# Send data to the visualizer process
173	self.__process_function[key].stdin.write(f"{json.dumps(data)}\n")	1✔
174	self.__process_function[key].stdin.flush()	1✔
175	self.__process_function[key].stdin.close()	1✔
176	except BrokenPipeError:	×
177	self.closeFunctions()	×
178	log.warning(f"The visualizer {functions} process has been closed.")	×
179
180	def closeFunctions(self, functions = None):	1✔
181	if functions is None:	1✔
182	for key in self.__process_function.keys():	1✔
183	self.__process_function[key].terminate()	1✔
184	self.__process_function[key].wait()	1✔
185	self.__process_function = {}	1✔
186	else:
187	for key in functions:	×
188	self.__process_function[key].terminate()	×
189	self.__process_function[key].wait()	×
190	self.__process_function.pop(key)	×
191
192	def closeTraining(self, minimizer = None):	1✔
193	if minimizer is None:	1✔
194	for key in self.modely._model_def['Minimizers'].keys():	1✔
195	if key in self.__process_training and self.__process_training[key].poll() is None:	1✔
196	self.__process_training[key].terminate()	1✔
197	self.__process_training[key].wait()	1✔
198	self.__process_training[key] = {}	1✔
199	else:
200	self.__process_training[minimizer].terminate()	×
201	self.__process_training[minimizer].wait()	×
202	self.__process_training.pop(minimizer)	×
203
204	def closeResult(self, name_data = None, minimizer = None):	1✔
205	if name_data is None:	1✔
206	check(minimizer is None, ValueError, "If name_data is None, minimizer must be None.")	1✔
207	for name_data in self.__process_results.keys():	1✔
208	for key in self.__process_results[name_data].keys():	1✔
209	self.__process_results[name_data][key].terminate()	1✔
210	self.__process_results[name_data][key].wait()	1✔
211	self.__process_results = {}	1✔
212	else:
213	if minimizer is None:	×
214	for key in self.__process_results[name_data].keys():	×
215	self.__process_results[name_data][key].terminate()	×
216	self.__process_results[name_data][key].wait()	×
217	self.__process_results[name_data] = {}	×
218	else:
219	self.__process_results[name_data][minimizer].terminate()	×
220	self.__process_results[name_data][minimizer].wait()	×
221	self.__process_results[name_data].pop(minimizer)	×
222
223
224
225
226

tonegas / nnodely / 17272281697

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous