COCTestRTC.py
Go to the documentation of this file.
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3 # -*- Python -*-
4 
5 import sys
6 
7 import RTC
8 import OpenRTM_aist
9 
10 coctestrtc_spec = ["implementation_id", "COCTestRTC",
11  "type_name", "COCTestRTC",
12  "description", "Console input component",
13  "version", "1.0",
14  "vendor", "Shinji Kurihara",
15  "category", "example",
16  "activity_type", "DataFlowComponent",
17  "max_instance", "10",
18  "language", "Python",
19  "lang_type", "script",
20  ""]
21 
22 
24  def __init__(self, name):
25  self._name = name
26 
27  def __del__(self):
28  print "dtor of ", self._name
29 
30  def __call__(self, info, cdrdata):
31  data = OpenRTM_aist.ConnectorDataListenerT.__call__(self, info, cdrdata, RTC.TimedLong(RTC.Time(0,0),0))
32  print "------------------------------"
33  print "Listener: ", self._name
34  print "Profile::name: ", info.name
35  print "Profile::id: ", info.id
36  print "Data: ", data.data
37  print "------------------------------"
38 
40  def __init__(self, name):
41  self._name = name
42 
43  def __del__(self):
44  print "dtor of ", self._name
45 
46  def __call__(self, info):
47  print "------------------------------"
48  print "Listener: ", self._name
49  print "Profile::name: ", info.name
50  print "Profile::id: ", info.id
51  print "------------------------------"
52 
53 
55  def __init__(self, manager):
56  OpenRTM_aist.DataFlowComponentBase.__init__(self, manager)
57  return
58 
59  def onInitialize(self):
60  self._data = RTC.TimedLong(RTC.Time(0,0),0)
61  self._outport = OpenRTM_aist.OutPort("out", self._data)
62  # Set OutPort buffer
63  self.addOutPort("out", self._outport)
64  self._outport.addConnectorDataListener(OpenRTM_aist.ConnectorDataListenerType.ON_BUFFER_WRITE,
65  DataListener("ON_BUFFER_WRITE"))
66  self._outport.addConnectorDataListener(OpenRTM_aist.ConnectorDataListenerType.ON_BUFFER_FULL,
67  DataListener("ON_BUFFER_FULL"))
68  self._outport.addConnectorDataListener(OpenRTM_aist.ConnectorDataListenerType.ON_BUFFER_WRITE_TIMEOUT,
69  DataListener("ON_BUFFER_WRITE_TIMEOUT"))
70  self._outport.addConnectorDataListener(OpenRTM_aist.ConnectorDataListenerType.ON_BUFFER_OVERWRITE,
71  DataListener("ON_BUFFER_OVERWRITE"))
72  self._outport.addConnectorDataListener(OpenRTM_aist.ConnectorDataListenerType.ON_BUFFER_READ,
73  DataListener("ON_BUFFER_READ"))
74  self._outport.addConnectorDataListener(OpenRTM_aist.ConnectorDataListenerType.ON_SEND,
75  DataListener("ON_SEND"))
76  self._outport.addConnectorDataListener(OpenRTM_aist.ConnectorDataListenerType.ON_RECEIVED,
77  DataListener("ON_RECEIVED"))
78  self._outport.addConnectorDataListener(OpenRTM_aist.ConnectorDataListenerType.ON_RECEIVER_FULL,
79  DataListener("ON_RECEIVER_FULL"))
80  self._outport.addConnectorDataListener(OpenRTM_aist.ConnectorDataListenerType.ON_RECEIVER_TIMEOUT,
81  DataListener("ON_RECEIVER_TIMEOUT"))
82  self._outport.addConnectorDataListener(OpenRTM_aist.ConnectorDataListenerType.ON_RECEIVER_ERROR,
83  DataListener("ON_RECEIVER_ERROR"))
84 
85  self._outport.addConnectorListener(OpenRTM_aist.ConnectorListenerType.ON_CONNECT,
86  ConnListener("ON_CONNECT"))
87  self._outport.addConnectorListener(OpenRTM_aist.ConnectorListenerType.ON_DISCONNECT,
88  ConnListener("ON_DISCONNECT"))
89 
90  return RTC.RTC_OK
91 
92 
93  def onExecute(self, ec_id):
94  print "Please input number: ",
95  self._data.data = long(sys.stdin.readline())
96  OpenRTM_aist.setTimestamp(self._data)
97  print "Sending to subscriber: ", self._data.data
98  self._outport.write()
99  return RTC.RTC_OK
100 
101 
102 def COCTestRTCInit(manager):
103  profile = OpenRTM_aist.Properties(defaults_str=coctestrtc_spec)
104  manager.registerFactory(profile,
105  COCTestRTC,
106  OpenRTM_aist.Delete)
107 
108 
109 def MyModuleInit(manager):
110  COCTestRTCInit(manager)
111 
112  # Create a component
113  comp = manager.createComponent("COCTestRTC")
114 
115 def main():
116  # Initialize manager
117  mgr = OpenRTM_aist.Manager.init(sys.argv)
118 
119  # Set module initialization proceduer
120  # This procedure will be invoked in activateManager() function.
121  mgr.setModuleInitProc(MyModuleInit)
122 
123  # Activate manager and register to naming service
124  mgr.activateManager()
125 
126  # run the manager in blocking mode
127  # runManager(False) is the default
128  mgr.runManager()
129 
130  # If you want to run the manager in non-blocking mode, do like this
131  # mgr.runManager(True)
132 
133 if __name__ == "__main__":
134  main()
The Properties class represents a persistent set of properties.
Definition: Properties.py:83
This class is abstract base class for listener classes that provides callbacks for various events in ...
def addOutPort(self, name, outport)
Definition: RTObject.py:2765
This class is abstract base class for listener classes that provides callbacks for various events in ...


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