19576012316

Committed 21 Nov 2025 03:56PM UTC coverage: 96.605% (-0.04%) from 96.644%

Build # 19576012316

Build Type

push

github

Committed by

MisterMandarino

Commit Message

Merge branch 'develop' of https://github.com/tonegas/nnodely into develop

Run Details

50 of 50 new or added lines in 7 files covered. (100.0%)

71 existing lines in 10 files now uncovered.

12947 of 13402 relevant lines covered (96.6%)

0.97 hits per line

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

72.41

/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] > 50:
                np_data_B = np.array(self.modely.prediction[name_data][key]['B'])
                indices = np.linspace(0, np_data_A.shape[1] - 1, 50, dtype=int)
                data_A = np_data_A[:, indices, :, :].tolist()
                data_B = np_data_B[:, indices, :, :].tolist()
                data_idxs = np.array(self.modely.prediction[name_data]['idxs'])[:,indices].tolist()
            else:
                data_A = self.modely.prediction[name_data][key]['A']
                data_B = self.modely.prediction[name_data][key]['B']
                data_idxs = None

            data = {"name_data": name_data,
                    "key": key,
                    "performance": self.modely.performance[name_data][key],
                    "prediction_A": data_A,
                    "prediction_B": data_B,
                    "data_idxs": data_idxs,
                    "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] > 50:	1✔
104	np_data_B = np.array(self.modely.prediction[name_data][key]['B'])	×
105	indices = np.linspace(0, np_data_A.shape[1] - 1, 50, dtype=int)	×
106	data_A = np_data_A[:, indices, :, :].tolist()	×
107	data_B = np_data_B[:, indices, :, :].tolist()	×
UNCOV 108	data_idxs = np.array(self.modely.prediction[name_data]['idxs'])[:,indices].tolist()	×
109	else:
110	data_A = self.modely.prediction[name_data][key]['A']	1✔
111	data_B = self.modely.prediction[name_data][key]['B']	1✔
112	data_idxs = None	1✔
113
114	data = {"name_data": name_data,	1✔
115	"key": key,
116	"performance": self.modely.performance[name_data][key],
117	"prediction_A": data_A,
118	"prediction_B": data_B,
119	"data_idxs": data_idxs,
120	"sample_time": self.modely._model_def['Info']["SampleTime"]}
121	try:	1✔
122	# Send data to the visualizer process
123	self.__process_results[name_data][key].stdin.write(f"{json.dumps(data)}\n")	1✔
124	self.__process_results[name_data][key].stdin.flush()	1✔
125	self.__process_results[name_data][key].stdin.close()	1✔
UNCOV 126	except BrokenPipeError:	×
UNCOV 127	self.closeResult(self, name_data)	×
128	log.warning(f"The visualizer {name_data} process has been closed.")	×
129
130	def showWeights(self, weights = None):	1✔
UNCOV 131	pass	×
132
133	def showFunctions(self, functions = None, xlim = None, num_points = 1000):	1✔
134	check(self.modely.neuralized, ValueError, "The model has not been neuralized.")	1✔
135	for key, value in self.modely._model_def['Functions'].items():	1✔
136	if key in functions:	1✔
137	if key in self.__process_function and self.__process_function[key].poll() is None:	1✔
UNCOV 138	self.__process_function[key].terminate()	×
UNCOV 139	self.__process_function[key].wait()	×
140
141	if 'functions' in self.modely._model_def['Functions'][key]:	1✔
142	x, activ_fun = return_fuzzify(value, xlim, num_points)	1✔
143	data = {"name": key,	1✔
144	"x": x,
145	"y": activ_fun,
146	"chan_centers": value['centers']}
147	# Start the data visualizer process
148	self.__process_function[key] = subprocess.Popen(['python', self.__fuzzy_visualizer_script],	1✔
149	stdin=subprocess.PIPE,
150	text=True)
151	elif 'code':	1✔
152	model_def = ModelDef(self.modely._model_def)	1✔
153	model_def.updateParameters(self.modely._model)	1✔
154	function_inputs = return_standard_inputs(value, model_def, xlim, num_points)	1✔
155	function_output, function_input_list = return_function(value, function_inputs)	1✔
156
157	data = {"name": key}	1✔
158	if value['n_input'] == 2:	1✔
159	data['x0'] = function_inputs[0].reshape(num_points, num_points).tolist()	1✔
160	data['x1'] = function_inputs[1].reshape(num_points, num_points).tolist()	1✔
161	data['output'] = function_output.reshape(num_points, num_points).tolist()	1✔
162	else:
UNCOV 163	data['x0'] = function_inputs[0].reshape(num_points).tolist()	×
UNCOV 164	data['output'] = function_output.reshape(num_points).tolist()	×
165	data['params'] = []	1✔
166	for i, key in enumerate(value['params_and_consts']):	1✔
167	data['params'] += [function_inputs[i+value['n_input']].tolist()]	1✔
168	data['input_names'] = function_input_list	1✔
169
170	# Start the data visualizer process
171	self.__process_function[key] = subprocess.Popen(['python', self.__function_visualizer_script],	1✔
172	stdin=subprocess.PIPE,
173	text=True)
174	try:	1✔
175	# Send data to the visualizer process
176	self.__process_function[key].stdin.write(f"{json.dumps(data)}\n")	1✔
177	self.__process_function[key].stdin.flush()	1✔
178	self.__process_function[key].stdin.close()	1✔
UNCOV 179	except BrokenPipeError:	×
UNCOV 180	self.closeFunctions()	×
UNCOV 181	log.warning(f"The visualizer {functions} process has been closed.")	×
182
183	def closeFunctions(self, functions = None):	1✔
184	if functions is None:	1✔
185	for key in self.__process_function.keys():	1✔
186	self.__process_function[key].terminate()	1✔
187	self.__process_function[key].wait()	1✔
188	self.__process_function = {}	1✔
189	else:
190	for key in functions:	×
UNCOV 191	self.__process_function[key].terminate()	×
UNCOV 192	self.__process_function[key].wait()	×
UNCOV 193	self.__process_function.pop(key)	×
194
195	def closeTraining(self, minimizer = None):	1✔
196	if minimizer is None:	1✔
197	for key in self.modely._model_def['Minimizers'].keys():	1✔
198	if key in self.__process_training and self.__process_training[key].poll() is None:	1✔
199	self.__process_training[key].terminate()	1✔
200	self.__process_training[key].wait()	1✔
201	self.__process_training[key] = {}	1✔
202	else:
UNCOV 203	self.__process_training[minimizer].terminate()	×
UNCOV 204	self.__process_training[minimizer].wait()	×
UNCOV 205	self.__process_training.pop(minimizer)	×
206
207	def closeResult(self, name_data = None, minimizer = None):	1✔
208	if name_data is None:	1✔
209	check(minimizer is None, ValueError, "If name_data is None, minimizer must be None.")	1✔
210	for name_data in self.__process_results.keys():	1✔
211	for key in self.__process_results[name_data].keys():	1✔
212	self.__process_results[name_data][key].terminate()	1✔
213	self.__process_results[name_data][key].wait()	1✔
214	self.__process_results = {}	1✔
215	else:
216	if minimizer is None:	×
217	for key in self.__process_results[name_data].keys():	×
UNCOV 218	self.__process_results[name_data][key].terminate()	×
219	self.__process_results[name_data][key].wait()	×
220	self.__process_results[name_data] = {}	×
221	else:
UNCOV 222	self.__process_results[name_data][minimizer].terminate()	×
UNCOV 223	self.__process_results[name_data][minimizer].wait()	×
UNCOV 224	self.__process_results[name_data].pop(minimizer)	×
225
226
227
228
229

tonegas / nnodely / 19576012316

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