InPort.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- coding: euc-jp -*-
 
 
 
from omniORB import *
from omniORB import any
import sys
import copy
import time
 
import OpenRTM_aist
 
 
class InPort(OpenRTM_aist.InPortBase):
  """
  """
 
 
 
  
  def __init__(self, name, value, buffer=None,
               read_block=False, write_block=False,
               read_timeout=0, write_timeout = 0):
    OpenRTM_aist.InPortBase.__init__(self, name, OpenRTM_aist.toTypename(value))
    self._name           = name
    self._value          = value
    self._OnRead         = None
    self._OnReadConvert  = None
 
 
  def __del__(self, InPortBase=OpenRTM_aist.InPortBase):
    InPortBase.__del__(self)
    return
 
  
  def name(self):
    return self._name
 
 
  
  def isNew(self):
    self._rtcout.RTC_TRACE("isNew()")
 
    if len(self._connectors) == 0:
      self._rtcout.RTC_DEBUG("no connectors")
      return False
 
    r = self._connectors[0].getBuffer().readable()
    if r > 0:
      self._rtcout.RTC_DEBUG("isNew() = True, readable data: %d",r)
      return True
 
    self._rtcout.RTC_DEBUG("isNew() = False, no readable data")
    return False
 
 
  
  def isEmpty(self):
    self._rtcout.RTC_TRACE("isEmpty()")
 
    if len(self._connectors) == 0:
      self._rtcout.RTC_DEBUG("no connectors")
      return True
 
    r = self._connectors[0].getBuffer().readable()
    if r == 0:
      self._rtcout.RTC_DEBUG("isEmpty() = true, buffer is empty")
      return True
      
    self._rtcout.RTC_DEBUG("isEmpty() = false, data exists in the buffer")
    return False
 
 
  
  def read(self):
    self._rtcout.RTC_TRACE("DataType read()")
 
    if self._OnRead is not None:
      self._OnRead()
      self._rtcout.RTC_TRACE("OnRead called")
 
    if len(self._connectors) == 0:
      self._rtcout.RTC_DEBUG("no connectors")
      return self._value
 
    _val = copy.deepcopy(self._value)
    cdr = [_val]
    ret = self._connectors[0].read(cdr)
 
 
    if ret == OpenRTM_aist.DataPortStatus.PORT_OK:
      self._rtcout.RTC_DEBUG("data read succeeded")
      self._value = cdr[0]
 
      if self._OnReadConvert is not None:
        self._value = self._OnReadConvert(self._value)
        self._rtcout.RTC_DEBUG("OnReadConvert called")
        return self._value
      return self._value
 
    elif ret == OpenRTM_aist.DataPortStatus.BUFFER_EMPTY:
      self._rtcout.RTC_WARN("buffer empty")
      return self._value
 
    elif ret == OpenRTM_aist.DataPortStatus.BUFFER_TIMEOUT:
      self._rtcout.RTC_WARN("buffer read timeout")
      return self._value
 
    self._rtcout.RTC_ERROR("unknown retern value from buffer.read()")
    return self._value
 
 
  
  def update(self):
    self.read()
 
 
  
  def setOnRead(self, on_read):
    self._OnRead = on_read
 
 
  
  def setOnReadConvert(self, on_rconvert):
    self._OnReadConvert = on_rconvert