16502811447

Committed 24 Jul 2025 04:44PM UTC coverage: 97.767% (+0.1%) from 97.651%

Build # 16502811447

Build Type

push

github

Committed by

web-flow

Commit Message

New version 1.5.0

This pull request introduces version 1.5.0 of **nnodely**, featuring several updates:
1. Improved clarity of documentation and examples.
2. Support for managing multi-dataset features is now available.
3. DataFrames can now be used to create datasets.
4. Datasets can now be resampled.
5. Random data training has been fixed for both classic and recurrent training.
6. The `state` variable has been removed.
7. It is now possible to add or remove a connection or a closed loop.
8. Partial models can now be exported.
9. The `train` function and the result analysis have been separated.
10. A new function, `trainAndAnalyse`, is now available.
11. The report now works across all network types.
12. The training function code has been reorganized.

Run Details

2901 of 2967 new or added lines in 53 files covered. (97.78%)

16 existing lines in 6 files now uncovered.

12652 of 12941 relevant lines covered (97.77%)

0.98 hits per line

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

73.62

/nnodely/visualizer/mplvisualizer.py

import subprocess, json, os, importlib

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

tonegas / nnodely / 16502811447

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