13775554764

Committed 10 Mar 2025 09:56PM UTC coverage: 96.46% (+0.4%) from 96.036%

Build # 13775554764

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #69 from tonegas/develop

Develop

Coverage Stats

861 of 868 new or added lines in 22 files covered. (99.19%)

10901 of 11301 relevant lines covered (96.46%)

0.96 hits per line

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

97.96

/nnodely/parametricfunction.py

import inspect, copy, textwrap, torch, math

import torch.nn as nn
import numpy as np

from typing import Union
from collections.abc import Callable

from nnodely.relation import NeuObj, Stream, toStream
from nnodely.model import Model
from nnodely.parameter import Parameter, Constant
from nnodely.utils import check, merge, enforce_types

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


paramfun_relation_name = 'ParamFun'

class ParamFun(NeuObj):
    """
    Represents a parametric function in the neural network model.

    Parameters
    ----------
    param_fun : Callable
        The parametric function to be used.
    constants : list or dict or None, optional
        A list or dictionary of constants to be used in the function. Default is None.
    parameters_dimensions : list or dict or None, optional
        A list or dictionary specifying the dimensions of the parameters. Default is None.
    parameters : list or dict or None, optional
        A list or dictionary of parameters to be used in the function. Default is None.
    map_over_batch : bool, optional
        A boolean indicating whether to map the function over the batch dimension. Default is False.

    Attributes
    ----------
    relation_name : str
        The name of the relation.
    param_fun : Callable
        The parametric function to be used.
    constants : list or dict or None
        A list or dictionary of constants to be used in the function.
    parameters_dimensions : list or dict or None
        A list or dictionary specifying the dimensions of the parameters.
    parameters : list or dict or None
        A list or dictionary of parameters to be used in the function.
    map_over_batch : bool
        A boolean indicating whether to map the function over the batch dimension.
    output_dimension : dict
        A dictionary containing the output dimensions of the function.
    json : dict
        A dictionary containing the configuration of the function.

    Example
    -------
        >>> input1 = Input('input1')
        >>> input2 = Input('input2')

        >>> def my_function(x, y, param1, const1):
        >>>     return param1 * x + const1 * y

        >>> param_fun = ParamFun(my_function, constants={'const1': 1.0}, parameters_dimensions={'param1': 1})
        >>> result = param_fun(input1, input2)
    """
    @enforce_types
    def __init__(self, param_fun:Callable,
                 parameters_and_constants:list|dict|None = None,
                 map_over_batch:bool = False) -> Stream:

        self.relation_name = paramfun_relation_name

        # input parameters
        self.param_fun = param_fun
        self.parameters_and_constants = parameters_and_constants
        self.map_over_batch = map_over_batch

        self.output_dimension = {}
        super().__init__('F'+paramfun_relation_name + str(NeuObj.count))
        code = textwrap.dedent(inspect.getsource(param_fun)).replace('\"', '\'')
        self.json['Functions'][self.name] = {
            'code' : code,
            'name' : param_fun.__name__,
        }
        self.json['Functions'][self.name]['params_and_consts'] = []

        funinfo = inspect.getfullargspec(self.param_fun)

        # Create the parameters and constants from list
        if type(self.parameters_and_constants) is list:
            n_pc = len(self.parameters_and_constants)
            n_input = len(funinfo.args)
            for pc, pc_name in zip(self.parameters_and_constants,funinfo.args[n_input-n_pc:]):
                self.__create_parameter(pc, pc_name)

        # Create the parameters and constants from list
        first = False
        if type(self.parameters_and_constants) is dict:
            for i, key in enumerate(funinfo.args):
                if key in self.parameters_and_constants:
                    first = True
                    pc = self.parameters_and_constants[key]
                    self.__create_parameter(pc, key)
                elif first == True:
                    p = Parameter(name=self.name + key, dimensions=1)
                    self.json['Functions'][self.name]['params_and_consts'].append(p.name)
                    self.json = merge(self.json, p.json)

        self.json_stream = {}

    @enforce_types
    def __call__(self, *obj:Union[Stream|Parameter|Constant|float|int]) -> Stream:
        stream_name = paramfun_relation_name + str(Stream.count)

        funinfo = inspect.getfullargspec(self.param_fun)
        n_function_input = len(funinfo.args)
        n_call_input = len(obj)
        n_parameters = n_function_input - n_call_input

        input_dimensions = []
        input_types = []
        for ind, o in enumerate(obj):
            if type(o) in (int, float, list):
                obj_type = Constant
            else:
                obj_type = type(o)
            o = toStream(o)
            check(type(o) is Stream, TypeError,
                  f"The type of {o} is {type(o)} and is not supported for ParamFun operation.")
            input_types.append(obj_type)
            input_dimensions.append(o.dim)

        if n_call_input not in self.json_stream:
            if len(self.json_stream) > 0:
                log.warning(f"The function {self.name} was called with a different number of inputs. If both functions enter in the model an error will be raised.")

            self.json_stream[n_call_input] = copy.deepcopy(self.json)
            self.json_stream[n_call_input]['Functions'][self.name]['n_input'] = n_call_input

            # Create the missing parameters
            n_created_parameters = len(self.json_stream[n_call_input]['Functions'][self.name]['params_and_consts'])
            n_missing_parameters = n_parameters - n_created_parameters
            check(n_missing_parameters >= 0, ValueError, f"The function is called with too many parameter and inputs.")
            self.__create_missing_parameters(self.json_stream[n_call_input], n_call_input, n_missing_parameters)

            self.json_stream[n_call_input]['Functions'][self.name]['in_dim'] = copy.deepcopy(input_dimensions)
            self.json_stream[n_call_input]['Functions'][self.name]['map_over_dim'] = self.__infer_map_over_batch(input_types, n_parameters)
            output_dimension = self.__infer_output_dimensions(self.json_stream[n_call_input], input_types, input_dimensions)
        else:
            map_over_batch = self.__infer_map_over_batch(input_types, n_parameters)
            check(map_over_batch == self.json_stream[n_call_input]['Functions'][self.name]['map_over_dim'], ValueError, f"The function {self.name} was called with different type of input using map_over_batch=True.")
            output_dimension = self.__infer_output_dimensions(self.json_stream[n_call_input], input_types, input_dimensions)

            # Save the all the input dimension used for call the parametric function
            in_dim = self.json_stream[n_call_input]['Functions'][self.name]['in_dim']
            if type(in_dim[0]) is dict:
                if in_dim != input_dimensions:
                    in_dim = [in_dim, input_dimensions]
                    log.warning(f"The function {self.name} was called with inputs with different dimensions.")
            elif input_dimensions not in in_dim:
                in_dim.append(input_dimensions)
                log.warning(f"The function {self.name} was called with inputs with different dimensions.")
            self.json_stream[n_call_input]['Functions'][self.name]['in_dim'] = in_dim

        stream_json = copy.deepcopy(self.json_stream[n_call_input])
        input_names = []
        for ind, o in enumerate(obj):
            o = toStream(o)
            check(type(o) is Stream, TypeError,
                  f"The type of {o} is {type(o)} and is not supported for ParamFun operation.")
            stream_json = merge(stream_json, o.json)
            input_names.append(o.name)

        stream_json['Relations'][stream_name] = [paramfun_relation_name, input_names, self.name]
        return Stream(stream_name, stream_json, output_dimension)

    def __create_parameter(self, pc, pc_name):
        if type(pc) is Parameter:
            self.json['Functions'][self.name]['params_and_consts'].append(pc.name)
            self.json = merge(self.json, pc.json)
        elif type(pc) is str:
            # TODO to remove! there is no reason to give a name to the parameter. The name of the parameter is the name of the function parameter
            p = Parameter(name=pc, dimensions=1)
            self.json['Functions'][self.name]['params_and_consts'].append(p.name)
            self.json = merge(self.json, p.json)
        elif type(pc) is tuple:
            p = Parameter(name=self.name + pc_name, dimensions=list(pc))
            self.json['Functions'][self.name]['params_and_consts'].append(p.name)
            self.json = merge(self.json, p.json)
        elif type(pc) is Constant:
            self.json['Functions'][self.name]['params_and_consts'].append(pc.name)
            self.json = merge(self.json, pc.json)
        elif type(pc) in (float, int, list):
            c = Constant(name=self.name + pc_name, values=pc)
            self.json['Functions'][self.name]['params_and_consts'].append(c.name)
            self.json = merge(self.json, c.json)
        else:
            check(type(pc) in (Parameter, str, tuple, Constant, float, int, list), TypeError,
                  f'The element inside the \"parameters_and_constants\" list or dict must be a Parameter, str, tuple to build a Parameter or Constant, int, float or list to build a Constant but was {type(pc)}.')

    def __infer_map_over_batch(self, input_types, n_constants_and_params):
        input_map_dim = ()

        for elem in input_types:
            if elem in (Parameter, Constant):
                input_map_dim += (None,)
            else:
                input_map_dim += (0,)

        for i in range(n_constants_and_params):
            input_map_dim += (None,)

        if self.map_over_batch:
            return list(input_map_dim)
        else:
            return False

    def __create_missing_parameters(self, stream_json, n_call_input, n_missing_parameters):
        funinfo = inspect.getfullargspec(self.param_fun)
        for i in range(n_missing_parameters):
            p_name = self.name + funinfo.args[n_call_input+i]
            stream_json['Functions'][self.name]['params_and_consts'].insert(i, p_name)
            stream_json['Parameters'][p_name] = {'dim': 1}

    def __infer_output_dimensions(self, stream_json, input_types, input_dimensions):
        import torch
        batch_dim = 5

        all_inputs_dim = copy.deepcopy(input_dimensions)
        all_inputs_type = copy.deepcopy(input_types)
        params_and_consts = stream_json['Constants'] | stream_json['Parameters']
        for name in stream_json['Functions'][self.name]['params_and_consts']:
            all_inputs_dim.append(params_and_consts[name])
            all_inputs_type.append(Constant)

        n_samples_sec = 0.1
        is_int = False
        while is_int == False:
            n_samples_sec *= 10
            vect_input_time = [math.isclose(d['tw']*n_samples_sec,round(d['tw']*n_samples_sec)) for d in all_inputs_dim if 'tw' in d]
            if len(vect_input_time) == 0:
                is_int = True
            else:
                is_int = sum(vect_input_time) == len(vect_input_time)

        # Build input with right dimensions
        inputs = []
        inputs_win_type = []
        inputs_win = []

        for t, dim in zip(all_inputs_type,all_inputs_dim):
            window = 'tw' if 'tw' in dim else ('sw' if 'sw' in dim else None)
            if window == 'tw':
                dim_win = round(dim[window] * n_samples_sec)
            elif window == 'sw':
                dim_win = dim[window]
            else:
                dim_win = None if t in (Parameter, Constant) else 1
            if t in (Parameter, Constant):
                if type(dim['dim']) is list:
                    if dim_win is not None:
                        inputs.append(torch.rand(size=(dim_win,) + tuple(dim['dim'])))
                    else:
                        inputs.append(torch.rand(size=tuple(dim['dim'])))
                else:
                    if dim_win is not None:
                        inputs.append(torch.rand(size=(dim_win, dim['dim'])))
                    else:
                        inputs.append(torch.rand(size=(dim['dim'],)))
            else:
                inputs.append(torch.rand(size=(batch_dim, dim_win, dim['dim'])))

            inputs_win_type.append(window)
            inputs_win.append(dim_win)

        if self.map_over_batch:
            function_to_call = torch.func.vmap(self.param_fun,in_dims=tuple(stream_json['Functions'][self.name]['map_over_dim']))
        else:
            function_to_call = self.param_fun
        out = function_to_call(*inputs)
        out_shape = out.shape
        check(out_shape[0] == batch_dim, ValueError, "The batch output dimension it is not correct.")
        out_dim = list(out_shape[2:])
        check(len(out_dim) == 1, ValueError, "The output dimension of the function is bigger than a vector.")
        out_win_type = 'sw'
        out_win = out_shape[1]
        for idx, win in enumerate(inputs_win):
            if out_shape[1] == win and all_inputs_type[idx] not in (Parameter, Constant):
                out_win_type = inputs_win_type[idx]
                out_win = all_inputs_dim[idx][out_win_type]

        return { 'dim': out_dim[0], out_win_type : out_win }

def return_standard_inputs(json, model_def, xlim = None, num_points = 1000):
    check(json['n_input'] == 1 or json['n_input'] == 2, ValueError, "The function must have only one or two inputs.")
    fun_inputs = tuple()
    for i in range(json['n_input']):
        dim = json['in_dim'][i]
        check(dim['dim'] == 1, ValueError, "The input dimension must be 1.")
        if 'tw' in dim:
            check(dim['tw'] == model_def['Info']['SampleTime'], ValueError, "The input window must be 1.")
        elif 'sw' in dim:
            check(dim['sw'] == 1, ValueError, "The input window must be 1.")
        if xlim is not None:
            if json['n_input'] == 2:
                check(np.array(xlim).shape == (json['n_input'], 2), ValueError,
                      "The xlim must have the same shape as the number of inputs.")
                x_value = np.linspace(xlim[i][0], xlim[i][1], num=num_points)
            else:
                check(np.array(xlim).shape == (2,), ValueError,
                      "The xlim must have the same shape as the number of inputs.")
                x_value = np.linspace(xlim[0], xlim[1], num=num_points)
        else:
            x_value = np.linspace(0, 1, num=num_points)
        if i == 0:
            x0_value = torch.from_numpy(x_value)
        else:
            x1_value = torch.from_numpy(x_value)

    if json['n_input'] == 2:
        x0_value, x1_value = torch.meshgrid(x0_value,x1_value,indexing="xy")
        x0_value = x0_value.flatten().unsqueeze(1).unsqueeze(1)
        x1_value = x1_value.flatten().unsqueeze(1).unsqueeze(1)
        fun_inputs += (x0_value,x1_value,)
    else:
        x0_value = x0_value.unsqueeze(1).unsqueeze(1)
        fun_inputs += (x0_value,)

    for key in json['params_and_consts']:
        val = model_def['Parameters'][key] if key in model_def['Parameters'] else model_def['Constants'][key]
        fun_inputs += tuple([torch.from_numpy(np.array(val['values']))]) # The vector is transform in a tuple

    return fun_inputs

def return_function(json, fun_inputs):
    exec(json['code'], globals())
    function_to_call = globals()[json['name']]
    output = function_to_call(*fun_inputs)
    check(output.shape[1] == 1, ValueError, "The output dimension must be 1.")
    check(output.shape[2] == 1, ValueError, "The output window must be 1.")
    funinfo = inspect.getfullargspec(function_to_call)
    return output, funinfo.args

class Parametric_Layer(nn.Module):
    def __init__(self, func, params_and_consts, map_over_batch):
        super().__init__()
        self.name = func['name']
        self.params_and_consts = params_and_consts
        if type(map_over_batch) is list:
            self.map_over_batch = True
            self.input_map_dim = tuple(map_over_batch)
        else:
            self.map_over_batch = False
        ## Add the function to the globals
        try:
            code = 'import torch\n@torch.fx.wrap\n' + func['code']
            exec(code, globals())
        except Exception as e:
            print(f"An error occurred: {e}")

    def forward(self, *inputs):
        args = list(inputs) + self.params_and_consts
        # Retrieve the function object from the globals dictionary
        function_to_call = globals()[self.name]
        # Call the function using the retrieved function object
        if self.map_over_batch:
            function_to_call = torch.func.vmap(function_to_call,in_dims=self.input_map_dim)
        result = function_to_call(*args)
        return result

def createParamFun(self, *func_params):
    return Parametric_Layer(func=func_params[0], params_and_consts=func_params[1], map_over_batch=func_params[2])

setattr(Model, paramfun_relation_name, createParamFun)

1	import inspect, copy, textwrap, torch, math	1✔
2
3	import torch.nn as nn	1✔
4	import numpy as np	1✔
5
6	from typing import Union	1✔
7	from collections.abc import Callable	1✔
8
9	from nnodely.relation import NeuObj, Stream, toStream	1✔
10	from nnodely.model import Model	1✔
11	from nnodely.parameter import Parameter, Constant	1✔
12	from nnodely.utils import check, merge, enforce_types	1✔
13
14	from nnodely.logger import logging, nnLogger	1✔
15	log = nnLogger(__name__, logging.WARNING)	1✔
16
17
18	paramfun_relation_name = 'ParamFun'	1✔
19
20	class ParamFun(NeuObj):	1✔
21	"""
22	Represents a parametric function in the neural network model.
23
24	Parameters
25	----------
26	param_fun : Callable
27	The parametric function to be used.
28	constants : list or dict or None, optional
29	A list or dictionary of constants to be used in the function. Default is None.
30	parameters_dimensions : list or dict or None, optional
31	A list or dictionary specifying the dimensions of the parameters. Default is None.
32	parameters : list or dict or None, optional
33	A list or dictionary of parameters to be used in the function. Default is None.
34	map_over_batch : bool, optional
35	A boolean indicating whether to map the function over the batch dimension. Default is False.
36
37	Attributes
38	----------
39	relation_name : str
40	The name of the relation.
41	param_fun : Callable
42	The parametric function to be used.
43	constants : list or dict or None
44	A list or dictionary of constants to be used in the function.
45	parameters_dimensions : list or dict or None
46	A list or dictionary specifying the dimensions of the parameters.
47	parameters : list or dict or None
48	A list or dictionary of parameters to be used in the function.
49	map_over_batch : bool
50	A boolean indicating whether to map the function over the batch dimension.
51	output_dimension : dict
52	A dictionary containing the output dimensions of the function.
53	json : dict
54	A dictionary containing the configuration of the function.
55
56	Example
57	-------
58	>>> input1 = Input('input1')
59	>>> input2 = Input('input2')
60
61	>>> def my_function(x, y, param1, const1):
62	>>> return param1 * x + const1 * y
63
64	>>> param_fun = ParamFun(my_function, constants={'const1': 1.0}, parameters_dimensions={'param1': 1})
65	>>> result = param_fun(input1, input2)
66	"""
67	@enforce_types	1✔
68	def __init__(self, param_fun:Callable,	1✔
69	parameters_and_constants:list\|dict\|None = None,
70	map_over_batch:bool = False) -> Stream:
71
72	self.relation_name = paramfun_relation_name	1✔
73
74	# input parameters
75	self.param_fun = param_fun	1✔
76	self.parameters_and_constants = parameters_and_constants	1✔
77	self.map_over_batch = map_over_batch	1✔
78
79	self.output_dimension = {}	1✔
80	super().__init__('F'+paramfun_relation_name + str(NeuObj.count))	1✔
81	code = textwrap.dedent(inspect.getsource(param_fun)).replace('\"', '\'')	1✔
82	self.json['Functions'][self.name] = {	1✔
83	'code' : code,
84	'name' : param_fun.__name__,
85	}
86	self.json['Functions'][self.name]['params_and_consts'] = []	1✔
87
88	funinfo = inspect.getfullargspec(self.param_fun)	1✔
89
90	# Create the parameters and constants from list
91	if type(self.parameters_and_constants) is list:	1✔
92	n_pc = len(self.parameters_and_constants)	1✔
93	n_input = len(funinfo.args)	1✔
94	for pc, pc_name in zip(self.parameters_and_constants,funinfo.args[n_input-n_pc:]):	1✔
95	self.__create_parameter(pc, pc_name)	1✔
96
97	# Create the parameters and constants from list
98	first = False	1✔
99	if type(self.parameters_and_constants) is dict:	1✔
100	for i, key in enumerate(funinfo.args):	1✔
101	if key in self.parameters_and_constants:	1✔
102	first = True	1✔
103	pc = self.parameters_and_constants[key]	1✔
104	self.__create_parameter(pc, key)	1✔
105	elif first == True:	1✔
106	p = Parameter(name=self.name + key, dimensions=1)	1✔
107	self.json['Functions'][self.name]['params_and_consts'].append(p.name)	1✔
108	self.json = merge(self.json, p.json)	1✔
109
110	self.json_stream = {}	1✔
111
112	@enforce_types	1✔
113	def __call__(self, *obj:Union[Stream\|Parameter\|Constant\|float\|int]) -> Stream:	1✔
114	stream_name = paramfun_relation_name + str(Stream.count)	1✔
115
116	funinfo = inspect.getfullargspec(self.param_fun)	1✔
117	n_function_input = len(funinfo.args)	1✔
118	n_call_input = len(obj)	1✔
119	n_parameters = n_function_input - n_call_input	1✔
120
121	input_dimensions = []	1✔
122	input_types = []	1✔
123	for ind, o in enumerate(obj):	1✔
124	if type(o) in (int, float, list):	1✔
125	obj_type = Constant	1✔
126	else:
127	obj_type = type(o)	1✔
128	o = toStream(o)	1✔
129	check(type(o) is Stream, TypeError,	1✔
130	f"The type of {o} is {type(o)} and is not supported for ParamFun operation.")
131	input_types.append(obj_type)	1✔
132	input_dimensions.append(o.dim)	1✔
133
134	if n_call_input not in self.json_stream:	1✔
135	if len(self.json_stream) > 0:	1✔
136	log.warning(f"The function {self.name} was called with a different number of inputs. If both functions enter in the model an error will be raised.")	1✔
137
138	self.json_stream[n_call_input] = copy.deepcopy(self.json)	1✔
139	self.json_stream[n_call_input]['Functions'][self.name]['n_input'] = n_call_input	1✔
140
141	# Create the missing parameters
142	n_created_parameters = len(self.json_stream[n_call_input]['Functions'][self.name]['params_and_consts'])	1✔
143	n_missing_parameters = n_parameters - n_created_parameters	1✔
144	check(n_missing_parameters >= 0, ValueError, f"The function is called with too many parameter and inputs.")	1✔
145	self.__create_missing_parameters(self.json_stream[n_call_input], n_call_input, n_missing_parameters)	1✔
146
147	self.json_stream[n_call_input]['Functions'][self.name]['in_dim'] = copy.deepcopy(input_dimensions)	1✔
148	self.json_stream[n_call_input]['Functions'][self.name]['map_over_dim'] = self.__infer_map_over_batch(input_types, n_parameters)	1✔
149	output_dimension = self.__infer_output_dimensions(self.json_stream[n_call_input], input_types, input_dimensions)	1✔
150	else:
151	map_over_batch = self.__infer_map_over_batch(input_types, n_parameters)	1✔
152	check(map_over_batch == self.json_stream[n_call_input]['Functions'][self.name]['map_over_dim'], ValueError, f"The function {self.name} was called with different type of input using map_over_batch=True.")	1✔
153	output_dimension = self.__infer_output_dimensions(self.json_stream[n_call_input], input_types, input_dimensions)	1✔
154
155	# Save the all the input dimension used for call the parametric function
156	in_dim = self.json_stream[n_call_input]['Functions'][self.name]['in_dim']	1✔
157	if type(in_dim[0]) is dict:	1✔
158	if in_dim != input_dimensions:	1✔
159	in_dim = [in_dim, input_dimensions]	1✔
160	log.warning(f"The function {self.name} was called with inputs with different dimensions.")	1✔
161	elif input_dimensions not in in_dim:	1✔
162	in_dim.append(input_dimensions)	1✔
163	log.warning(f"The function {self.name} was called with inputs with different dimensions.")	1✔
164	self.json_stream[n_call_input]['Functions'][self.name]['in_dim'] = in_dim	1✔
165
166	stream_json = copy.deepcopy(self.json_stream[n_call_input])	1✔
167	input_names = []	1✔
168	for ind, o in enumerate(obj):	1✔
169	o = toStream(o)	1✔
170	check(type(o) is Stream, TypeError,	1✔
171	f"The type of {o} is {type(o)} and is not supported for ParamFun operation.")
172	stream_json = merge(stream_json, o.json)	1✔
173	input_names.append(o.name)	1✔
174
175	stream_json['Relations'][stream_name] = [paramfun_relation_name, input_names, self.name]	1✔
176	return Stream(stream_name, stream_json, output_dimension)	1✔
177
178	def __create_parameter(self, pc, pc_name):	1✔
179	if type(pc) is Parameter:	1✔
180	self.json['Functions'][self.name]['params_and_consts'].append(pc.name)	1✔
181	self.json = merge(self.json, pc.json)	1✔
182	elif type(pc) is str:	1✔
183	# TODO to remove! there is no reason to give a name to the parameter. The name of the parameter is the name of the function parameter
184	p = Parameter(name=pc, dimensions=1)	1✔
185	self.json['Functions'][self.name]['params_and_consts'].append(p.name)	1✔
186	self.json = merge(self.json, p.json)	1✔
187	elif type(pc) is tuple:	1✔
188	p = Parameter(name=self.name + pc_name, dimensions=list(pc))	1✔
189	self.json['Functions'][self.name]['params_and_consts'].append(p.name)	1✔
190	self.json = merge(self.json, p.json)	1✔
191	elif type(pc) is Constant:	1✔
192	self.json['Functions'][self.name]['params_and_consts'].append(pc.name)	1✔
193	self.json = merge(self.json, pc.json)	1✔
194	elif type(pc) in (float, int, list):	1✔
195	c = Constant(name=self.name + pc_name, values=pc)	1✔
196	self.json['Functions'][self.name]['params_and_consts'].append(c.name)	1✔
197	self.json = merge(self.json, c.json)	1✔
198	else:
NEW 199	check(type(pc) in (Parameter, str, tuple, Constant, float, int, list), TypeError,	×
200	f'The element inside the \"parameters_and_constants\" list or dict must be a Parameter, str, tuple to build a Parameter or Constant, int, float or list to build a Constant but was {type(pc)}.')
201
202	def __infer_map_over_batch(self, input_types, n_constants_and_params):	1✔
203	input_map_dim = ()	1✔
204
205	for elem in input_types:	1✔
206	if elem in (Parameter, Constant):	1✔
207	input_map_dim += (None,)	1✔
208	else:
209	input_map_dim += (0,)	1✔
210
211	for i in range(n_constants_and_params):	1✔
212	input_map_dim += (None,)	1✔
213
214	if self.map_over_batch:	1✔
215	return list(input_map_dim)	1✔
216	else:
217	return False	1✔
218
219	def __create_missing_parameters(self, stream_json, n_call_input, n_missing_parameters):	1✔
220	funinfo = inspect.getfullargspec(self.param_fun)	1✔
221	for i in range(n_missing_parameters):	1✔
222	p_name = self.name + funinfo.args[n_call_input+i]	1✔
223	stream_json['Functions'][self.name]['params_and_consts'].insert(i, p_name)	1✔
224	stream_json['Parameters'][p_name] = {'dim': 1}	1✔
225
226	def __infer_output_dimensions(self, stream_json, input_types, input_dimensions):	1✔
227	import torch	1✔
228	batch_dim = 5	1✔
229
230	all_inputs_dim = copy.deepcopy(input_dimensions)	1✔
231	all_inputs_type = copy.deepcopy(input_types)	1✔
232	params_and_consts = stream_json['Constants'] \| stream_json['Parameters']	1✔
233	for name in stream_json['Functions'][self.name]['params_and_consts']:	1✔
234	all_inputs_dim.append(params_and_consts[name])	1✔
235	all_inputs_type.append(Constant)	1✔
236
237	n_samples_sec = 0.1	1✔
238	is_int = False	1✔
239	while is_int == False:	1✔
240	n_samples_sec *= 10	1✔
241	vect_input_time = [math.isclose(d['tw']n_samples_sec,round(d['tw']n_samples_sec)) for d in all_inputs_dim if 'tw' in d]	1✔
242	if len(vect_input_time) == 0:	1✔
243	is_int = True	1✔
244	else:
245	is_int = sum(vect_input_time) == len(vect_input_time)	1✔
246
247	# Build input with right dimensions
248	inputs = []	1✔
249	inputs_win_type = []	1✔
250	inputs_win = []	1✔
251
252	for t, dim in zip(all_inputs_type,all_inputs_dim):	1✔
253	window = 'tw' if 'tw' in dim else ('sw' if 'sw' in dim else None)	1✔
254	if window == 'tw':	1✔
255	dim_win = round(dim[window] * n_samples_sec)	1✔
256	elif window == 'sw':	1✔
257	dim_win = dim[window]	1✔
258	else:
259	dim_win = None if t in (Parameter, Constant) else 1	1✔
260	if t in (Parameter, Constant):	1✔
261	if type(dim['dim']) is list:	1✔
262	if dim_win is not None:	1✔
263	inputs.append(torch.rand(size=(dim_win,) + tuple(dim['dim'])))	1✔
264	else:
265	inputs.append(torch.rand(size=tuple(dim['dim'])))	1✔
266	else:
267	if dim_win is not None:	1✔
268	inputs.append(torch.rand(size=(dim_win, dim['dim'])))	1✔
269	else:
270	inputs.append(torch.rand(size=(dim['dim'],)))	1✔
271	else:
272	inputs.append(torch.rand(size=(batch_dim, dim_win, dim['dim'])))	1✔
273
274	inputs_win_type.append(window)	1✔
275	inputs_win.append(dim_win)	1✔
276
277	if self.map_over_batch:	1✔
278	function_to_call = torch.func.vmap(self.param_fun,in_dims=tuple(stream_json['Functions'][self.name]['map_over_dim']))	1✔
279	else:
280	function_to_call = self.param_fun	1✔
281	out = function_to_call(*inputs)	1✔
282	out_shape = out.shape	1✔
283	check(out_shape[0] == batch_dim, ValueError, "The batch output dimension it is not correct.")	1✔
284	out_dim = list(out_shape[2:])	1✔
285	check(len(out_dim) == 1, ValueError, "The output dimension of the function is bigger than a vector.")	1✔
286	out_win_type = 'sw'	1✔
287	out_win = out_shape[1]	1✔
288	for idx, win in enumerate(inputs_win):	1✔
289	if out_shape[1] == win and all_inputs_type[idx] not in (Parameter, Constant):	1✔
290	out_win_type = inputs_win_type[idx]	1✔
291	out_win = all_inputs_dim[idx][out_win_type]	1✔
292
293	return { 'dim': out_dim[0], out_win_type : out_win }	1✔
294
295	def return_standard_inputs(json, model_def, xlim = None, num_points = 1000):	1✔
296	check(json['n_input'] == 1 or json['n_input'] == 2, ValueError, "The function must have only one or two inputs.")	1✔
297	fun_inputs = tuple()	1✔
298	for i in range(json['n_input']):	1✔
299	dim = json['in_dim'][i]	1✔
300	check(dim['dim'] == 1, ValueError, "The input dimension must be 1.")	1✔
301	if 'tw' in dim:	1✔
302	check(dim['tw'] == model_def['Info']['SampleTime'], ValueError, "The input window must be 1.")	1✔
303	elif 'sw' in dim:	1✔
304	check(dim['sw'] == 1, ValueError, "The input window must be 1.")	1✔
305	if xlim is not None:	1✔
306	if json['n_input'] == 2:	1✔
307	check(np.array(xlim).shape == (json['n_input'], 2), ValueError,	1✔
308	"The xlim must have the same shape as the number of inputs.")
309	x_value = np.linspace(xlim[i][0], xlim[i][1], num=num_points)	1✔
310	else:
311	check(np.array(xlim).shape == (2,), ValueError,	×
312	"The xlim must have the same shape as the number of inputs.")
313	x_value = np.linspace(xlim[0], xlim[1], num=num_points)	×
314	else:
315	x_value = np.linspace(0, 1, num=num_points)	1✔
316	if i == 0:	1✔
317	x0_value = torch.from_numpy(x_value)	1✔
318	else:
319	x1_value = torch.from_numpy(x_value)	1✔
320
321	if json['n_input'] == 2:	1✔
322	x0_value, x1_value = torch.meshgrid(x0_value,x1_value,indexing="xy")	1✔
323	x0_value = x0_value.flatten().unsqueeze(1).unsqueeze(1)	1✔
324	x1_value = x1_value.flatten().unsqueeze(1).unsqueeze(1)	1✔
325	fun_inputs += (x0_value,x1_value,)	1✔
326	else:
327	x0_value = x0_value.unsqueeze(1).unsqueeze(1)	1✔
328	fun_inputs += (x0_value,)	1✔
329
330	for key in json['params_and_consts']:	1✔
331	val = model_def['Parameters'][key] if key in model_def['Parameters'] else model_def['Constants'][key]	1✔
332	fun_inputs += tuple([torch.from_numpy(np.array(val['values']))]) # The vector is transform in a tuple	1✔
333
334	return fun_inputs	1✔
335
336	def return_function(json, fun_inputs):	1✔
337	exec(json['code'], globals())	1✔
338	function_to_call = globals()[json['name']]	1✔
339	output = function_to_call(*fun_inputs)	1✔
340	check(output.shape[1] == 1, ValueError, "The output dimension must be 1.")	1✔
341	check(output.shape[2] == 1, ValueError, "The output window must be 1.")	1✔
342	funinfo = inspect.getfullargspec(function_to_call)	1✔
343	return output, funinfo.args	1✔
344
345	class Parametric_Layer(nn.Module):	1✔
346	def __init__(self, func, params_and_consts, map_over_batch):	1✔
347	super().__init__()	1✔
348	self.name = func['name']	1✔
349	self.params_and_consts = params_and_consts	1✔
350	if type(map_over_batch) is list:	1✔
351	self.map_over_batch = True	1✔
352	self.input_map_dim = tuple(map_over_batch)	1✔
353	else:
354	self.map_over_batch = False	1✔
355	## Add the function to the globals
356	try:	1✔
357	code = 'import torch\n@torch.fx.wrap\n' + func['code']	1✔
358	exec(code, globals())	1✔
359	except Exception as e:	×
360	print(f"An error occurred: {e}")	×
361
362	def forward(self, *inputs):	1✔
363	args = list(inputs) + self.params_and_consts	1✔
364	# Retrieve the function object from the globals dictionary
365	function_to_call = globals()[self.name]	1✔
366	# Call the function using the retrieved function object
367	if self.map_over_batch:	1✔
368	function_to_call = torch.func.vmap(function_to_call,in_dims=self.input_map_dim)	1✔
369	result = function_to_call(*args)	1✔
370	return result	1✔
371
372	def createParamFun(self, *func_params):	1✔
373	return Parametric_Layer(func=func_params[0], params_and_consts=func_params[1], map_over_batch=func_params[2])	1✔
374
375	setattr(Model, paramfun_relation_name, createParamFun)	1✔

tonegas / nnodely / 13775554764

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