1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
# Copyright 2008-2016 Free Software Foundation, Inc.
# This file is part of GNU Radio
#
# GNU Radio Companion is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# GNU Radio Companion is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
from __future__ import absolute_import
from . import _virtual_connections
from .. import Constants
from ..base import Element
from ..utils.descriptors import (
EvaluatedFlag, EvaluatedEnum, EvaluatedPInt,
setup_names, lazy_property
)
@setup_names
class Port(Element):
is_port = True
dtype = EvaluatedEnum(list(Constants.TYPE_TO_SIZEOF.keys()), default='')
vlen = EvaluatedPInt()
multiplicity = EvaluatedPInt()
hidden = EvaluatedFlag()
optional = EvaluatedFlag()
def __init__(self, parent, direction, id, label='', domain=Constants.DEFAULT_DOMAIN, dtype='',
vlen='', multiplicity=1, optional=False, hide='', **_):
"""Make a new port from nested data."""
Element.__init__(self, parent)
self._dir = direction
self.key = id
if not label:
label = id if not id.isdigit() else {'sink': 'in', 'source': 'out'}[direction] + id
self.name = self._base_name = label
self.domain = domain
self.dtype = dtype
self.vlen = vlen
if domain == Constants.GR_MESSAGE_DOMAIN: # ToDo: message port class
self.key = self.name
self.dtype = 'message'
self.multiplicity = multiplicity
self.optional = optional
self.hidden = hide
# end of args ########################################################
self.clones = [] # References to cloned ports (for nports > 1)
def __str__(self):
if self.is_source:
return 'Source - {}({})'.format(self.name, self.key)
if self.is_sink:
return 'Sink - {}({})'.format(self.name, self.key)
def __repr__(self):
return '{!r}.{}[{}]'.format(self.parent, 'sinks' if self.is_sink else 'sources', self.key)
@property
def item_size(self):
return Constants.TYPE_TO_SIZEOF[self.dtype] * self.vlen
@lazy_property
def is_sink(self):
return self._dir == 'sink'
@lazy_property
def is_source(self):
return self._dir == 'source'
@property
def inherit_type(self):
"""always empty for e.g. virtual blocks, may eval to empty for 'Wildcard'"""
return not self.dtype
def validate(self):
Element.validate(self)
platform = self.parent_platform
num_connections = len(list(self.connections(enabled=True)))
need_connection = not self.optional and not self.hidden
if need_connection and num_connections == 0:
self.add_error_message('Port is not connected.')
if self.dtype not in Constants.TYPE_TO_SIZEOF.keys():
self.add_error_message('Type "{}" is not a possible type.'.format(self.dtype))
try:
domain = platform.domains[self.domain]
if self.is_sink and not domain.multi_in and num_connections > 1:
self.add_error_message('Domain "{}" can have only one upstream block'
''.format(self.domain))
if self.is_source and not domain.multi_out and num_connections > 1:
self.add_error_message('Domain "{}" can have only one downstream block'
''.format(self.domain))
except KeyError:
self.add_error_message('Domain key "{}" is not registered.'.format(self.domain))
def rewrite(self):
del self.vlen
del self.multiplicity
del self.hidden
del self.optional
del self.dtype
if self.inherit_type:
self.resolve_empty_type()
Element.rewrite(self)
# Update domain if was deduced from (dynamic) port type
if self.domain == Constants.GR_STREAM_DOMAIN and self.dtype == "message":
self.domain = Constants.GR_MESSAGE_DOMAIN
self.key = self.name
if self.domain == Constants.GR_MESSAGE_DOMAIN and self.dtype != "message":
self.domain = Constants.GR_STREAM_DOMAIN
self.key = '0' # Is rectified in rewrite()
def resolve_virtual_source(self):
"""Only used by Generator after validation is passed"""
return _virtual_connections.upstream_ports(self)
def resolve_empty_type(self):
def find_port(finder):
try:
return next((p for p in finder(self) if not p.inherit_type), None)
except _virtual_connections.LoopError as error:
self.add_error_message(str(error))
except (StopIteration, Exception):
pass
try:
port = find_port(_virtual_connections.upstream_ports) or \
find_port(_virtual_connections.downstream_ports)
self.set_evaluated('dtype', port.dtype) # we don't want to override the template
self.set_evaluated('vlen', port.vlen) # we don't want to override the template
self.domain = port.domain
except AttributeError:
self.domain = Constants.DEFAULT_DOMAIN
def add_clone(self):
"""
Create a clone of this (master) port and store a reference in self._clones.
The new port name (and key for message ports) will have index 1... appended.
If this is the first clone, this (master) port will get a 0 appended to its name (and key)
Returns:
the cloned port
"""
# Add index to master port name if there are no clones yet
if not self.clones:
self.name = self._base_name + '0'
# Also update key for none stream ports
if not self.key.isdigit():
self.key = self.name
name = self._base_name + str(len(self.clones) + 1)
# Dummy value 99999 will be fixed later
key = '99999' if self.key.isdigit() else name
# Clone
port_factory = self.parent_platform.make_port
port = port_factory(self.parent, direction=self._dir,
name=name, key=key,
master=self, cls_key='clone')
self.clones.append(port)
return port
def remove_clone(self, port):
"""
Remove a cloned port (from the list of clones only)
Remove the index 0 of the master port name (and key9 if there are no more clones left
"""
self.clones.remove(port)
# Remove index from master port name if there are no more clones
if not self.clones:
self.name = self._base_name
# Also update key for none stream ports
if not self.key.isdigit():
self.key = self.name
def connections(self, enabled=None):
"""Iterator over all connections to/from this port
enabled: None for all, True for enabled only, False for disabled only
"""
for con in self.parent_flowgraph.connections:
if self in con and (enabled is None or enabled == con.enabled):
yield con
|
