cogen_test

In [2]:

    

from cogen.core.coroutines import coro
from cogen.core.schedulers import Scheduler


    

In [4]:

    

@coro
def a(foo, times):
    for i in range(times):
        print foo, ':', i
        yield


    

In [5]:

    

sched=Scheduler()


    

In [6]:

    

sched.add(a, args=('foo', 5))


    

    
Out[6]:





<a Coroutine instance at 0x01671AD8 wrapping <function a at 0x134c4f0>, state: NOTSTARTED>

In [8]:

    

sched.add(a, args=('bar', 10))


    

    
Out[8]:





<a Coroutine instance at 0x0139EFA8 wrapping <function a at 0x134c4f0>, state: NOTSTARTED>

In [9]:

    

sched.run()


    

    




foo : 0
bar : 0
foo : 1
bar : 1
foo : 2
bar : 2
foo : 3
bar : 3
foo : 4
bar : 4
bar : 5
bar : 6
bar : 7
bar : 8
bar : 9

In [ ]:

    

import time
import numpy as np
times=np.zeros(shape=[100], dtype=np.float32)


    

In [49]:

    

@coro
def time_test(times):
    last_time=time.time()
    print times
    for i in range(100):
        now=time.time()
        times[0][i]=now-last_time
#         print i, times[i]
        last_time=now
        time.sleep(.01)
        yield


    

In [63]:

    

times1=np.zeros(shape=[100], dtype=np.float32)
times2=np.zeros(shape=[100], dtype=np.float32)


    

In [52]:

    

sched.add?


    

In [34]:

    

sched.add(time_test, args=times1)


    

    
Out[34]:





<time_test Coroutine instance at 0x013A3FA8 wrapping <function time_test at 0x134c5b0>, state: NOTSTARTED>

In [64]:

    

sched.add(time_test, args=(times2))


    

    
Out[64]:





<time_test Coroutine instance at 0x01387E48 wrapping <function time_test at 0x13a29b0>, state: NOTSTARTED>

In [65]:

    

sched.run()


    

    




----------------------------------------
Exception happened during processing of coroutine.
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/cogen/core/coroutines.py", line 218, in run_op
    self.coro = self.coro(*self.f_args, **self.f_kws)
TypeError: time_test() takes exactly 1 argument (100 given)
Coroutine <time_test Coroutine instance at 0x01387E48 wrapping <function time_test at 0x13a29b0>, state: FAILED> killed. 
----------------------------------------

In [13]:

    

%matplotlib inline


    

In [14]:

    

import matplotlib.pyplot as plt


    

In [28]:

    

plt.plot(times2)
plt.xlabel


    

    
Out[28]:





<function matplotlib.pyplot.xlabel>






    

In [ ]: