rtc_handle10_11.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 from omniORB import any, cdrMarshal, cdrUnmarshal
9 
10 import OpenRTM_aist
11 import RTC
12 
13 
14 from CorbaNaming import *
15 import SDOPackage
16 # from EmbryonicRtc import *
17 
18 # class RtmEnv :
19 # rtm environment manager
20 # orb, naming service, rtc proxy list
21 #
22 class RtmEnv :
23 
24  def __init__(self, orb_args, nserver_names=["localhost"],
25  orb=None, naming=None):
26  if not orb :
27  orb = CORBA.ORB_init(orb_args)
28  self.orb = orb
29  self.name_space = {}
30  if naming : # naming can specify only one naming service
31  self.name_space['default']=NameSpace(orb, naming=naming)
32  else :
33  for ns in nserver_names :
34  self.name_space[ns]=NameSpace(orb, server_name=ns)
35 
36  def __del__(self):
37  self.orb.shutdown(wait_for_completion=CORBA.FALSE)
38  self.orb.destroy()
39 #
40 # class NameSpace :
41 # rtc_handles and object list in naming service
42 #
43 class NameSpace :
44  def __init__(self, orb, server_name=None, naming=None):
45  self.orb = orb
46  self.name = server_name
47  if naming :
48  self.naming = naming
49  else :
50  self.naming = CorbaNaming(self.orb, server_name)
51 
52  self.b_len = 10 # iteration cut off no.
53  self.rtc_handles = {}
54  self.obj_list = {}
55 
56  def get_object_by_name(self, name, cl=RTC.RTObject):
57  ref = self.naming.resolveStr(name)
58  if ref is None: return None # return CORBA.nil ?
59  if cl :
60  return ref._narrow(cl)
61  else :
62  return ref
63 
64  def list_obj(self) :
65  self.rtc_handes = {}
66  self.obj_list = {}
67  return self.list_obj1(self.naming._rootContext, "")
68 
69  def list_obj1(self, name_context, parent) :
70  if not name_context :
71  name_context = self.naming._rootContext
72  rslt = []
73  b_list = name_context.list(self.b_len)
74  for bd in b_list[0] :
75  rslt = rslt + self.proc_bd(bd, name_context, parent)
76  if b_list[1] : # iterator : there exists remaining.
77  t_list = b_list[1].next_n(self.b_len)
78  while t_list[0] :
79  for bd in t_list[1] :
80  rslt = rslt + self.proc_bd(bd, name_context, parent)
81  t_list = b_list[1].next_n(self.b_len)
82  return rslt
83 
84  def proc_bd(self, bd, name_context, parent) :
85 # print '-------------------------------------------------------------------'
86 # print 'bd= ', bd
87 # print 'name_context= ', name_context
88 # print 'parent= ', parent
89  rslt = []
90  pre = ""
91  if parent :
92  pre = parent + "/"
93  nam = pre + URI.nameToString(bd.binding_name)
94  if bd.binding_type == CosNaming.nobject :
95  tmp = name_context.resolve(bd.binding_name)
96  self.obj_list[nam]=tmp
97  print 'objcet '+nam+' was listed.'
98  try :
99  tmp = tmp._narrow(RTC.RTObject)
100  except :
101  print nam+' is not RTC.'
102  tmp = None
103  try :
104  if tmp :
105  rslt = [[nam, tmp]]
106  self.rtc_handles[nam]=RtcHandle(nam,self,tmp)
107  print 'handle for '+nam+' was created.'
108  else :
109  pass
110  except :
111  print nam+' is not alive.' , sys.exc_info()[0]
112  pass
113  else :
114  tmp = name_context.resolve(bd.binding_name)
115  tmp = tmp._narrow(CosNaming.NamingContext)
116  rslt = self.list_obj1(tmp, nam)
117  return rslt
118 
119 #
120 # data conversion
121 #
122 def nvlist2dict(nvlist) :
123  rslt = {}
124  for tmp in nvlist :
125  rslt[tmp.name]=tmp.value.value() # nv.value and any.value()
126  return rslt
127 def dict2nvlist(dict) :
128  rslt = []
129  for tmp in dict.keys() :
130  rslt.append(SDOPackage.NameValue(tmp, any.to_any(dict[tmp])))
131  return rslt
132 #
133 # connector, port, inport, outport, service
134 #
135 
136 class Connector :
137  def __init__(self, plist, name = None, id="", prop_dict={}) :
138  self.connectp=False
139  self.plist = plist
140  self.port_reflist = [tmp.port_profile.port_ref for tmp in plist]
141  if name :
142  self.name = name
143  else :
144  self.name = string.join([tmp.name for tmp in plist],'_')
145  self.prop_dict_req = prop_dict
147  self.profile_req = RTC.ConnectorProfile(self.name, id, self.port_reflist,
148  self.prop_nvlist_req)
149  self.nego_prop()
150 
151  def nego_prop(self) :
152  self.possible = True
153  for kk in self.def_prop :
154  if kk in self.prop_dict_req :
155  if not self.prop_dict_req[kk] :
156  self.prop_dict_req[kk]=self.def_prop[kk]
157  else :
158  self.prop_dict_req[kk]=self.def_prop[kk]
159  for pp in self.plist :
160  if not ((self.prop_dict_req[kk] in pp.prop[kk]) or
161  ('Any' in pp.prop[kk])) :
162  print kk, self.prop_dict_req[kk]
163  self.prop_dict_req[kk] = ""
164  self.possible = False
166  self.profile_req.properties = self.prop_nvlist_req
167  return self.possible
168 
169  def connect(self) :
170 #
171 # out and inout parameters are retuned as a tuple
172 #
173  if self.connectp == False :
174  ret, self.profile = self.port_reflist[0].connect(self.profile_req)
175  self.prop_nvlist = self.profile.properties
177  if ret == RTC.RTC_OK :
178  self.connectp=True
179  else :
180  ret = "?"
181  return ret
182 
183  def disconnect(self) :
184  if self.connectp == True :
185  ret = self.port_reflist[0].disconnect(self.profile.connector_id)
186  else :
187  ret = "?"
188  self.connectp = False
189  return ret
190 
192  def __init__(self, plist, name = None, id="", prop_dict={}) :
193 # self.def_prop = {'dataport.dataflow_type':'Push' ,
194 # 'dataport.interface_type':'CORBA_Any' ,
195 # 'dataport.subscription_type':'Flush'}
196  self.def_prop = {'dataport.dataflow_type':'push',
197  'dataport.interface_type':'corba_cdr' ,
198  'dataport.subscription_type':'flush'}
199  Connector.__init__(self, plist, name, id, prop_dict)
200 
202  def __init__(self, plist, name = None, id="", prop_dict={}) :
203  self.def_prop = {'port.port_type':'CorbaPort' }
204  Connector.__init__(self, plist, name, id, prop_dict)
205 
206 
207 class Port :
208  def __init__(self, profile,nv_dict=None,handle=None) :
209  self.handle=handle
210  self.name=profile.name
211  self.port_profile = profile
212  if not nv_dict :
213  nv_dict = nvlist2dict(profile.properties)
214  self.prop = nv_dict
215  self.con = None # this must be set in each subclasses
216  def get_info(self) :
217  self.con.connect()
218  tmp1 = self.get_connections()
219  tmp2 = [pp.connector_id for pp in tmp1]
220  if self.con.profile.connector_id in tmp2 :
221  print "connecting"
222  self.con.disconnect()
223 
224  def get_connections(self) :
225  return self.port_profile.port_ref.get_connector_profiles()
226 
227 class CorbaServer :
228  def __init__(self, profile, port) :
229  self.profile = profile
230  self.port = port
231  self.name = profile.instance_name
232  self.type = profile.type_name
233  self.ref = None
234  ref_key = 'port.' + self.type + '.' + self.name
235  self.ref=self.port.con.prop_dict[ref_key]
236  if isinstance(self.ref,str) :
237  self.ref=port.handle.env.orb.string_to_object(self.ref)
238 #
239 # if we import stubs before we create instances,
240 # we rarely need to narrow the object references.
241 # we need to specify global symbol table to evaluate class symbols.
242 #
243  def narrow_ref(self, gls) :
244  if self.type.find('::') == -1 :
245  self.narrow_sym = eval('_GlobalIDL.' + self.type, gls)
246  else :
247  self.narrow_sym = eval(self.type.replace('::','.'), gls)
248  self.ref = self.ref._narrow(self.narrow_sym)
249 
250 class CorbaClient :
251  def __init__(self, profile) :
252  self.profile = profile
253  self.name = profile.instance_name
254  self.type = profile.type_name
255 #
256 # to connect to an outside corba client,
257 # we need an implementation of the corresponding corba server.
258 # but ....
259 #
260 
261 class RtcService(Port) :
262  def __init__(self, profile,nv_dict=None, handle=None) :
263  Port.__init__(self, profile, nv_dict, handle)
264  self.con = ServiceConnector([self])
265  self.get_info()
266  self.provided={}
267  self.required={}
268  tmp = self.port_profile.interfaces
269  for itf in tmp :
270  if itf.polarity == RTC.PROVIDED :
271  self.provided[itf.instance_name] = CorbaServer(itf,self)
272  elif itf.polarity == RTC.REQUIRED :
273  self.required[itf.instance_name] = CorbaClient(itf)
274 
275 # def open(self) :
276 # self.con.connect()
277 # self.provided={}
278 # self.required={}
279 # tmp = self.port_profile.interfaces
280 # for itf in tmp :
281 # if itf.polarity == RTC.PROVIDED :
282 # self.provided[itf.instance_name] = CorbaServer(itf,self)
283 # elif itf.polarity == RTC.REQUIRED :
284 # self.required[itf.instance_name] = CorbaClient(itf)
285 
286 # def close(self) :
287 # return self.con.disconnect()
288 def strip_data_class(data_class_str) :
289  tmp = data_class_str.split(':')
290  if len(tmp) == 1 :
291  return data_class_str
292  else :
293  tmp = tmp[1].split('/')
294  return tmp[1]
295 
296 class RtcInport(Port) :
297  def __init__(self, profile, nv_dict=None, handle=None) :
298  Port.__init__(self, profile, nv_dict, handle)
299  self.con = IOConnector([self], prop_dict={'dataport.dataflow_type':'push'})
300  self.get_info()
301 # self.ref = self.con.prop_dict['dataport.corba_any.inport_ref']
302  self.ref = self.con.prop_dict['dataport.corba_cdr.inport_ref']
303 # self.data_class = eval('RTC.' + self.prop['dataport.data_type'])
304 # self.data_tc = eval('RTC._tc_' + self.prop['dataport.data_type'])
305  tmp=strip_data_class(self.prop['dataport.data_type'])
306  print tmp
307  self.data_class = eval('RTC.' + tmp)
308  self.data_tc = eval('RTC._tc_' + tmp)
309  def write(self,data) :
310 # self.ref.put(CORBA.Any(self.data_tc,
311 # self.data_class(RTC.Time(0,0),data)))
312  self.ref.put(cdrMarshal(self.data_tc,
313  self.data_class(RTC.Time(0,0),data), 1))
314  def open(self) :
315  self.con.connect()
316  self.ref = self.con.prop_dict['dataport.corba_cdr.inport_ref']
317 
318  def close(self) :
319  return self.con.disconnect()
320 
321 class RtcOutport(Port) :
322  def __init__(self, profile,nv_dict=None, handle=None) :
323  Port.__init__(self, profile, nv_dict, handle)
324  con_prop_dict={'dataport.dataflow_type':'pull',
325  'dataport.buffer.type':'ringbuffer',
326  'dataport.buffer.read.empty_policy':'last',
327  'dataport.buffer.length':'1'}
328  self.con = IOConnector([self], prop_dict=con_prop_dict)
329  self.get_info()
330 # if 'dataport.corba_any.outport_ref' in self.con.prop_dict :
331 # self.ref = self.con.prop_dict['dataport.corba_any.outport_ref']
332  if 'dataport.corba_cdr.outport_ref' in self.con.prop_dict :
333  self.ref = self.con.prop_dict['dataport.corba_cdr.outport_ref']
334  else :
335  self.ref=None
336 # self.data_class = eval('RTC.' + self.prop['dataport.data_type'])
337 # self.data_tc = eval('RTC._tc_' + self.prop['dataport.data_type'])
338  tmp=strip_data_class(self.prop['dataport.data_type'])
339  self.data_class = eval('RTC.' + tmp)
340  self.data_tc = eval('RTC._tc_' + tmp)
341 
342  def read(self) :
343  if self.ref :
344  try :
345  tmp1=self.ref.get()
346  tmp2= cdrUnmarshal(self.data_tc,tmp1[1], 1)
347 # return tmp2.data
348  return tmp2
349  except :
350  return None
351  else :
352  print "not supported"
353  return None
354 
355  def open(self) :
356  self.con.connect()
357  if 'dataport.corba_cdr.outport_ref' in self.con.prop_dict :
358  self.ref = self.con.prop_dict['dataport.corba_cdr.outport_ref']
359 
360  def close(self) :
361  return self.con.disconnect()
362 
363 #
364 # RtcHandle
365 #
366 class RtcHandle :
367  def __init__(self, name, env, ref=None) :
368  self.name = name
369  self.env = env
370  if ref :
371  self.rtc_ref = ref
372  else :
373  self.rtc_ref = env.naming.resolve(name)._narrow(RTC.RTObject)
374  self.conf_ref = None
375  self.retrieve_info()
376 
377  def retrieve_info(self) :
378  self.conf_set={}
380  self.port_refs = []
382  if self.rtc_ref :
383  self.conf_ref = self.rtc_ref.get_configuration()
384  conf_set = self.conf_ref.get_configuration_sets()
385  for cc in conf_set :
386  self.conf_set[cc.id]=cc
387  self.conf_set_data[cc.id]=nvlist2dict(cc.configuration_data)
388  self.profile = self.rtc_ref.get_component_profile()
389  self.prop = nvlist2dict(self.profile.properties)
390  #self.execution_contexts = self.rtc_ref.get_contexts()
391  self.execution_contexts = self.rtc_ref.get_owned_contexts()
392  self.port_refs = self.rtc_ref.get_ports()
393  # this includes inports, outports and service ports
394  self.ports = {}
395  self.services = {}
396  self.inports = {}
397  self.outports = {}
398  for pp in self.port_refs :
399  tmp = pp.get_port_profile()
400  tmp_prop = nvlist2dict(tmp.properties)
401  tmp_name = tmp.name.lstrip(self.name.split('.')[0]).lstrip('.')
402  print 'port_name:', tmp_name
403 # self.ports[tmp.name]=Port(tmp, tmp_prop)
404  if tmp_prop['port.port_type']=='DataInPort' :
405  self.inports[tmp_name]=RtcInport(tmp,tmp_prop, self)
406 # self.inports[tmp.name]=Port(tmp, tmp_prop)
407  elif tmp_prop['port.port_type']=='DataOutPort' :
408  self.outports[tmp_name]=RtcOutport(tmp, tmp_prop, self)
409 # self.outports[tmp.name]=Port(tmp, tmp_prop)
410  elif tmp_prop['port.port_type']=='CorbaPort' :
411  self.services[tmp_name]=RtcService(tmp, tmp_prop, self)
412 # self.services[tmp.name]=Port(tmp, tmp_prop)
413 
414  def set_conf(self,conf_set_name,param_name,value) :
415  conf_set=self.conf_set[conf_set_name]
416  conf_set_data=self.conf_set_data[conf_set_name]
417  conf_set_data[param_name]=value
418  conf_set.configuration_data=dict2nvlist(conf_set_data)
419 # self.conf_ref.set_configuration_set_values(conf_set_name,conf_set)
420  self.conf_ref.set_configuration_set_values(conf_set)
421  def set_conf_activate(self,conf_set_name,param_name,value) :
422  self.set_conf(conf_set_name,param_name,value)
423  self.conf_ref.activate_configuration_set(conf_set_name)
424  def activate(self):
425  return self.execution_contexts[0].activate_component(self.rtc_ref)
426  def deactivate(self):
427  return self.execution_contexts[0].deactivate_component(self.rtc_ref)
428  def reset(self):
429  return self.execution_contexts[0].reset_component(self.rtc_ref)
430  def get_state(self):
431  return self.execution_contexts[0].get_component_state(self.rtc_ref)
432 
433 #
434 # pipe
435 # a pipe is an port (interface & implementation)
436 # for a port(an RtcInport or RtcOutport object) of an outside rtc.
437 # you need an empty rtc (comp) to create pipes.
438 # you can subscribe and communicate to the outside port with the pipe.
439 #
440 #
441 class InPipe() :
442  def __init__(self,comp, port) :
443  self.comp=comp
444  self.port=port
445  self.pname=port.name.replace('.','_')
446  self.pipe=comp.makeOutPort(self.pname,port.data_class(RTC.Time(0,0),[]),OpenRTM_aist.RingBuffer(1))
447  self.buf=getattr(comp,'_d_'+self.pname)
448  tmp = self.pipe.getPortProfile()
449  self.pipe_port = RtcOutport(tmp, nvlist2dict(tmp.properties))
450  self.con = IOConnector([self.pipe_port,self.port])
451  def connect(self):
452  return self.con.connect()
453  def disconnect(self):
454  return self.con.disconnect()
455  def write(self, data) :
456  self.buf.data=data
457  self.pipe.write()
458 class OutPipe() :
459  def __init__(self,comp, port) :
460  self.comp=comp
461  self.port=port
462  self.pname=port.name.replace('.','_')
463  self.pipe=comp.makeInPort(self.pname,port.data_class(RTC.Time(0,0),[]),OpenRTM_aist.RingBuffer(1))
464  self.buf=getattr(comp,'_d_'+self.pname)
465  tmp = self.pipe.getPortProfile()
466  self.pipe_port = RtcInport(tmp, nvlist2dict(tmp.properties))
467  self.con = IOConnector([self.pipe_port,self.port])
468  def connect(self):
469  return self.con.connect()
470  def disconnect(self):
471  return self.con.disconnect()
472  def read(self) :
473  return self.pipe.read().data
474 #
475 #
476 #
477 def make_pipe(comp, handle) :
478  handle.in_pipe={}
479  for i_port in handle.inports :
480  handle.in_pipe[i_port]=InPipe(comp, handle.inports[i_port])
481  handle.out_pipe={}
482  for o_port in handle.outports :
483  handle.out_pipe[o_port]=OutPipe(comp, handle.outports[o_port])
def __init__(self, orb_args, nserver_names=["localhost"], orb=None, naming=None)
def __init__(self, plist, name=None, id="", prop_dict={})
def __init__(self, plist, name=None, id="", prop_dict={})
def __init__(self, plist, name=None, id="", prop_dict={})
def split(input, delimiter)
Split string by delimiter.
Definition: StringUtil.py:323
def __init__(self, profile, nv_dict=None, handle=None)
def set_conf_activate(self, conf_set_name, param_name, value)
def __init__(self, profile, nv_dict=None, handle=None)
def __init__(self, orb, server_name=None, naming=None)
def __init__(self, profile, nv_dict=None, handle=None)
def set_conf(self, conf_set_name, param_name, value)
def find(seq, f)
Return the index of CORBA sequence element that functor matches.
def __init__(self, profile, nv_dict=None, handle=None)


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