13520426074

Committed 25 Feb 2025 11:43AM UTC coverage: 95.385% (-0.01%) from 95.395%

Build # 13520426074

Build Type

Pull #63

github

Committed by

web-flow

Commit Message

Merge 6e8d2fa3d into 16b35cf8f

Pull Request Pull Request #63: 49 general bug fix

Run Details

185 of 189 new or added lines in 16 files covered. (97.88%)

5 existing lines in 3 files now uncovered.

10521 of 11030 relevant lines covered (95.39%)

0.95 hits per line

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

96.9

/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
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.")
            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 __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
        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

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

tonegas / nnodely / 13520426074

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