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

94.67

/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_relation, check_model, get_models_json
from nnodely.basic.relation import MAIN_JSON, Stream, check_names
from nnodely.layers.output import Output

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):
        new_json = subjson_from_model(self.__json, list(models_names))

        if 'Minimizers' in self.__json:
            rel_A = [self.__json['Minimizers'][key]['A'] for key in minimizers]
            rel_B = [self.__json['Minimizers'][key]['B'] for key in minimizers]
            relations_name = set(rel_A) | set(rel_B)
            if len(relations_name) != 0:
                minimizers_json = subjson_from_relation(self.__json, list(relations_name))
                new_json = merge(new_json, minimizers_json)

        return copy.deepcopy(new_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 addConnect(self, stream_name:str, input_name:str, local:bool = False):
        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]['connect'] = stream_name
        self.__json['Inputs'][input_name]['local'] = int(local)

    def addClosedLoop(self, stream_name:str, input_name:str, local:bool = False):
        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]['closedLoop'] = 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}

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

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