rtc_handle.py
Go to the documentation of this file.
1 #/usr/bin/env python
2 # -*- coding: utf-8 -*-
3 # -*- Python -*-
4 
5 import sys
6 from omniORB import CORBA, URI
7 from omniORB import any
8 
9 import OpenRTM_aist
10 import RTC
11 
12 
13 from CorbaNaming import *
14 import SDOPackage
15 
16 # class RtmEnv :
17 # rtm environment manager
18 # orb, naming service, rtc proxy list
19 #
20 class RtmEnv :
21 
22  def __init__(self, orb_args, nserver_names=["localhost"],
23  orb=None, naming=None):
24  if not orb :
25  orb = CORBA.ORB_init(orb_args)
26  self.orb = orb
27  self.name_space = {}
28  if naming : # naming can specify only one naming service
29  self.name_space['default']=NameSpace(orb, naming=naming)
30  else :
31  for ns in nserver_names :
32  self.name_space[ns]=NameSpace(orb, server_name=ns)
33 
34  def __del__(self):
35  self.orb.shutdown(wait_for_completion=CORBA.FALSE)
36  self.orb.destroy()
37 #
38 # class NameSpace :
39 # rtc_handles and object list in naming service
40 #
41 class NameSpace :
42  def __init__(self, orb, server_name=None, naming=None):
43  self.orb = orb
44  self.name = server_name
45  if naming :
46  self.naming = naming
47  else :
48  self.naming = CorbaNaming(self.orb, server_name)
49 
50  self.b_len = 10 # iteration cut off no.
51  self.rtc_handles = {}
52  self.obj_list = {}
53 
54  def get_object_by_name(self, name, cl=RTC.RTObject):
55  ref = self.naming.resolveStr(name)
56  if ref is None: return None # return CORBA.nil ?
57  if cl :
58  return ref._narrow(cl)
59  else :
60  return ref
61 
62  def list_obj(self) :
63  self.rtc_handes = {}
64  self.obj_list = {}
65  return self.list_obj1(self.naming._rootContext, "")
66 
67  def list_obj1(self, name_context, parent) :
68  if not name_context :
69  name_context = self.naming._rootContext
70  rslt = []
71  b_list = name_context.list(self.b_len)
72  for bd in b_list[0] :
73  rslt = rslt + self.proc_bd(bd, name_context, parent)
74  if b_list[1] : # iterator : there exists remaining.
75  t_list = b_list[1].next_n(self.b_len)
76  while t_list[0] :
77  for bd in t_list[1] :
78  rslt = rslt + self.proc_bd(bd, name_context, parent)
79  t_list = b_list[1].next_n(self.b_len)
80  return rslt
81 
82  def proc_bd(self, bd, name_context, parent) :
83 # print '-------------------------------------------------------------------'
84 # print 'bd= ', bd
85 # print 'name_context= ', name_context
86 # print 'parent= ', parent
87  rslt = []
88  pre = ""
89  if parent :
90  pre = parent + "/"
91  nam = pre + URI.nameToString(bd.binding_name)
92  if bd.binding_type == CosNaming.nobject :
93  tmp = name_context.resolve(bd.binding_name)
94  self.obj_list[nam]=tmp
95  print 'objcet '+nam+' was listed.'
96  try :
97  tmp = tmp._narrow(RTC.RTObject)
98  except :
99  print nam+' is not RTC.'
100  tmp = None
101  try :
102  if tmp :
103  rslt = [[nam, tmp]]
104  self.rtc_handles[nam]=RtcHandle(nam,self,tmp)
105  print 'handle for '+nam+' was created.'
106  else :
107  pass
108  except :
109  print nam+' is not alive.'
110  pass
111  else :
112  tmp = name_context.resolve(bd.binding_name)
113  tmp = tmp._narrow(CosNaming.NamingContext)
114  rslt = self.list_obj1(tmp, nam)
115  return rslt
116 
117 #
118 # data conversion
119 #
120 def nvlist2dict(nvlist) :
121  rslt = {}
122  for tmp in nvlist :
123  rslt[tmp.name]=tmp.value.value() # nv.value and any.value()
124  return rslt
125 def dict2nvlist(dict) :
126  rslt = []
127  for tmp in dict.keys() :
128  rslt.append(SDOPackage.NameValue(tmp, any.to_any(dict[tmp])))
129  return rslt
130 #
131 # connector, port, inport, outport, service
132 #
133 
134 class Connector :
135  def __init__(self, plist, name = None, id="", prop_dict={}) :
136  self.plist = plist
137  self.port_reflist = [tmp.port_profile.port_ref for tmp in plist]
138  if name :
139  self.name = name
140  else :
141  self.name = string.join([tmp.name for tmp in plist],'_')
142  self.prop_dict = prop_dict
144  self.profile = RTC.ConnectorProfile(self.name, id, self.port_reflist, self.prop_nvlist)
145  self.nego_prop()
146 
147  def nego_prop(self) :
148  self.possible = True
149  for kk in self.def_prop :
150  if kk in self.prop_dict :
151  if not self.prop_dict[kk] :
152  self.prop_dict[kk]=self.def_prop[kk]
153  else :
154  self.prop_dict[kk]=self.def_prop[kk]
155  for pp in self.plist :
156  if not ((self.prop_dict[kk] in pp.prop[kk]) or
157  ('Any' in pp.prop[kk])) :
158  self.prop_dict[kk] = ""
159  self.possible = False
160  self.prop_nvlist = dict2nvlist(self.prop_dict)
161  self.profile.properties = self.prop_nvlist
162  return self.possible
163 
164  def connect(self) :
165 #
166 # out and inout parameters are retuned as a tuple
167 #
168  ret, self.profile = self.port_reflist[0].connect(self.profile)
169  self.prop_nvlist = self.profile.properties
170  self.prop_dict = nvlist2dict(self.prop_nvlist)
171  return ret
172 
173  def disconnect(self) :
174  ret = self.port_reflist[0].disconnect(self.profile.connector_id)
175  return ret
176 
178  def __init__(self, plist, name = None, id="", prop_dict={}) :
179 # self.def_prop = {'dataport.dataflow_type':'Push' ,
180 # 'dataport.interface_type':'CORBA_Any' ,
181 # 'dataport.subscription_type':'Flush'}
182  self.def_prop = {'dataport.dataflow_type':'push' ,
183  'dataport.interface_type':'corba_cdr' ,
184  'dataport.subscription_type':'flush'}
185  Connector.__init__(self, plist, name, id, prop_dict)
186 
188  def __init__(self, plist, name = None, id="", prop_dict={}) :
189  self.def_prop = {'port.port_type':'CorbaPort' }
190  Connector.__init__(self, plist, name, id, prop_dict)
191 
192 
193 class Port :
194  def __init__(self, profile,nv_dict=None) :
195  self.name=profile.name
196  self.port_profile = profile
197  if not nv_dict :
198  nv_dict = nvlist2dict(profile.properties)
199  self.prop = nv_dict
200  self.con = None # this must be set in each subclasses
201  def get_info(self) :
202  self.con.connect()
203  tmp1 = self.get_connections()
204  tmp2 = [pp.connector_id for pp in tmp1]
205  if self.con.profile.connector_id in tmp2 :
206  self.con.disconnect()
207 
208  def get_connections(self) :
209  return self.port_profile.port_ref.get_connector_profiles()
210 
211 class CorbaServer :
212  def __init__(self, profile, port) :
213  self.profile = profile
214  self.port = port
215  self.name = profile.instance_name
216  self.type = profile.type_name
217  self.ref = None
218  ref_key = 'port.' + self.type + '.' + self.name
219  self.ref=self.port.con.prop_dict[ref_key]
220 #
221 # if we import stubs before we create instances,
222 # we rarely need to narrow the object references.
223 # we need to specify global symbol table to evaluate class symbols.
224 #
225  def narrow_ref(self, gls) :
226  if self.type.find('::') == -1 :
227  self.narrow_sym = eval('_GlobalIDL.' + self.type, gls)
228  else :
229  self.narrow_sym = eval(self.type.replace('::','.'), gls)
230  self.ref = self.ref._narrow(self.narrow_sym)
231 
232 class CorbaClient :
233  def __init__(self, profile) :
234  self.profile = profile
235  self.name = profile.instance_name
236  self.type = profile.type_name
237 #
238 # to connect to an outside corba client,
239 # we need an implementation of the corresponding corba server.
240 # but ....
241 #
242 
243 class RtcService(Port) :
244  def __init__(self, profile,nv_dict=None) :
245  Port.__init__(self, profile, nv_dict)
246  self.con = ServiceConnector([self])
247  self.get_info()
248  self.provided={}
249  self.required={}
250  tmp = self.port_profile.interfaces
251  for itf in tmp :
252  if itf.polarity == RTC.PROVIDED :
253  self.provided[itf.instance_name] = CorbaServer(itf,self)
254  elif itf.polarity == RTC.REQUIRED :
255  self.required[itf.instance_name] = CorbaClient(itf)
256 
257 class RtcInport(Port) :
258  def __init__(self, profile, nv_dict=None) :
259  Port.__init__(self, profile, nv_dict)
260  self.con = IOConnector([self])
261  self.get_info()
262 # self.ref = self.con.prop_dict['dataport.corba_any.inport_ref']
263  self.ref = self.con.prop_dict['dataport.corba_cdr.inport_ref']
264  self.data_class = eval('RTC.' + self.prop['dataport.data_type'])
265  self.data_tc = eval('RTC._tc_' + self.prop['dataport.data_type'])
266  def write(self,data) :
267  self.ref.put(CORBA.Any(self.data_tc,
268  self.data_class(RTC.Time(0,0),data)))
269 
270 class RtcOutport(Port) :
271  def __init__(self, profile,nv_dict=None) :
272  Port.__init__(self, profile, nv_dict)
273  self.con = IOConnector([self])
274  self.get_info()
275  if 'dataport.corba_any.outport_ref' in self.con.prop_dict :
276  self.ref = self.con.prop_dict['dataport.corba_cdr.outport_ref']
277 # self.ref = self.con.prop_dict['dataport.corba_any.outport_ref']
278  else :
279  self.ref=None
280  self.data_class = eval('RTC.' + self.prop['dataport.data_type'])
281  self.data_tc = eval('RTC._tc_' + self.prop['dataport.data_type'])
282 
283  def read(self) :
284  if self.ref :
285  return self.ref.get().value()
286  else :
287  print "not supported"
288  return None
289 #
290 # RtcHandle
291 #
292 class RtcHandle :
293  def __init__(self, name, env, ref=None) :
294  self.name = name
295  self.env = env
296  if ref :
297  self.rtc_ref = ref
298  else :
299  self.rtc_ref = env.naming.resolve(name)._narrow(RTC.RTObject)
300  self.conf_ref = None
301  self.retrieve_info()
302 
303  def retrieve_info(self) :
304  self.conf_set={}
306  self.port_refs = []
308  if self.rtc_ref :
309  self.conf_ref = self.rtc_ref.get_configuration()
310  conf_set = self.conf_ref.get_configuration_sets()
311  for cc in conf_set :
312  self.conf_set[cc.id]=cc
313  self.conf_set_data[cc.id]=nvlist2dict(cc.configuration_data)
314  self.profile = self.rtc_ref.get_component_profile()
315  self.prop = nvlist2dict(self.profile.properties)
316  #self.execution_contexts = self.rtc_ref.get_contexts()
317  self.execution_contexts = self.rtc_ref.get_owned_contexts()
318  self.port_refs = self.rtc_ref.get_ports()
319  # this includes inports, outports and service ports
320  self.ports = {}
321  self.services = {}
322  self.inports = {}
323  self.outports = {}
324  for pp in self.port_refs :
325  tmp = pp.get_port_profile()
326  tmp_prop = nvlist2dict(tmp.properties)
327 # self.ports[tmp.name]=Port(tmp, tmp_prop)
328  if tmp_prop['port.port_type']=='DataInPort' :
329  self.inports[tmp.name]=RtcInport(tmp,tmp_prop)
330 # self.inports[tmp.name]=Port(tmp, tmp_prop)
331  elif tmp_prop['port.port_type']=='DataOutPort' :
332  self.outports[tmp.name]=RtcOutport(tmp, tmp_prop)
333 # self.outports[tmp.name]=Port(tmp, tmp_prop)
334  elif tmp_prop['port.port_type']=='CorbaPort' :
335  self.services[tmp.name]=RtcService(tmp, tmp_prop)
336 # self.services[tmp.name]=Port(tmp, tmp_prop)
337 
338  def set_conf(self,conf_set_name,param_name,value) :
339  conf_set=self.conf_set[conf_set_name]
340  conf_set_data=self.conf_set_data[conf_set_name]
341  conf_set_data[param_name]=value
342  conf_set.configuration_data=dict2nvlist(conf_set_data)
343  self.conf_ref.set_configuration_set_values(conf_set_name,conf_set)
344  def set_conf_activate(self,conf_set_name,param_name,value) :
345  self.set_conf(conf_set_name,param_name,value)
346  self.conf_ref.activate_configuration_set(conf_set_name)
347  def activate(self):
348  return self.execution_contexts[0].activate_component(self.rtc_ref)
349  def deactivate(self):
350  return self.execution_contexts[0].deactivate_component(self.rtc_ref)
351 #
352 # pipe
353 # a pipe is an port (interface & implementation)
354 # whhich corresponds to an outside port interface.
355 # you can subscribe and communicate to the outside port with the pipe.
356 # you need an rtc implementation to use pipes.
357 #
358 class Pipe :
359  pass
360 
361 class PipeOut(Pipe):
362  def __init__(self, name, data_buf, size=8) :
363  self.name = name
364  self.data = data_buf
365  self.OpenRTM_aist.InPort(name,data_buf,OpenRTM_aist.RingBuffer(size))
def __init__(self, orb_args, nserver_names=["localhost"], orb=None, naming=None)
Definition: rtc_handle.py:23
def get_object_by_name(self, name, cl=RTC.RTObject)
Definition: rtc_handle.py:54
def __init__(self, name, data_buf, size=8)
Definition: rtc_handle.py:362
def __init__(self, profile, nv_dict=None)
Definition: rtc_handle.py:258
def __init__(self, plist, name=None, id="", prop_dict={})
Definition: rtc_handle.py:178
def set_conf_activate(self, conf_set_name, param_name, value)
Definition: rtc_handle.py:344
def proc_bd(self, bd, name_context, parent)
Definition: rtc_handle.py:82
def __init__(self, orb, server_name=None, naming=None)
Definition: rtc_handle.py:42
def __init__(self, profile, nv_dict=None)
Definition: rtc_handle.py:271
def __init__(self, profile, nv_dict=None)
Definition: rtc_handle.py:244
def set_conf(self, conf_set_name, param_name, value)
Definition: rtc_handle.py:338
InPort template class.
Definition: InPort.py:58
def find(seq, f)
Return the index of CORBA sequence element that functor matches.
def list_obj1(self, name_context, parent)
Definition: rtc_handle.py:67
def __init__(self, plist, name=None, id="", prop_dict={})
Definition: rtc_handle.py:135
def __init__(self, plist, name=None, id="", prop_dict={})
Definition: rtc_handle.py:188
def __init__(self, profile, nv_dict=None)
Definition: rtc_handle.py:194


openrtm_aist_python
Author(s): Shinji Kurihara
autogenerated on Mon Feb 28 2022 23:01:06