20075618707

Committed 09 Dec 2025 07:16PM UTC coverage: 96.949% (-0.8%) from 97.767%

Build # 20075618707

Build Type

Pull #109

github

Committed by

tonegas

Commit Message

Improved the coverage on train

Pull Request Pull Request #109: New version of nnodely

Run Details

878 of 893 new or added lines in 37 files covered. (98.32%)

157 existing lines in 5 files now uncovered.

13091 of 13503 relevant lines covered (96.95%)

0.97 hits per line

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

95.32

/nnodely/basic/modeldef.py

import copy

import numpy as np

from nnodely.support.utils import check, check_and_get_list
from nnodely.support.jsonutils import merge, subjson_from_model, subjson_from_minimize, check_model, get_models_json
from nnodely.basic.relation import MAIN_JSON, Stream, check_names
from nnodely.layers.output import Output
from nnodely.layers.input import Input

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


class ModelDef:
    def __init__(self, model_def = MAIN_JSON):
        # Models definition
        self.__json_base = copy.deepcopy(model_def)

        # Initialize the model definition
        self.__json = copy.deepcopy(self.__json_base)
        if "SampleTime" in self.__json['Info']:
            self.__sample_time = self.__json['Info']["SampleTime"]
        else:
            self.__sample_time = None

    def __contains__(self, key):
        return key in self.__json

    def __getitem__(self, key):
        if key in self.__json:
            return self.__json[key]
        else:
            return None

    def __setitem__(self, key, value):
        self.__json[key] = value

    def __rebuild_json(self, models_names, minimizers):
        models_json = subjson_from_model(self.__json, list(models_names))
        if 'Minimizers' in self.__json and len(minimizers) > 0:
            minimizers_json = subjson_from_minimize(self.__json, list(minimizers))
            models_json = merge(models_json, minimizers_json)
        return copy.deepcopy(models_json)

    def recurrentInputs(self):
        return {key:value for key, value in self.__json['Inputs'].items() if ('closedLoop' in value.keys() or 'connect' in value.keys())}

    def getJson(self, models:list|str|None = None) -> dict:
        if models is None:
            return copy.deepcopy(self.__json)
        else:
            json = subjson_from_model(self.__json, models)
            check_model(json)
            return copy.deepcopy(json)

    def getSampleTime(self):
        check(self.__sample_time is not None, AttributeError, "Sample time is not defined the model is not neuralized!")
        return self.__sample_time

    def isDefined(self):
        return self.__json is not None

    def addConnection(self, stream_out:str|Output|Stream, input_in:str|Input, type:str, local:bool = False):
        outputs = self.__json['Outputs']

        if isinstance(stream_out, (Output, Stream)):
            stream_name = outputs[stream_out.name] if stream_out.name in outputs.keys() else stream_out.name
        else:
            output_name = check_and_get_list(stream_out, set(outputs.keys()),
                                             lambda name: f"The name {name} is not part of the available Outputs")[0]
            stream_name = outputs[output_name]

        if isinstance(input_in, Input):
            input_name = input_in.name
        else:
            input_name = input_in #TODO Add tests

        input_name = check_and_get_list(input_name, set(self.__json['Inputs'].keys()),
                                       lambda name: f"The name {name} is not part of the available Inputs")[0]
        stream_name = check_and_get_list(stream_name, set(self.__json['Relations'].keys()),
                                        lambda name: f"The name {name} is not part of the available Relations")[0]
        self.__json['Inputs'][input_name][type] = stream_name
        self.__json['Inputs'][input_name]['local'] = int(local)

    def removeConnection(self, name_list:str|list[str]):
        name_list = check_and_get_list(name_list, set(self.__json['Inputs'].keys()), lambda name: f"The name {name} is not part of the available Inputs")
        for input_in in name_list:
            if 'closedLoop' in self.__json['Inputs'][input_in].keys():
                del self.__json['Inputs'][input_in]['closedLoop']
                del self.__json['Inputs'][input_in]['local']
            elif 'connect' in self.__json['Inputs'][input_in].keys():
                del self.__json['Inputs'][input_in]['connect']
                del self.__json['Inputs'][input_in]['local']
            else:
                raise ValueError(f"The input '{input_in}' has no connection or closed loop defined")

    def addModel(self, name:str, stream_list):
        if isinstance(stream_list, Output):
            stream_list = [stream_list]

        json = MAIN_JSON
        for stream in stream_list:
            json = merge(json, stream.json)
        check_model(json)

        if 'Models' not in self.__json:
            self.__json = merge(self.__json, json)
            self.__json['Models'] = name
        else:
            models_names = set((self.__json['Models'],)) if type(self.__json['Models']) is str else set(self.__json['Models'].keys())
            check_names(name, models_names, 'Models')
            if type(self.__json['Models']) is str:
                self.__json['Models'] = {self.__json['Models']: get_models_json(self.__json)}
            self.__json = merge(self.__json, json)
            self.__json['Models'][name] = get_models_json(json)

    def removeModel(self, name_list):
        if 'Models' not in self.__json:
            raise ValueError("No Models are defined")
        models_names = {self.__json['Models']} if type(self.__json['Models']) is str else set(self.__json['Models'].keys())
        name_list = check_and_get_list(name_list, models_names, lambda name: f"The name {name} is not part of the available models")
        models_names -= set(name_list)
        minimizers = set(self.__json['Minimizers'].keys()) if 'Minimizers' in self.__json else None
        self.__json = self.__rebuild_json(models_names, minimizers)

    def addMinimize(self, name, streamA, streamB, loss_function='mse'):
        if 'Minimizers' not in self.__json:
            self.__json['Minimizers'] = {}
        check_names(name, set(self.__json['Minimizers'].keys()), 'Minimizers')

        if isinstance(streamA, str):
            streamA_name = streamA
        else:
            check(isinstance(streamA, (Output, Stream)), TypeError, 'streamA must be an instance of Output or Stream')
            streamA_name = streamA.json['Outputs'][streamA.name] if isinstance(streamA, Output) else streamA.name
            self.__json = merge(self.__json, streamA.json)

        if isinstance(streamB, str):
            streamB_name = streamB
        else:
            check(isinstance(streamB, (Output, Stream)), TypeError, 'streamA must be an instance of Output or Stream')
            streamB_name = streamB.json['Outputs'][streamB.name] if isinstance(streamB, Output) else streamB.name
            self.__json = merge(self.__json, streamB.json)
        #check(streamA.dim == streamB.dim, ValueError, f'Dimension of streamA={streamA.dim} and streamB={streamB.dim} are not equal.')

        self.__json['Minimizers'][name] = {}
        self.__json['Minimizers'][name]['A'] = streamA_name
        self.__json['Minimizers'][name]['B'] = streamB_name
        self.__json['Minimizers'][name]['loss'] = loss_function

    def removeMinimize(self, name_list):
        if 'Minimizers' not in self.__json:
            raise ValueError("No Minimizers are defined")
        name_list = check_and_get_list(name_list, self.__json['Minimizers'].keys(), lambda name: f"The name {name} is not part of the available minimizers")
        models_names = {self.__json['Models']} if type(self.__json['Models']) is str else set(self.__json['Models'].keys())
        remaining_minimizers = set(self.__json['Minimizers'].keys()) - set(name_list) if 'Minimizers' in self.__json else None
        self.__json = self.__rebuild_json(models_names, remaining_minimizers)

    def setBuildWindow(self, sample_time = None):
        check(self.__json is not None, RuntimeError, "No model is defined!")
        if sample_time is not None:
            check(sample_time > 0, RuntimeError, 'Sample time must be strictly positive!')
            self.__sample_time = sample_time
        else:
            if self.__sample_time is None:
                self.__sample_time = 1

        self.__json['Info'] = {"SampleTime": self.__sample_time}
        if 'SampleTime' in self.__json['Constants']:
            self.__json['Constants']['SampleTime'] = {'dim': 1, 'values': self.__sample_time}

        check(self.__json['Inputs'] != {}, RuntimeError, "No model is defined!")
        json_inputs = self.__json['Inputs']

        input_ns_backward, input_ns_forward = {}, {}
        for key, value in json_inputs.items():
            if 'sw' not in value and 'tw' not in value:
                assert False, f"Input '{key}' has no time window or sample window"
            if 'sw' not in value and self.__sample_time is not None:
                ## check if value['tw'] is a multiple of sample_time
                absolute_tw = abs(value['tw'][0]) + abs(value['tw'][1])
                check(round(absolute_tw % self.__sample_time) == 0, ValueError,
                      f"Time window of input '{key}' is not a multiple of sample time. This network cannot be neuralized")
                input_ns_backward[key] = round(-value['tw'][0] / self.__sample_time)
                input_ns_forward[key] = round(value['tw'][1] / self.__sample_time)
            elif self.__sample_time is not None:
                if 'tw' in value:
                    input_ns_backward[key] = max(round(-value['tw'][0] / self.__sample_time), -value['sw'][0])
                    input_ns_forward[key] = max(round(value['tw'][1] / self.__sample_time), value['sw'][1])
                else:
                    input_ns_backward[key] = -value['sw'][0]
                    input_ns_forward[key] =  value['sw'][1]
            else:
                check(value['tw'] == [0,0], RuntimeError, f"Sample time is not defined for input '{key}'")
                input_ns_backward[key] = -value['sw'][0]
                input_ns_forward[key] = value['sw'][1]
            value['ns'] = [input_ns_backward[key], input_ns_forward[key]]
            value['ntot'] = sum(value['ns'])

        self.__json['Info']['ns'] = [max(input_ns_backward.values()), max(input_ns_forward.values())]
        self.__json['Info']['ntot'] = sum(self.__json['Info']['ns'])
        if self.__json['Info']['ns'][0] < 0:
            log.warning(
                f"The input is only in the far past the max_samples_backward is: {self.__json['Info']['ns'][0]}")
        if self.__json['Info']['ns'][1] < 0:
            log.warning(
                f"The input is only in the far future the max_sample_forward is: {self.__json['Info']['ns'][1]}")

        for k, v in (self.__json['Parameters'] | self.__json['Constants']).items():
            if 'values' in v:
                window = 'tw' if 'tw' in v.keys() else ('sw' if 'sw' in v.keys() else None)
                if window == 'tw':
                    check(np.array(v['values']).shape[0] == v['tw'] / self.__sample_time, ValueError,
                      f"{k} has a different number of values for this sample time.")
                if v['values'] == "SampleTime":
                    v['values'] = self.__sample_time

    def updateParameters(self, model = None, *, clear_model = False):
        if clear_model:
            for key in self.__json['Parameters'].keys():
                if 'init_values' in self.__json['Parameters'][key]:
                    self.__json['Parameters'][key]['values'] = self.__json['Parameters'][key]['init_values']
                elif 'values' in self.__json['Parameters'][key]:
                    del self.__json['Parameters'][key]['values']
        elif model is not None:
            for key in self.__json['Parameters'].keys():
                if key in model.all_parameters:
                    self.__json['Parameters'][key]['values'] = model.all_parameters[key].tolist()


1	import copy	1✔
2
3	import numpy as np	1✔
4
5	from nnodely.support.utils import check, check_and_get_list	1✔
6	from nnodely.support.jsonutils import merge, subjson_from_model, subjson_from_minimize, check_model, get_models_json	1✔
7	from nnodely.basic.relation import MAIN_JSON, Stream, check_names	1✔
8	from nnodely.layers.output import Output	1✔
9	from nnodely.layers.input import Input	1✔
10
11	from nnodely.support.logger import logging, nnLogger	1✔
12	log = nnLogger(__name__, logging.INFO)	1✔
13
14
15	class ModelDef:	1✔
16	def __init__(self, model_def = MAIN_JSON):	1✔
17	# Models definition
18	self.__json_base = copy.deepcopy(model_def)	1✔
19
20	# Initialize the model definition
21	self.__json = copy.deepcopy(self.__json_base)	1✔
22	if "SampleTime" in self.__json['Info']:	1✔
23	self.__sample_time = self.__json['Info']["SampleTime"]	1✔
24	else:
25	self.__sample_time = None	1✔
26
27	def __contains__(self, key):	1✔
28	return key in self.__json	1✔
29
30	def __getitem__(self, key):	1✔
31	if key in self.__json:	1✔
32	return self.__json[key]	1✔
33	else:
34	return None	1✔
35
36	def __setitem__(self, key, value):	1✔
37	self.__json[key] = value	×
38
39	def __rebuild_json(self, models_names, minimizers):	1✔
40	models_json = subjson_from_model(self.__json, list(models_names))	1✔
41	if 'Minimizers' in self.__json and len(minimizers) > 0:	1✔
42	minimizers_json = subjson_from_minimize(self.__json, list(minimizers))	1✔
43	models_json = merge(models_json, minimizers_json)	1✔
44	return copy.deepcopy(models_json)	1✔
45
46	def recurrentInputs(self):	1✔
47	return {key:value for key, value in self.__json['Inputs'].items() if ('closedLoop' in value.keys() or 'connect' in value.keys())}	1✔
48
49	def getJson(self, models:list\|str\|None = None) -> dict:	1✔
50	if models is None:	1✔
51	return copy.deepcopy(self.__json)	1✔
52	else:
53	json = subjson_from_model(self.__json, models)	1✔
54	check_model(json)	1✔
55	return copy.deepcopy(json)	1✔
56
57	def getSampleTime(self):	1✔
58	check(self.__sample_time is not None, AttributeError, "Sample time is not defined the model is not neuralized!")	1✔
59	return self.__sample_time	1✔
60
61	def isDefined(self):	1✔
62	return self.__json is not None	1✔
63
64	def addConnection(self, stream_out:str\|Output\|Stream, input_in:str\|Input, type:str, local:bool = False):	1✔
65	outputs = self.__json['Outputs']	1✔
66
67	if isinstance(stream_out, (Output, Stream)):	1✔
68	stream_name = outputs[stream_out.name] if stream_out.name in outputs.keys() else stream_out.name	1✔
69	else:
70	output_name = check_and_get_list(stream_out, set(outputs.keys()),	1✔
71	lambda name: f"The name {name} is not part of the available Outputs")[0]
72	stream_name = outputs[output_name]	1✔
73
74	if isinstance(input_in, Input):	1✔
75	input_name = input_in.name	1✔
76	else:
77	input_name = input_in #TODO Add tests	1✔
78
79	input_name = check_and_get_list(input_name, set(self.__json['Inputs'].keys()),	1✔
80	lambda name: f"The name {name} is not part of the available Inputs")[0]
81	stream_name = check_and_get_list(stream_name, set(self.__json['Relations'].keys()),	1✔
82	lambda name: f"The name {name} is not part of the available Relations")[0]
83	self.__json['Inputs'][input_name][type] = stream_name	1✔
84	self.__json['Inputs'][input_name]['local'] = int(local)	1✔
85
86	def removeConnection(self, name_list:str\|list[str]):	1✔
87	name_list = check_and_get_list(name_list, set(self.__json['Inputs'].keys()), lambda name: f"The name {name} is not part of the available Inputs")	1✔
88	for input_in in name_list:	1✔
89	if 'closedLoop' in self.__json['Inputs'][input_in].keys():	1✔
90	del self.__json['Inputs'][input_in]['closedLoop']	1✔
91	del self.__json['Inputs'][input_in]['local']	1✔
92	elif 'connect' in self.__json['Inputs'][input_in].keys():	1✔
93	del self.__json['Inputs'][input_in]['connect']	1✔
94	del self.__json['Inputs'][input_in]['local']	1✔
95	else:
96	raise ValueError(f"The input '{input_in}' has no connection or closed loop defined")	1✔
97
98	def addModel(self, name:str, stream_list):	1✔
99	if isinstance(stream_list, Output):	1✔
100	stream_list = [stream_list]	1✔
101
102	json = MAIN_JSON	1✔
103	for stream in stream_list:	1✔
104	json = merge(json, stream.json)	1✔
105	check_model(json)	1✔
106
107	if 'Models' not in self.__json:	1✔
108	self.__json = merge(self.__json, json)	1✔
109	self.__json['Models'] = name	1✔
110	else:
111	models_names = set((self.__json['Models'],)) if type(self.__json['Models']) is str else set(self.__json['Models'].keys())	1✔
112	check_names(name, models_names, 'Models')	1✔
113	if type(self.__json['Models']) is str:	1✔
114	self.__json['Models'] = {self.__json['Models']: get_models_json(self.__json)}	1✔
115	self.__json = merge(self.__json, json)	1✔
116	self.__json['Models'][name] = get_models_json(json)	1✔
117
118	def removeModel(self, name_list):	1✔
119	if 'Models' not in self.__json:	1✔
120	raise ValueError("No Models are defined")	×
121	models_names = {self.__json['Models']} if type(self.__json['Models']) is str else set(self.__json['Models'].keys())	1✔
122	name_list = check_and_get_list(name_list, models_names, lambda name: f"The name {name} is not part of the available models")	1✔
123	models_names -= set(name_list)	1✔
124	minimizers = set(self.__json['Minimizers'].keys()) if 'Minimizers' in self.__json else None	1✔
125	self.__json = self.__rebuild_json(models_names, minimizers)	1✔
126
127	def addMinimize(self, name, streamA, streamB, loss_function='mse'):	1✔
128	if 'Minimizers' not in self.__json:	1✔
129	self.__json['Minimizers'] = {}	1✔
130	check_names(name, set(self.__json['Minimizers'].keys()), 'Minimizers')	1✔
131
132	if isinstance(streamA, str):	1✔
133	streamA_name = streamA	1✔
134	else:
135	check(isinstance(streamA, (Output, Stream)), TypeError, 'streamA must be an instance of Output or Stream')	1✔
136	streamA_name = streamA.json['Outputs'][streamA.name] if isinstance(streamA, Output) else streamA.name	1✔
137	self.__json = merge(self.__json, streamA.json)	1✔
138
139	if isinstance(streamB, str):	1✔
140	streamB_name = streamB	1✔
141	else:
142	check(isinstance(streamB, (Output, Stream)), TypeError, 'streamA must be an instance of Output or Stream')	1✔
143	streamB_name = streamB.json['Outputs'][streamB.name] if isinstance(streamB, Output) else streamB.name	1✔
144	self.__json = merge(self.__json, streamB.json)	1✔
145	#check(streamA.dim == streamB.dim, ValueError, f'Dimension of streamA={streamA.dim} and streamB={streamB.dim} are not equal.')
146
147	self.__json['Minimizers'][name] = {}	1✔
148	self.__json['Minimizers'][name]['A'] = streamA_name	1✔
149	self.__json['Minimizers'][name]['B'] = streamB_name	1✔
150	self.__json['Minimizers'][name]['loss'] = loss_function	1✔
151
152	def removeMinimize(self, name_list):	1✔
153	if 'Minimizers' not in self.__json:	1✔
154	raise ValueError("No Minimizers are defined")	×
155	name_list = check_and_get_list(name_list, self.__json['Minimizers'].keys(), lambda name: f"The name {name} is not part of the available minimizers")	1✔
156	models_names = {self.__json['Models']} if type(self.__json['Models']) is str else set(self.__json['Models'].keys())	1✔
157	remaining_minimizers = set(self.__json['Minimizers'].keys()) - set(name_list) if 'Minimizers' in self.__json else None	1✔
158	self.__json = self.__rebuild_json(models_names, remaining_minimizers)	1✔
159
160	def setBuildWindow(self, sample_time = None):	1✔
161	check(self.__json is not None, RuntimeError, "No model is defined!")	1✔
162	if sample_time is not None:	1✔
163	check(sample_time > 0, RuntimeError, 'Sample time must be strictly positive!')	1✔
164	self.__sample_time = sample_time	1✔
165	else:
166	if self.__sample_time is None:	1✔
167	self.__sample_time = 1	1✔
168
169	self.__json['Info'] = {"SampleTime": self.__sample_time}	1✔
170	if 'SampleTime' in self.__json['Constants']:	1✔
171	self.__json['Constants']['SampleTime'] = {'dim': 1, 'values': self.__sample_time}	1✔
172
173	check(self.__json['Inputs'] != {}, RuntimeError, "No model is defined!")	1✔
174	json_inputs = self.__json['Inputs']	1✔
175
176	input_ns_backward, input_ns_forward = {}, {}	1✔
177	for key, value in json_inputs.items():	1✔
178	if 'sw' not in value and 'tw' not in value:	1✔
179	assert False, f"Input '{key}' has no time window or sample window"	×
180	if 'sw' not in value and self.__sample_time is not None:	1✔
181	## check if value['tw'] is a multiple of sample_time
182	absolute_tw = abs(value['tw'][0]) + abs(value['tw'][1])	1✔
183	check(round(absolute_tw % self.__sample_time) == 0, ValueError,	1✔
184	f"Time window of input '{key}' is not a multiple of sample time. This network cannot be neuralized")
185	input_ns_backward[key] = round(-value['tw'][0] / self.__sample_time)	1✔
186	input_ns_forward[key] = round(value['tw'][1] / self.__sample_time)	1✔
187	elif self.__sample_time is not None:	1✔
188	if 'tw' in value:	1✔
189	input_ns_backward[key] = max(round(-value['tw'][0] / self.__sample_time), -value['sw'][0])	1✔
190	input_ns_forward[key] = max(round(value['tw'][1] / self.__sample_time), value['sw'][1])	1✔
191	else:
192	input_ns_backward[key] = -value['sw'][0]	1✔
193	input_ns_forward[key] = value['sw'][1]	1✔
194	else:
195	check(value['tw'] == [0,0], RuntimeError, f"Sample time is not defined for input '{key}'")	×
196	input_ns_backward[key] = -value['sw'][0]	×
197	input_ns_forward[key] = value['sw'][1]	×
198	value['ns'] = [input_ns_backward[key], input_ns_forward[key]]	1✔
199	value['ntot'] = sum(value['ns'])	1✔
200
201	self.__json['Info']['ns'] = [max(input_ns_backward.values()), max(input_ns_forward.values())]	1✔
202	self.__json['Info']['ntot'] = sum(self.__json['Info']['ns'])	1✔
203	if self.__json['Info']['ns'][0] < 0:	1✔
204	log.warning(	1✔
205	f"The input is only in the far past the max_samples_backward is: {self.__json['Info']['ns'][0]}")
206	if self.__json['Info']['ns'][1] < 0:	1✔
207	log.warning(	1✔
208	f"The input is only in the far future the max_sample_forward is: {self.__json['Info']['ns'][1]}")
209
210	for k, v in (self.__json['Parameters'] \| self.__json['Constants']).items():	1✔
211	if 'values' in v:	1✔
212	window = 'tw' if 'tw' in v.keys() else ('sw' if 'sw' in v.keys() else None)	1✔
213	if window == 'tw':	1✔
214	check(np.array(v['values']).shape[0] == v['tw'] / self.__sample_time, ValueError,	1✔
215	f"{k} has a different number of values for this sample time.")
216	if v['values'] == "SampleTime":	1✔
UNCOV 217	v['values'] = self.__sample_time	×
218
219	def updateParameters(self, model = None, *, clear_model = False):	1✔
220	if clear_model:	1✔
221	for key in self.__json['Parameters'].keys():	1✔
222	if 'init_values' in self.__json['Parameters'][key]:	1✔
223	self.__json['Parameters'][key]['values'] = self.__json['Parameters'][key]['init_values']	1✔
224	elif 'values' in self.__json['Parameters'][key]:	1✔
225	del self.__json['Parameters'][key]['values']	1✔
226	elif model is not None:	1✔
227	for key in self.__json['Parameters'].keys():	1✔
228	if key in model.all_parameters:	1✔
229	self.__json['Parameters'][key]['values'] = model.all_parameters[key].tolist()	1✔
230

tonegas / nnodely / 20075618707

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