test_PeriodicTask.py

Go to the documentation of this file.

#!/usr/bin/env/python
# -*- Python -*-
 
#
# \file PeriodicTask.py
# \brief PeriodicTask class
# \date $Date: $
# \author Shinji Kurihara
#
# Copyright (C) 2007
#     Task-intelligence Research Group,
#     Intelligent Systems Research Institute,
#     National Institute of
#         Advanced Industrial Science and Technology (AIST), Japan
#     All rights reserved.
#
 
import sys,time
sys.path.insert(1,"../")
 
import OpenRTM_aist
import unittest
 
from PeriodicTask import *
from TimeMeasure import *
 
class LoggerMock:
        def __init__(self):
                self.m_log = []
                return
 
        def log(self, msg):
                self.m_log.append(msg);
                return
 
        def countLog(self, msg):
                self.count = 0
                for i in range(len(self.m_log)):
                        if self.m_log[i] == msg:
                                self.count += 1
                return self.count
 
        def clearLog(self):
                self.m_log = []
                return
 
class MyMock:
        def __init__(self):
                return
 
        class mysvc2:
                def __init__(self):
                        self.m_logger = None
                        return
 
                def __call__(self):
                        if self.m_logger is not None:
                                self.m_logger.log("mysvc2")
                        return 0
 
                def setLogger(self, logger):
                        self.m_logger = logger
                        return
 
        class mysvc3:
                def __init__(self):
                        self.m_logger = None
                        return
 
                def __call__(self):
                        if self.m_logger is not None:
                                self.m_logger.log("mysvc3")
                        return 0
 
                def setLogger(self, logger):
                        self.m_logger = logger
                        return
 
class TestPeriodicTask(unittest.TestCase):
        def setUp(self):
                return
 
        def test_setTask(self):
                self._pt = PeriodicTask()
                self._my = MyMock()
                self._my.mysvc2 = MyMock().mysvc2()
                self._logger = LoggerMock()
                self._my.mysvc2.setLogger(self._logger)
 
                self.assertTrue(self._pt.setTask(self._my.mysvc2))
                self.assertEqual(0, self._logger.countLog("mysvc2"))
                self._pt.activate()
                time.sleep(0.005)
                self._pt.finalize()
                time.sleep(0.005)
                self.assertTrue(1 < self._logger.countLog("mysvc2"))
                return
 
        def test_setPeriodic(self):
                self._pt = PeriodicTask()
                self._my = MyMock()
                self._my.mysvc2 = MyMock().mysvc2()
                self._logger = LoggerMock()
                self._my.mysvc2.setLogger(self._logger)
 
                self._pt.setTask(self._my.mysvc2)
                self._pt.setPeriod(0.05)
                self.assertEqual(0, self._logger.countLog("mysvc2"))
 
                self._pt.activate()
                time.sleep(0.1)
                self._pt.suspend()
                time.sleep(0.05)
                self.assertTrue(4 > self._logger.countLog("mysvc2"))
                self.assertTrue(0 < self._logger.countLog("mysvc2"))
 
                self._logger.clearLog()
                self._pt.setPeriod(0.01)
                self.assertEqual(0, self._logger.countLog("mysvc2"))
 
                self._pt.resume()
                time.sleep(0.1)
                self._pt.suspend()
                time.sleep(0.01)
                self.assertTrue(12 > self._logger.countLog("mysvc2"))
                self.assertTrue( 8 < self._logger.countLog("mysvc2"))
 
                self._logger.clearLog()
                self._pt.setPeriod(0.05)
                self.assertEqual(0, self._logger.countLog("mysvc2"))
 
                self._pt.resume()
                time.sleep(0.1)
                self._pt.finalize()
                time.sleep(0.05)
                self.assertTrue(4 > self._logger.countLog("mysvc2"))
                self.assertTrue(0 < self._logger.countLog("mysvc2"))
                return
 
        def test_signal(self):
                self._pt = PeriodicTask()
                self._my = MyMock()
                self._my.mysvc2 = MyMock().mysvc2()
                self._logger = LoggerMock()
                self._my.mysvc2.setLogger(self._logger)
 
                self._pt.setTask(self._my.mysvc2)
                self._pt.setPeriod(0.05)
                self.assertEqual(0, self._logger.countLog("mysvc2"))
 
                self._pt.activate()
                time.sleep(0.2)
                self._pt.suspend()
                count = self._logger.countLog("mysvc2")
 
                time.sleep(0.2)
                self.assertEqual(count, self._logger.countLog("mysvc2"))
 
                self._pt.signal()
                time.sleep(0.2)
                self.assertEqual(count+1, self._logger.countLog("mysvc2"))
 
                self._pt.signal()
                time.sleep(0.2)
                self.assertEqual(count+2, self._logger.countLog("mysvc2"))
 
                self._logger.clearLog()
                self._pt.resume()
                time.sleep(0.2)
                self._pt.suspend()
                time.sleep(0.2)
                self.assertTrue(6 > self._logger.countLog("mysvc2"))
                self.assertTrue(2 < self._logger.countLog("mysvc2"))
 
                self._pt.finalize()
                time.sleep(0.2)
                return
 
        def test_executionMeasure(self):
                self._pt = PeriodicTask()
                self._my = MyMock()
                self._my.mysvc3 = MyMock().mysvc3()
                self._logger = LoggerMock()
                self._my.mysvc3.setLogger(self._logger)
 
                wait = 0.03     # sec
                self._pt.setTask(self._my.mysvc3)
                self._pt.setPeriod(0.05)
                self._pt.executionMeasure(True)
 
                # executionMeasureConut: 10
                self._pt.activate()
                time.sleep(0.6)
                self._pt.suspend()
                time.sleep(0.05)
                estat = self._pt.getExecStat()
                ss = ""
                ss = ss + "wait:  " + str(wait) + "\n"
                ss = ss + "estat max:  " + str(estat._max_interval) + "\n"
                ss = ss + "estat min:  " + str(estat._min_interval) + "\n"
                ss = ss + "estat mean: " + str(estat._mean_interval) + "\n"
                ss = ss + "estat sdev: " + str(estat._std_deviation) + "\n"
                self.assertTrue(estat._max_interval < (wait + 0.030), ss)
                self.assertTrue(estat._min_interval > (wait - 0.03), ss)
                self.assertTrue(abs(estat._mean_interval - wait) < 0.03, ss)
                self.assertTrue(estat._std_deviation < (wait / 5.0), ss)
 
                # executionMeasureConut: 5
                self._pt.executionMeasureCount(5)
                self._pt.resume()
                time.sleep(0.3)
                self._pt.suspend()
                time.sleep(0.05)
                estat = self._pt.getExecStat()
                ss = ""
                ss = ss + "wait:  " + str(wait) + "\n"
                ss = ss + "estat max:  " + str(estat._max_interval) + "\n"
                ss = ss + "estat min:  " + str(estat._min_interval) + "\n"
                ss = ss + "estat mean: " + str(estat._mean_interval) + "\n"
                ss = ss + "estat sdev: " + str(estat._std_deviation) + "\n"
                self.assertTrue(estat._max_interval < (wait + 0.030), ss)
                self.assertTrue(estat._min_interval > (wait - 0.03), ss)
                self.assertTrue(abs(estat._mean_interval - wait) < 0.03, ss)
                self.assertTrue(estat._std_deviation < (wait / 5.0), ss)
 
                # executionMeasureConut: lessthan
                self._pt.executionMeasureCount(10)
                self._pt.resume()
                time.sleep(0.3)
                self._pt.suspend()
                time.sleep(0.05)
                self._pt.finalize()
                estat2 = self._pt.getExecStat()
 
                # periodicMeasureConut: lessthan
                self.assertTrue(estat._max_interval == estat2._max_interval)
                self.assertTrue(estat._min_interval == estat2._min_interval)
                self.assertTrue(estat._mean_interval == estat2._mean_interval)
                self.assertTrue(estat._std_deviation == estat2._std_deviation)
                return
 
        def test_periodicMeasure(self):
                self._pt = PeriodicTask()
                self._my = MyMock()
                self._my.mysvc3 = MyMock().mysvc3()
                self._logger = LoggerMock()
                self._my.mysvc3.setLogger(self._logger)
 
                wait = 0.05     # sec
                self._pt.setTask(self._my.mysvc3)
                self._pt.setPeriod(0.05)
                self._pt.periodicMeasure(True)
 
                # periodicMeasureConut: 10
                self._pt.activate()
                time.sleep(0.6)
                self._pt.suspend()
                time.sleep(0.05)
                pstat = self._pt.getPeriodStat()
                self.assertTrue(pstat._max_interval < (wait + 0.030))
                self.assertTrue(pstat._min_interval > (wait - 0.050))
                self.assertTrue(abs(pstat._mean_interval - wait) < 0.03)
                self.assertTrue(pstat._std_deviation < (wait / 1.0))
 
                # periodicMeasureConut:5
                self._pt.periodicMeasureCount(5)
                self._pt.resume()
                time.sleep(0.3)
                self._pt.suspend()
                time.sleep(0.05)
                pstat = self._pt.getPeriodStat()
                self.assertTrue(pstat._max_interval < (wait + 0.030))
                self.assertTrue(pstat._min_interval > (wait - 0.010))
                self.assertTrue(abs(pstat._mean_interval - wait) < 0.03)
                self.assertTrue(pstat._std_deviation < (wait / 5.0))
 
                # periodicMeasureConut: lessthan
                self._pt.periodicMeasureCount(10)
                self._pt.resume()
                time.sleep(0.3)
                self._pt.suspend()
                time.sleep(0.05)
                self._pt.finalize()
                pstat2 = self._pt.getPeriodStat()
 
                # periodicMeasureConut: lessthan
                # Comment: Some errors may be observed.
                #self.assert_(pstat._max_interval == pstat2._max_interval)
                #self.assert_(pstat._min_interval == pstat2._min_interval)
                #self.assert_(pstat._mean_interval == pstat2._mean_interval)
                #self.assert_(pstat._std_deviation == pstat2._std_deviation)
                return
 
 
 
if __name__ == '__main__':
        unittest.main()