test_translation.py 文件源码

def test_walk_up():
    raw = Stream()
    a_translation = FromEventStream(raw, 'start', ('time',))
    b_translation = FromEventStream(raw, 'event', ('data', 'pe1_image'))

    d = b_translation.zip_latest(a_translation)
    dd = d.map(op.truediv)
    e = ToEventStream(dd, ('data',))

    g = nx.DiGraph()
    walk_up(e, g)
    att = []
    for node, attrs in g.node.items():
        att.append(attrs['stream'])
    s = {a_translation, b_translation, d, dd, e}
    assert s == set(att)
    assert {hash(k) for k in s} == set(g.nodes)