13547687602

Committed 26 Feb 2025 03:39PM UTC coverage: 95.825% (-0.09%) from 95.915%

Build # 13547687602

Build Type

push

github

Committed by

tonegas

Commit Message

Added rpow to the relation

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

11 existing lines in 1 file now uncovered.

10674 of 11139 relevant lines covered (95.83%)

0.96 hits per line

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

87.2

/nnodely/relation.py

import copy

import numpy as np

from nnodely.utils import check, merge, enforce_types, ForbiddenTags

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

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

CHECK_NAMES = True

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
    names = []
    @classmethod
    @enforce_types
    def clearNames(self, names:str|list|None=None):
        if names is None:
            NeuObj.count = 0
            NeuObj.names = []
        else:
            if type(names) is list:
                for name in names:
                    if name in NeuObj.names:
                        NeuObj.names.remove(name)
            else:
                if names in NeuObj.names:
                    NeuObj.names.remove(names)
    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.")
            check(name not in ForbiddenTags, NameError, f"The name '{name}' is a forbidden tag.")
            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 __radd__(self, obj):
        from nnodely.arithmetic import Add
        return Add(obj, self)

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

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

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

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

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

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

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

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

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

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

    def __init__(self, name, json, dim, count = 1):
        Stream.count += count
        check(name not in ForbiddenTags, NameError, f"The name '{name}' is a forbidden tag.")
        self.name = name
        self.json = copy.deepcopy(json)
        self.dim = dim

    def __str__(self):
        from nnodely.visualizer.visualizer import color, GREEN
        from pprint import pformat
        stream = f" Stream "
        stream_name = f" {self.name} {self.dim} "

        title = color((stream).center(80, '='), GREEN, True)
        json = color(pformat(self.json), GREEN)
        stream = color((stream_name).center(80, '-'), GREEN, True)
        return title + '\n' + json + '\n' + stream

    def __repr__(self):
        return self.__str__()

    @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, ForbiddenTags	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
23	def toStream(obj):	1✔
24	from nnodely.parameter import Parameter, Constant	1✔
25	if type(obj) in (int,float,list,np.ndarray):	1✔
26	obj = Constant('Constant'+str(NeuObj.count), obj)	1✔
27	#obj = Stream(obj, MAIN_JSON, {'dim': 1}) if type(obj) in (int, float) else obj
28	if type(obj) is Parameter or type(obj) is Constant:	1✔
29	obj = Stream(obj.name, obj.json, obj.dim)	1✔
30	return obj	1✔
31
32
33	class NeuObj():	1✔
34	count = 0	1✔
35	names = []	1✔
36	@classmethod	1✔
37	@enforce_types	1✔
38	def clearNames(self, names:str\|list\|None=None):	1✔
39	if names is None:	1✔
40	NeuObj.count = 0	1✔
41	NeuObj.names = []	1✔
42	else:
43	if type(names) is list:	×
44	for name in names:	×
45	if name in NeuObj.names:	×
46	NeuObj.names.remove(name)	×
47	else:
48	if names in NeuObj.names:	×
49	NeuObj.names.remove(names)	×
50	def __init__(self, name='', json={}, dim=0):	1✔
51	NeuObj.count += 1	1✔
52	if name == '':	1✔
53	name = 'Auto'+str(NeuObj.count)	×
54	if CHECK_NAMES == True:	1✔
55	check(name not in NeuObj.names, NameError, f"The name {name} is already used change the name of NeuObj.")	×
56	check(name not in ForbiddenTags, NameError, f"The name '{name}' is a forbidden tag.")	×
57	NeuObj.names.append(name)	×
58	self.name = name	1✔
59	self.dim = dim	1✔
60	if json:	1✔
61	self.json = copy.deepcopy(json)	1✔
62	else:
63	self.json = copy.deepcopy(MAIN_JSON)	1✔
64
65	class Relation():	1✔
66	def __add__(self, obj):	1✔
67	from nnodely.arithmetic import Add	1✔
68	return Add(self, obj)	1✔
69
70	def __radd__(self, obj):	1✔
UNCOV 71	from nnodely.arithmetic import Add	×
UNCOV 72	return Add(obj, self)	×
73
74	def __sub__(self, obj):	1✔
75	from nnodely.arithmetic import Sub	1✔
76	return Sub(self, obj)	1✔
77
78	def __rsub__(self, obj):	1✔
UNCOV 79	from nnodely.arithmetic import Sub	×
UNCOV 80	return Sub(obj, self)	×
81
82	def __truediv__(self, obj):	1✔
83	from nnodely.arithmetic import Div	1✔
84	return Div(self, obj)	1✔
85
86	def __rtruediv__(self, obj):	1✔
UNCOV 87	from nnodely.arithmetic import Div	×
UNCOV 88	return Div(obj, self)	×
89
90	def __mul__(self, obj):	1✔
91	from nnodely.arithmetic import Mul	1✔
92	return Mul(self, obj)	1✔
93
94	def __rmul__(self, obj):	1✔
UNCOV 95	from nnodely.arithmetic import Mul	×
UNCOV 96	return Mul(obj, self)	×
97
98	def __pow__(self, obj):	1✔
99	from nnodely.arithmetic import Pow	1✔
100	return Pow(self, obj)	1✔
101
102	def __rpow__(self, obj):	1✔
UNCOV 103	from nnodely.arithmetic import Pow	×
UNCOV 104	return Pow(obj, self)	×
105
106	def __neg__(self):	1✔
107	from nnodely.arithmetic import Neg	1✔
108	return Neg(self)	1✔
109
110	class Stream(Relation):	1✔
111	count = 0	1✔
112	@classmethod	1✔
113	def resetCount(self):	1✔
114	Stream.count = 0	1✔
115
116	def __init__(self, name, json, dim, count = 1):	1✔
117	Stream.count += count	1✔
118	check(name not in ForbiddenTags, NameError, f"The name '{name}' is a forbidden tag.")	1✔
119	self.name = name	1✔
120	self.json = copy.deepcopy(json)	1✔
121	self.dim = dim	1✔
122
123	def __str__(self):	1✔
124	from nnodely.visualizer.visualizer import color, GREEN	1✔
125	from pprint import pformat	1✔
126	stream = f" Stream "	1✔
127	stream_name = f" {self.name} {self.dim} "	1✔
128
129	title = color((stream).center(80, '='), GREEN, True)	1✔
130	json = color(pformat(self.json), GREEN)	1✔
131	stream = color((stream_name).center(80, '-'), GREEN, True)	1✔
132	return title + '\n' + json + '\n' + stream	1✔
133
134	def __repr__(self):	1✔
UNCOV 135	return self.__str__()	×
136
137	@enforce_types	1✔
138	def tw(self, tw:float\|int\|list, offset:float\|int\|None = None) -> "Stream":	1✔
139	"""
140	Selects a time window on Stream. It is possible to create a smaller or bigger time window on the stream.
141	The Time Window must be in the past not in the future.
142
143	Parameters
144	----------
145	tw : float, int, list
146	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.
147	offset : float, int, optional
148	The offset for the sample window. Default is None.
149
150	Returns
151	-------
152	Stream
153	A Stream representing the TimePart object with the selected time window.
154
155	"""
156	from nnodely.input import State, Connect	1✔
157	if type(tw) is list:	1✔
158	check(0 >= tw[1] > tw[0] and tw[0] < 0, ValueError, "The dimension of the sample window must be in the past.")	1✔
159	s = State(self.name+"_state",dimensions=self.dim['dim'])	1✔
160	out_connect = Connect(self, s)	1✔
161	win_state = s.tw(tw, offset)	1✔
162	return Stream(win_state.name, merge(win_state.json, out_connect.json), win_state.dim,0 )	1✔
163
164	@enforce_types	1✔
165	def sw(self, sw:int\|list, offset:int\|None = None) -> "Stream":	1✔
166	"""
167	Selects a sample window on Stream. It is possible to create a smaller or bigger window on the stream.
168	The Sample Window must be in the past not in the future.
169
170	Parameters
171	----------
172	sw : int, list
173	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.
174	offset : int, optional
175	The offset for the sample window. Default is None.
176
177	Returns
178	-------
179	Stream
180	A Stream representing the SamplePart object with the selected samples.
181
182	"""
183	from nnodely.input import State, Connect	1✔
184	if type(sw) is list:	1✔
185	check(0 >= sw[1] > sw[0] and sw[0] < 0, ValueError, "The dimension of the sample window must be in the past.")	1✔
186	s = State(self.name+"_state",dimensions=self.dim['dim'])	1✔
187	out_connect = Connect(self, s)	1✔
188	win_state = s.sw(sw, offset)	1✔
189	return Stream(win_state.name, merge(win_state.json, out_connect.json), win_state.dim,0 )	1✔
190
191	@enforce_types	1✔
192	def z(self, delay:int\|float) -> "Stream":	1✔
193	"""
194	Considering the Zeta transform notation. The function is used to delay a Stream.
195	The value of the delay can be only positive.
196
197	Parameters
198	----------
199	delay : int
200	The delay value.
201
202	Returns
203	-------
204	Stream
205	A Stream representing the delayed Stream
206	"""
207	check(delay > 0, ValueError, "The delay must be a positive integer")	1✔
208	check('sw' in self.dim, TypeError, "The stream is not defined in samples but in time")	1✔
209	return self.sw([-self.dim['sw']-delay,-delay])	1✔
210
211	@enforce_types	1✔
212	def delay(self, delay:int\|float) -> "Stream":	1✔
213	"""
214	The function is used to delay a Stream.
215	The value of the delay can be only positive.
216
217	Parameters
218	----------
219	delay : int, float
220	The delay value.
221
222	Returns
223	-------
224	Stream
225	A Stream representing the delayed Stream
226	"""
227	check(delay > 0, ValueError, "The delay must be a positive integer")	1✔
228	check('tw' in self.dim, TypeError, "The stream is not defined in time but in sample")	1✔
229	return self.tw([-self.dim['tw']-delay,-delay])	1✔
230
231	@enforce_types	1✔
232	def s(self, order:int, method:str = 'ForwardEuler') -> "Stream":	1✔
233	"""
234	Considering the Laplace transform notation. The function is used to operate an integral or derivate operation on a Stream.
235	The order of the integral or the derivative operation is indicated by the order parameter.
236
237	Parameters
238	----------
239	order : int
240	Order of the Laplace transform
241	method : str, optional
242	Integration or derivation method
243
244	Returns
245	-------
246	Stream
247	A Stream of the signal represents the integral or derivation operation.
248	"""
249	from nnodely.timeoperation import Derivate, Integrate	1✔
250	check(order != 0, ValueError, "The order must be a positive or negative integer not a zero")	1✔
251	if order > 0:	1✔
252	for i in range(order):	1✔
253	o = Derivate(self, method = method)	1✔
254	elif order < 0:	1✔
255	for i in range(-order):	1✔
256	o = Integrate(self, method = method)	1✔
257	return o	1✔
258
259	def connect(self, obj) -> "Stream":	1✔
260	"""
261	Connects the current stream to a given state object.
262
263	Parameters
264	----------
265	obj : State
266	The state object to connect to.
267
268	Returns
269	-------
270	Stream
271	A new Stream object representing the connected state.
272
273	Raises
274	------
275	TypeError
276	If the provided object is not of type State.
277	KeyError
278	If the state variable is already connected.
279	"""
280	from nnodely.input import State	1✔
281	check(type(obj) is State, TypeError,	1✔
282	f"The {obj} must be a State and not a {type(obj)}.")
283	self.json = merge(self.json, obj.json)	1✔
284	check('closedLoop' not in self.json['States'][obj.name] or 'connect' not in self.json['States'][obj.name], KeyError,	1✔
285	f"The state variable {obj.name} is already connected.")
286	self.json['States'][obj.name]['connect'] = self.name	1✔
287	return Stream(self.name, self.json, self.dim,0 )	1✔
288
289	def closedLoop(self, obj) -> "Stream":	1✔
290	"""
291	Creates a closed loop connection with a given state object.
292
293	Parameters
294	----------
295	obj : State
296	The state object to create a closed loop with.
297
298	Returns
299	-------
300	Stream
301	A new Stream object representing the closed loop state.
302
303	Raises
304	------
305	TypeError
306	If the provided object is not of type State.
307	KeyError
308	If the state variable is already connected.
309	"""
310	from nnodely.input import State	1✔
311	check(type(obj) is State, TypeError,	1✔
312	f"The {obj} must be a State and not a {type(obj)}.")
313	self.json = merge(self.json, obj.json)	1✔
314	check('closedLoop' not in self.json['States'][obj.name] or 'connect' not in self.json['States'][obj.name],	1✔
315	KeyError,
316	f"The state variable {obj.name} is already connected.")
317	self.json['States'][obj.name]['closedLoop'] = self.name	1✔
318	return Stream(self.name, self.json, self.dim,0 )	1✔
319
320	class ToStream():	1✔
321	def __new__(cls, args, *kwargs):	1✔
322	out = super(ToStream,cls).__new__(cls)	1✔
323	out.__init__(args, *kwargs)	1✔
324	return Stream(out.name,out.json,out.dim,0)	1✔
325
326	class AutoToStream():	1✔
327	def __new__(cls, args, *kwargs):	1✔
328	if len(args) > 0 and (issubclass(type(args[0]),NeuObj) or type(args[0]) is Stream):	1✔
329	instance = super().__new__(cls)	1✔
330	#instance.__init__(**kwargs)
331	instance.__init__()	1✔
332	return instance(args[0])	1✔
333	instance = super().__new__(cls)	1✔
334	return instance	1✔

tonegas / nnodely / 13547687602

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