13504505673

Committed 24 Feb 2025 05:53PM UTC coverage: 95.261% (+0.3%) from 94.961%

Build # 13504505673

Build Type

Pull #59

github

Committed by

web-flow

Commit Message

Merge c15288a9e into 0c108fc0d

Pull Request Pull Request #59: Features/56 dynamic parametric function

Run Details

567 of 582 new or added lines in 24 files covered. (97.42%)

3 existing lines in 3 files now uncovered.

10171 of 10677 relevant lines covered (95.26%)

0.95 hits per line

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

97.64

/nnodely/relation.py

import copy

import numpy as np

from nnodely.utils import check, merge, enforce_types

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

MAIN_JSON = {
                'Info' : {},
                'Inputs' : {},
                'States' : {},
                'Constants': {},
                'Parameters' : {},
                'Functions' : {},
                'Relations': {},
                'Outputs': {}
            }

CHECK_NAMES = True
NeuObj_names = []

def toStream(obj):
    from nnodely.parameter import Parameter, Constant
    if type(obj) in (int,float,list,np.ndarray):
        obj = Constant('Constant'+str(NeuObj.count), obj)
        #obj = Stream(obj, MAIN_JSON, {'dim': 1}) if type(obj) in (int, float) else obj
    if type(obj) is Parameter or type(obj) is Constant:
        obj = Stream(obj.name, obj.json, obj.dim)
    return obj


class NeuObj():
    count = 0
    @classmethod
    def reset_count(self):
        NeuObj.count = 0
    def __init__(self, name='', json={}, dim=0):
        NeuObj.count += 1
        if name == '':
            name = 'Auto'+str(NeuObj.count)
        if CHECK_NAMES == True:
            check(name not in NeuObj_names, NameError, f"The name {name} is already used change the name of NeuObj.")
            NeuObj_names.append(name)
        self.name = name
        self.dim = dim
        if json:
            self.json = copy.deepcopy(json)
        else:
            self.json = copy.deepcopy(MAIN_JSON)

class Relation():
    def __add__(self, obj):
        from nnodely.arithmetic import Add
        return Add(self, obj)

    def __sub__(self, obj):
        from nnodely.arithmetic import Sub
        return Sub(self, obj)

    def __truediv__(self, obj):
        from nnodely.arithmetic import Div
        return Div(self, obj)

    def __mul__(self, obj):
        from nnodely.arithmetic import Mul
        return Mul(self, obj)

    def __pow__(self, obj):
        from nnodely.arithmetic import Pow
        return Pow(self, obj)

    def __neg__(self):
        from nnodely.arithmetic import Neg
        return Neg(self)

class Stream(Relation):
    count = 0
    @classmethod
    def reset_count(self):
        Stream.count = 0

    def __init__(self, name, json, dim, count = 1):
        Stream.count += count
        self.name = name
        self.json = copy.deepcopy(json)
        self.dim = dim

    @enforce_types
    def tw(self, tw:float|int|list, offset:float|int|None = None) -> "Stream":
        """
        Selects a time window on Stream. It is possible to create a smaller or bigger time window on the stream.
        The Time Window must be in the past not in the future.

        Parameters
        ----------
        tw : float, int, list
            The time window represents the time in the past. If a list, it should contain the start and end times, both indexes must be in the past.
        offset : float, int, optional
            The offset for the sample window. Default is None.

        Returns
        -------
        Stream
            A Stream representing the TimePart object with the selected time window.

        """
        from nnodely.input import State, Connect
        if type(tw) is list:
            check(0 >= tw[1] > tw[0] and tw[0] < 0, ValueError, "The dimension of the sample window must be in the past.")
        s = State(self.name+"_state",dimensions=self.dim['dim'])
        out_connect = Connect(self, s)
        win_state = s.tw(tw, offset)
        return Stream(win_state.name, merge(win_state.json, out_connect.json), win_state.dim,0 )

    @enforce_types
    def sw(self, sw:int|list, offset:int|None = None) -> "Stream":
        """
        Selects a sample window on Stream. It is possible to create a smaller or bigger window on the stream.
        The Sample Window must be in the past not in the future.

        Parameters
        ----------
        sw : int, list
            The sample window represents the number of steps in the past. If a list, it should contain the start and end indices, both indexes must be in the past.
        offset : int, optional
            The offset for the sample window. Default is None.

        Returns
        -------
        Stream
            A Stream representing the SamplePart object with the selected samples.

        """
        from nnodely.input import State, Connect
        if type(sw) is list:
            check(0 >= sw[1] > sw[0] and sw[0] < 0, ValueError, "The dimension of the sample window must be in the past.")
        s = State(self.name+"_state",dimensions=self.dim['dim'])
        out_connect = Connect(self, s)
        win_state = s.sw(sw, offset)
        return Stream(win_state.name, merge(win_state.json, out_connect.json), win_state.dim,0 )

    @enforce_types
    def z(self, delay:int|float) -> "Stream":
        """
        Considering the Zeta transform notation. The function is used to delay a Stream.
        The value of the delay can be only positive.

        Parameters
        ----------
        delay : int
            The delay value.

        Returns
        -------
        Stream
            A Stream representing the delayed Stream
        """
        check(delay > 0, ValueError, "The delay must be a positive integer")
        check('sw' in self.dim, TypeError, "The stream is not defined in samples but in time")
        return self.sw([-self.dim['sw']-delay,-delay])

    @enforce_types
    def delay(self, delay:int|float) -> "Stream":
        """
        The function is used to delay a Stream.
        The value of the delay can be only positive.

        Parameters
        ----------
        delay : int, float
            The delay value.

        Returns
        -------
        Stream
            A Stream representing the delayed Stream
        """
        check(delay > 0, ValueError, "The delay must be a positive integer")
        check('tw' in self.dim, TypeError, "The stream is not defined in time but in sample")
        return self.tw([-self.dim['tw']-delay,-delay])

    @enforce_types
    def s(self, order:int, method:str = 'ForwardEuler') -> "Stream":
        """
        Considering the Laplace transform notation. The function is used to operate an integral or derivate operation on a Stream.
        The order of the integral or the derivative operation is indicated by the order parameter.

        Parameters
        ----------
        order : int
            Order of the Laplace transform
        method : str, optional
            Integration or derivation method

        Returns
        -------
        Stream
            A Stream of the signal represents the integral or derivation operation.
        """
        from nnodely.timeoperation import Derivate, Integrate
        check(order != 0, ValueError, "The order must be a positive or negative integer not a zero")
        if order > 0:
            for i in range(order):
                o = Derivate(self, method = method)
        elif order < 0:
            for i in range(-order):
                o = Integrate(self, method = method)
        return o

    def connect(self, obj) -> "Stream":
        """
        Connects the current stream to a given state object.

        Parameters
        ----------
        obj : State
            The state object to connect to.

        Returns
        -------
        Stream
            A new Stream object representing the connected state.

        Raises
        ------
        TypeError
            If the provided object is not of type State.
        KeyError
            If the state variable is already connected.
        """
        from nnodely.input import State
        check(type(obj) is State, TypeError,
              f"The {obj} must be a State and not a {type(obj)}.")
        self.json = merge(self.json, obj.json)
        check('closedLoop' not in self.json['States'][obj.name] or 'connect' not in self.json['States'][obj.name], KeyError,
              f"The state variable {obj.name} is already connected.")
        self.json['States'][obj.name]['connect'] = self.name
        return Stream(self.name, self.json, self.dim,0 )

    def closedLoop(self, obj) -> "Stream":
        """
        Creates a closed loop connection with a given state object.

        Parameters
        ----------
        obj : State
            The state object to create a closed loop with.

        Returns
        -------
        Stream
            A new Stream object representing the closed loop state.

        Raises
        ------
        TypeError
            If the provided object is not of type State.
        KeyError
            If the state variable is already connected.
        """
        from nnodely.input import State
        check(type(obj) is State, TypeError,
              f"The {obj} must be a State and not a {type(obj)}.")
        self.json = merge(self.json, obj.json)
        check('closedLoop' not in self.json['States'][obj.name] or 'connect' not in self.json['States'][obj.name],
              KeyError,
              f"The state variable {obj.name} is already connected.")
        self.json['States'][obj.name]['closedLoop'] = self.name
        return Stream(self.name, self.json, self.dim,0 )

class ToStream():
    def __new__(cls, *args, **kwargs):
        out = super(ToStream,cls).__new__(cls)
        out.__init__(*args, **kwargs)
        return Stream(out.name,out.json,out.dim,0)

class AutoToStream():
    def __new__(cls, *args,  **kwargs):
        if len(args) > 0 and (issubclass(type(args[0]),NeuObj) or type(args[0]) is Stream):
            instance = super().__new__(cls)
            #instance.__init__(**kwargs)
            instance.__init__()
            return instance(args[0])
        instance = super().__new__(cls)
        return instance

1	import copy	1✔
2
3	import numpy as np	1✔
4
5	from nnodely.utils import check, merge, enforce_types	1✔
6
7	from nnodely.logger import logging, nnLogger	1✔
8	log = nnLogger(__name__, logging.CRITICAL)	1✔
9
10	MAIN_JSON = {	1✔
11	'Info' : {},
12	'Inputs' : {},
13	'States' : {},
14	'Constants': {},
15	'Parameters' : {},
16	'Functions' : {},
17	'Relations': {},
18	'Outputs': {}
19	}
20
21	CHECK_NAMES = True	1✔
22	NeuObj_names = []	1✔
23
24	def toStream(obj):	1✔
25	from nnodely.parameter import Parameter, Constant	1✔
26	if type(obj) in (int,float,list,np.ndarray):	1✔
27	obj = Constant('Constant'+str(NeuObj.count), obj)	1✔
28	#obj = Stream(obj, MAIN_JSON, {'dim': 1}) if type(obj) in (int, float) else obj
29	if type(obj) is Parameter or type(obj) is Constant:	1✔
30	obj = Stream(obj.name, obj.json, obj.dim)	1✔
31	return obj	1✔
32
33
34	class NeuObj():	1✔
35	count = 0	1✔
36	@classmethod	1✔
37	def reset_count(self):	1✔
38	NeuObj.count = 0	1✔
39	def __init__(self, name='', json={}, dim=0):	1✔
40	NeuObj.count += 1	1✔
41	if name == '':	1✔
UNCOV 42	name = 'Auto'+str(NeuObj.count)	×
43	if CHECK_NAMES == True:	1✔
44	check(name not in NeuObj_names, NameError, f"The name {name} is already used change the name of NeuObj.")	×
45	NeuObj_names.append(name)	×
46	self.name = name	1✔
47	self.dim = dim	1✔
48	if json:	1✔
49	self.json = copy.deepcopy(json)	1✔
50	else:
51	self.json = copy.deepcopy(MAIN_JSON)	1✔
52
53	class Relation():	1✔
54	def __add__(self, obj):	1✔
55	from nnodely.arithmetic import Add	1✔
56	return Add(self, obj)	1✔
57
58	def __sub__(self, obj):	1✔
59	from nnodely.arithmetic import Sub	1✔
60	return Sub(self, obj)	1✔
61
62	def __truediv__(self, obj):	1✔
63	from nnodely.arithmetic import Div	1✔
64	return Div(self, obj)	1✔
65
66	def __mul__(self, obj):	1✔
67	from nnodely.arithmetic import Mul	1✔
68	return Mul(self, obj)	1✔
69
70	def __pow__(self, obj):	1✔
71	from nnodely.arithmetic import Pow	1✔
72	return Pow(self, obj)	1✔
73
74	def __neg__(self):	1✔
75	from nnodely.arithmetic import Neg	1✔
76	return Neg(self)	1✔
77
78	class Stream(Relation):	1✔
79	count = 0	1✔
80	@classmethod	1✔
81	def reset_count(self):	1✔
82	Stream.count = 0	1✔
83
84	def __init__(self, name, json, dim, count = 1):	1✔
85	Stream.count += count	1✔
86	self.name = name	1✔
87	self.json = copy.deepcopy(json)	1✔
88	self.dim = dim	1✔
89
90	@enforce_types	1✔
91	def tw(self, tw:float\|int\|list, offset:float\|int\|None = None) -> "Stream":	1✔
92	"""
93	Selects a time window on Stream. It is possible to create a smaller or bigger time window on the stream.
94	The Time Window must be in the past not in the future.
95
96	Parameters
97	----------
98	tw : float, int, list
99	The time window represents the time in the past. If a list, it should contain the start and end times, both indexes must be in the past.
100	offset : float, int, optional
101	The offset for the sample window. Default is None.
102
103	Returns
104	-------
105	Stream
106	A Stream representing the TimePart object with the selected time window.
107
108	"""
109	from nnodely.input import State, Connect	1✔
110	if type(tw) is list:	1✔
111	check(0 >= tw[1] > tw[0] and tw[0] < 0, ValueError, "The dimension of the sample window must be in the past.")	1✔
112	s = State(self.name+"_state",dimensions=self.dim['dim'])	1✔
113	out_connect = Connect(self, s)	1✔
114	win_state = s.tw(tw, offset)	1✔
115	return Stream(win_state.name, merge(win_state.json, out_connect.json), win_state.dim,0 )	1✔
116
117	@enforce_types	1✔
118	def sw(self, sw:int\|list, offset:int\|None = None) -> "Stream":	1✔
119	"""
120	Selects a sample window on Stream. It is possible to create a smaller or bigger window on the stream.
121	The Sample Window must be in the past not in the future.
122
123	Parameters
124	----------
125	sw : int, list
126	The sample window represents the number of steps in the past. If a list, it should contain the start and end indices, both indexes must be in the past.
127	offset : int, optional
128	The offset for the sample window. Default is None.
129
130	Returns
131	-------
132	Stream
133	A Stream representing the SamplePart object with the selected samples.
134
135	"""
136	from nnodely.input import State, Connect	1✔
137	if type(sw) is list:	1✔
138	check(0 >= sw[1] > sw[0] and sw[0] < 0, ValueError, "The dimension of the sample window must be in the past.")	1✔
139	s = State(self.name+"_state",dimensions=self.dim['dim'])	1✔
140	out_connect = Connect(self, s)	1✔
141	win_state = s.sw(sw, offset)	1✔
142	return Stream(win_state.name, merge(win_state.json, out_connect.json), win_state.dim,0 )	1✔
143
144	@enforce_types	1✔
145	def z(self, delay:int\|float) -> "Stream":	1✔
146	"""
147	Considering the Zeta transform notation. The function is used to delay a Stream.
148	The value of the delay can be only positive.
149
150	Parameters
151	----------
152	delay : int
153	The delay value.
154
155	Returns
156	-------
157	Stream
158	A Stream representing the delayed Stream
159	"""
160	check(delay > 0, ValueError, "The delay must be a positive integer")	1✔
161	check('sw' in self.dim, TypeError, "The stream is not defined in samples but in time")	1✔
162	return self.sw([-self.dim['sw']-delay,-delay])	1✔
163
164	@enforce_types	1✔
165	def delay(self, delay:int\|float) -> "Stream":	1✔
166	"""
167	The function is used to delay a Stream.
168	The value of the delay can be only positive.
169
170	Parameters
171	----------
172	delay : int, float
173	The delay value.
174
175	Returns
176	-------
177	Stream
178	A Stream representing the delayed Stream
179	"""
180	check(delay > 0, ValueError, "The delay must be a positive integer")	1✔
181	check('tw' in self.dim, TypeError, "The stream is not defined in time but in sample")	1✔
182	return self.tw([-self.dim['tw']-delay,-delay])	1✔
183
184	@enforce_types	1✔
185	def s(self, order:int, method:str = 'ForwardEuler') -> "Stream":	1✔
186	"""
187	Considering the Laplace transform notation. The function is used to operate an integral or derivate operation on a Stream.
188	The order of the integral or the derivative operation is indicated by the order parameter.
189
190	Parameters
191	----------
192	order : int
193	Order of the Laplace transform
194	method : str, optional
195	Integration or derivation method
196
197	Returns
198	-------
199	Stream
200	A Stream of the signal represents the integral or derivation operation.
201	"""
202	from nnodely.timeoperation import Derivate, Integrate	1✔
203	check(order != 0, ValueError, "The order must be a positive or negative integer not a zero")	1✔
204	if order > 0:	1✔
205	for i in range(order):	1✔
206	o = Derivate(self, method = method)	1✔
207	elif order < 0:	1✔
208	for i in range(-order):	1✔
209	o = Integrate(self, method = method)	1✔
210	return o	1✔
211
212	def connect(self, obj) -> "Stream":	1✔
213	"""
214	Connects the current stream to a given state object.
215
216	Parameters
217	----------
218	obj : State
219	The state object to connect to.
220
221	Returns
222	-------
223	Stream
224	A new Stream object representing the connected state.
225
226	Raises
227	------
228	TypeError
229	If the provided object is not of type State.
230	KeyError
231	If the state variable is already connected.
232	"""
233	from nnodely.input import State	1✔
234	check(type(obj) is State, TypeError,	1✔
235	f"The {obj} must be a State and not a {type(obj)}.")
236	self.json = merge(self.json, obj.json)	1✔
237	check('closedLoop' not in self.json['States'][obj.name] or 'connect' not in self.json['States'][obj.name], KeyError,	1✔
238	f"The state variable {obj.name} is already connected.")
239	self.json['States'][obj.name]['connect'] = self.name	1✔
240	return Stream(self.name, self.json, self.dim,0 )	1✔
241
242	def closedLoop(self, obj) -> "Stream":	1✔
243	"""
244	Creates a closed loop connection with a given state object.
245
246	Parameters
247	----------
248	obj : State
249	The state object to create a closed loop with.
250
251	Returns
252	-------
253	Stream
254	A new Stream object representing the closed loop state.
255
256	Raises
257	------
258	TypeError
259	If the provided object is not of type State.
260	KeyError
261	If the state variable is already connected.
262	"""
263	from nnodely.input import State	1✔
264	check(type(obj) is State, TypeError,	1✔
265	f"The {obj} must be a State and not a {type(obj)}.")
266	self.json = merge(self.json, obj.json)	1✔
267	check('closedLoop' not in self.json['States'][obj.name] or 'connect' not in self.json['States'][obj.name],	1✔
268	KeyError,
269	f"The state variable {obj.name} is already connected.")
270	self.json['States'][obj.name]['closedLoop'] = self.name	1✔
271	return Stream(self.name, self.json, self.dim,0 )	1✔
272
273	class ToStream():	1✔
274	def __new__(cls, args, *kwargs):	1✔
275	out = super(ToStream,cls).__new__(cls)	1✔
276	out.__init__(args, *kwargs)	1✔
277	return Stream(out.name,out.json,out.dim,0)	1✔
278
279	class AutoToStream():	1✔
280	def __new__(cls, args, *kwargs):	1✔
281	if len(args) > 0 and (issubclass(type(args[0]),NeuObj) or type(args[0]) is Stream):	1✔
282	instance = super().__new__(cls)	1✔
283	#instance.__init__(**kwargs)
284	instance.__init__()	1✔
285	return instance(args[0])	1✔
286	instance = super().__new__(cls)	1✔
287	return instance	1✔

tonegas / nnodely / 13504505673

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