6 from omniORB
import CORBA, URI
7 from omniORB
import any
13 from CorbaNaming
import *
22 def __init__(self, orb_args, nserver_names=["localhost"],
23 orb=None, naming=None):
25 orb = CORBA.ORB_init(orb_args)
31 for ns
in nserver_names :
35 self.
orb.shutdown(wait_for_completion=CORBA.FALSE)
42 def __init__(self, orb, server_name=None, naming=None):
55 ref = self.
naming.resolveStr(name)
56 if ref
is None:
return None 58 return ref._narrow(cl)
69 name_context = self.
naming._rootContext
71 b_list = name_context.list(self.
b_len)
73 rslt = rslt + self.
proc_bd(bd, name_context, parent)
75 t_list = b_list[1].next_n(self.
b_len)
78 rslt = rslt + self.
proc_bd(bd, name_context, parent)
79 t_list = b_list[1].next_n(self.
b_len)
82 def proc_bd(self, bd, name_context, parent) :
91 nam = pre + URI.nameToString(bd.binding_name)
92 if bd.binding_type == CosNaming.nobject :
93 tmp = name_context.resolve(bd.binding_name)
95 print 'objcet '+nam+
' was listed.' 97 tmp = tmp._narrow(RTC.RTObject)
99 print nam+
' is not RTC.' 105 print 'handle for '+nam+
' was created.' 109 print nam+
' is not alive.' 112 tmp = name_context.resolve(bd.binding_name)
113 tmp = tmp._narrow(CosNaming.NamingContext)
123 rslt[tmp.name]=tmp.value.value()
127 for tmp
in dict.keys() :
128 rslt.append(SDOPackage.NameValue(tmp, any.to_any(dict[tmp])))
135 def __init__(self, plist, name = None, id="", prop_dict={}) :
141 self.
name = string.join([tmp.name
for tmp
in plist],
'_')
149 for kk
in self.def_prop :
155 for pp
in self.
plist :
156 if not ((self.
prop_dict[kk]
in pp.prop[kk])
or 157 (
'Any' in pp.prop[kk])) :
178 def __init__(self, plist, name = None, id="", prop_dict={}) :
183 'dataport.interface_type':
'corba_cdr' ,
184 'dataport.subscription_type':
'flush'}
185 Connector.__init__(self, plist, name, id, prop_dict)
188 def __init__(self, plist, name = None, id="", prop_dict={}) :
190 Connector.__init__(self, plist, name, id, prop_dict)
204 tmp2 = [pp.connector_id
for pp
in tmp1]
205 if self.
con.profile.connector_id
in tmp2 :
206 self.
con.disconnect()
209 return self.
port_profile.port_ref.get_connector_profiles()
215 self.
name = profile.instance_name
218 ref_key =
'port.' + self.
type +
'.' + self.
name 219 self.
ref=self.
port.con.prop_dict[ref_key]
235 self.
name = profile.instance_name
245 Port.__init__(self, profile, nv_dict)
252 if itf.polarity == RTC.PROVIDED :
254 elif itf.polarity == RTC.REQUIRED :
259 Port.__init__(self, profile, nv_dict)
263 self.
ref = self.
con.prop_dict[
'dataport.corba_cdr.inport_ref']
272 Port.__init__(self, profile, nv_dict)
275 if 'dataport.corba_any.outport_ref' in self.
con.prop_dict :
276 self.
ref = self.
con.prop_dict[
'dataport.corba_cdr.outport_ref']
285 return self.
ref.get().value()
287 print "not supported" 299 self.
rtc_ref = env.naming.resolve(name)._narrow(RTC.RTObject)
310 conf_set = self.
conf_ref.get_configuration_sets()
325 tmp = pp.get_port_profile()
328 if tmp_prop[
'port.port_type']==
'DataInPort' :
331 elif tmp_prop[
'port.port_type']==
'DataOutPort' :
334 elif tmp_prop[
'port.port_type']==
'CorbaPort' :
338 def set_conf(self,conf_set_name,param_name,value) :
339 conf_set=self.
conf_set[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)
345 self.
set_conf(conf_set_name,param_name,value)
346 self.
conf_ref.activate_configuration_set(conf_set_name)
def __init__(self, orb_args, nserver_names=["localhost"], orb=None, naming=None)
def get_object_by_name(self, name, cl=RTC.RTObject)
def __init__(self, name, data_buf, size=8)
def __init__(self, profile, nv_dict=None)
CORBA Naming Service helper class.
def __init__(self, plist, name=None, id="", prop_dict={})
def __init__(self, name, env, ref=None)
def set_conf_activate(self, conf_set_name, param_name, value)
def proc_bd(self, bd, name_context, parent)
def __init__(self, orb, server_name=None, naming=None)
def __init__(self, profile, nv_dict=None)
def __init__(self, profile, nv_dict=None)
def narrow_ref(self, gls)
def set_conf(self, conf_set_name, param_name, value)
def get_connections(self)
def find(seq, f)
Return the index of CORBA sequence element that functor matches.
def list_obj1(self, name_context, parent)
def __init__(self, plist, name=None, id="", prop_dict={})
def __init__(self, profile)
def __init__(self, plist, name=None, id="", prop_dict={})
def __init__(self, profile, port)
def __init__(self, profile, nv_dict=None)