AutoTestIn.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- coding:utf-8 -*-



import sys
import string
import time
sys.path.append(".")

# Import RTM module
import RTC
import OpenRTM_aist
import math
import AutoTest, AutoTest__POA
import os.path

# </rtc-template>

# Import Service stub modules
# <rtc-template block="consumer_import">
# </rtc-template>


# This module's spesification
# <rtc-template block="module_spec">
AutoTestIn_spec = ["implementation_id", "AutoTestIn", 
                   "type_name",         "AutoTestIn", 
                   "description",       "ModuleDescription", 
                   "version",           "1.0.0", 
                   "vendor",            "HarumiMiyamoto", 
                   "category",          "example", 
                   "activity_type",     "STATIC", 
                   "max_instance",      "1", 
                   "language",          "Python", 
                   "lang_type",         "SCRIPT",
                   "exec_cxt.periodic.rate", "1.0",
                   ""]
# </rtc-template>


# Class implementing IDL interface MyService(MyService.idl)
class MyServiceSVC_impl(AutoTest__POA.MyService):
  def __init__(self):
    self._echoList = []
    self._valueList = []
    self._value = 0
    self.__echo_msg= ""
    self._isNew = False

  def __del__(self):
    pass

  def echo(self, msg):
    OpenRTM_aist.CORBA_SeqUtil.push_back(self._echoList, msg)
    self.__echo_msg = msg
    if self._isNew:
      #print "echo's message was overwritten !!!"
      pass
    self._isNew = True
    return msg

  def get_echo(self):
    if self._isNew:
      self._isNew = False
      echomsg = self.__echo_msg
      return echomsg

    return ""

  def reset_message(self):
    self._isNew = False
    self.__echo_msg = ""

class AutoTestIn(OpenRTM_aist.DataFlowComponentBase):
  
  """
  \class AutoTestIn
  \brief ModuleDescription
  
  """
  def __init__(self, manager):
    """
    \brief constructor
    \param manager Maneger Object
    """
    OpenRTM_aist.DataFlowComponentBase.__init__(self, manager)
    self._cnt=0
    self._flag=0

  def onInitialize(self):
    self._d_dp_vIn = RTC.TimedFloat(RTC.Time(0,0),0)
    self._d_dp_vSeqIn = RTC.TimedFloatSeq(RTC.Time(0,0),[])

    self._InIn = OpenRTM_aist.InPort("in", self._d_dp_vIn, OpenRTM_aist.RingBuffer(8))
    self._SeqInIn = OpenRTM_aist.InPort("seqin", self._d_dp_vSeqIn, OpenRTM_aist.RingBuffer(8))
    

    # Set InPort buffers
    self.addInPort("in",self._InIn)
    self.addInPort("seqin",self._SeqInIn)
    
    # Set OutPort buffers
    self._MyServicePort = OpenRTM_aist.CorbaPort("MyService")
    #self._myservice0_var = MyService_i()
    
    # initialization of Provider
    self._myservice0_var = MyServiceSVC_impl()

    # Set service provider to Ports
    self._MyServicePort.registerProvider("myservice0", "MyService", self._myservice0_var)
    
    # Set CORBA Service Ports
    self.addPort(self._MyServicePort)

    return RTC.RTC_OK

  
  def onActivated(self, ec_id):
    self._file=open('received-data','w')
    self._msg = ""
    return RTC.RTC_OK
  
  def onDeactivated(self, ec_id): 
    self._file.close()
    self._myservice0_var.reset_message()
    return RTC.RTC_OK
  
  def onExecute(self, ec_id):
    if not self._msg:
      self._msg = self._myservice0_var.get_echo()
      if self._msg:
        self._msg += "\n"

    if self._InIn.isNew() and self._SeqInIn.isNew() and self._msg:
      floatdata = self._InIn.read()
      fdata = "%.6f\n" % floatdata.data
      print "fdata:", fdata
      self._file.write(fdata)
      
      seqfloatdata = self._SeqInIn.read()
      
      t_sdata = tuple(d for d in seqfloatdata.data)
      sdata = "%.6f %.6f %.6f %.6f %.6f\n" % t_sdata

      print "sdata: ", sdata
      self._file.write(sdata)
      
      self._file.write(self._msg)
      self._msg = ""

      return RTC.RTC_OK
    else:
      return RTC.RTC_OK
  



def AutoTestInInit(manager):
  profile = OpenRTM_aist.Properties(defaults_str=AutoTestIn_spec)
  manager.registerFactory(profile,
                          AutoTestIn,
                          OpenRTM_aist.Delete)

def MyModuleInit(manager):
  AutoTestInInit(manager)

  # Create a component
  comp = manager.createComponent("AutoTestIn")



def main():
  mgr = OpenRTM_aist.Manager.init(len(sys.argv), sys.argv)
  mgr.setModuleInitProc(MyModuleInit)
  mgr.activateManager()
  mgr.runManager()

if __name__ == "__main__":
  main()