14359872492

Committed 09 Apr 2025 02:33PM UTC coverage: 97.602% (+0.6%) from 97.035%

Build # 14359872492

Build Type

Pull #86

github

Committed by

web-flow

Commit Message

Merge ec719935a into e9c323c4f

Pull Request Pull Request #86: Smallclasses

Run Details

2291 of 2418 new or added lines in 54 files covered. (94.75%)

3 existing lines in 1 file now uncovered.

11683 of 11970 relevant lines covered (97.6%)

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.run_training_params['num_of_epochs']
        train_dataset = self.modely.run_training_params['train_dataset']
        validation_dataset = self.modely.run_training_params['validation_dataset']
        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_dataset} and {validation_dataset}", "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✔
NEW 33	for key in self.__process_training.keys():	×
NEW 34	self.__process_training[key].terminate()	×
NEW 35	self.__process_training[key].wait()	×
NEW 36	for name_data in self.__process_results.keys():	×
NEW 37	for key in self.__process_results[name_data].keys():	×
NEW 38	self.__process_results[name_data][key].terminate()	×
NEW 39	self.__process_results[name_data][key].wait()	×
NEW 40	self.__process_results = {}	×
NEW 41	for key in self.__process_function.keys():	×
NEW 42	self.__process_function[key].terminate()	×
NEW 43	self.__process_functios[key].wait()	×
UNCOV 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.run_training_params['num_of_epochs']	1✔
64	train_dataset = self.modely.run_training_params['train_dataset']	1✔
65	validation_dataset = self.modely.run_training_params['validation_dataset']	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✔
69	val_loss = val_losses[key][epoch]	1✔
70	else:
71	val_loss = []	×
72	data = {"title":f"Training on {train_dataset} and {validation_dataset}", "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✔
NEW 124	self.__process_function[key].terminate()	×
NEW 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:
UNCOV 149	data['x0'] = function_inputs[0].reshape(num_points).tolist()	×
UNCOV 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:	×
NEW 177	self.__process_function[key].terminate()	×
NEW 178	self.__process_function[key].wait()	×
NEW 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:
NEW 189	self.__process_training[minimizer].terminate()	×
NEW 190	self.__process_training[minimizer].wait()	×
NEW 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:	×
NEW 203	for key in self.__process_results[name_data].keys():	×
NEW 204	self.__process_results[name_data][key].terminate()	×
NEW 205	self.__process_results[name_data][key].wait()	×
NEW 206	self.__process_results[name_data] = {}	×
207	else:
NEW 208	self.__process_results[name_data][minimizer].terminate()	×
NEW 209	self.__process_results[name_data][minimizer].wait()	×
NEW 210	self.__process_results[name_data].pop(minimizer)	×
211
212
213
214
215

tonegas / nnodely / 14359872492

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