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

---

openrtm_aist_python
Author(s): Shinji Kurihara
autogenerated on Thu Jun 6 2019 19:11:34