pepu.py 2.74 KB
Newer Older
Vincent Michel's avatar
Vincent Michel committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89
"""PEPU hardware tests.

Run with:

    $ pytest -c /dev/null tests/controllers_hw/pepu.py -v \
    --cov bliss.controllers.pepu --cov-report html --cov-report term

"""

import pytest
from bliss.controllers.pepu import PEPU, Signal, Trigger


@pytest.fixture
def pepu():
    pepu = PEPU('test', {'tcp': {'url': 'pepudcm2'}})
    try:
        pepu.calc_channels[1].formula = '1.5'
        pepu.calc_channels[2].formula = '-1.5'
        pepu.out_channels[7].source = pepu.calc_channels[1].name
        pepu.out_channels[8].source = pepu.calc_channels[2].name
        yield pepu
    finally:
        pepu.conn.close()


def test_simple_connection(pepu):
    assert pepu.app_name == 'PEPU'
    assert pepu.version == '00.01'
    assert pepu.up_time > 0
    assert pepu.sys_info.startswith('DANCE')
    uptime, uname = pepu.dance_info.splitlines()
    assert uptime.startswith('UPTIME')
    assert uname.startswith('UNAME')
    assert pepu.config.startswith('# %APPNAME% PEPU')


@pytest.mark.parametrize("channel_id", range(1, 7))
def test_read_in_channels(pepu, channel_id):
    channel = pepu.in_channels[channel_id]
    assert channel.value in [-1., 0.]


@pytest.mark.parametrize("channel_id", [7, 8])
def test_read_out_channels(pepu, channel_id):
    channel = pepu.out_channels[channel_id]
    value = channel.value
    pytest.xfail()
    assert value in (1.5, -1.5)


@pytest.mark.parametrize("channel_id", [1, 2])
def test_read_calc_channels(pepu, channel_id):
    channel = pepu.calc_channels[channel_id]
    value = channel.value
    pytest.xfail()
    assert value in (1.5, -1.5)


@pytest.mark.parametrize("acquisitions", [1, 2, 10])
@pytest.mark.parametrize("blocks", [1, 2, 10])
@pytest.mark.parametrize("block_size", [1, 2, 10])
def test_streams_acquisition(pepu, acquisitions, blocks, block_size):
    # Create stream
    stream = pepu.create_stream(
        name='TEST',
        trigger=Trigger(Signal.SOFT, Signal.SOFT),
        frequency=10, nb_points=blocks * block_size,
        sources=('CALC1', 'CALC2'),
        overwrite=True)
    # Argument testing
    assert stream.name == 'TEST'
    assert stream.trigger == Trigger(Signal.SOFT, Signal.SOFT)
    assert stream.frequency == 10
    assert stream.nb_points == blocks * block_size
    assert stream.sources == ['CALC1', 'CALC2']
    assert not stream.active
    # Loop over acquisitions
    for _ in range(acquisitions):
        stream.start()
        # Loop over blocks
        for _ in range(blocks):
            # Loop over points
            for _ in range(block_size):
                pepu.software_trigger()
            # Read block
            assert stream.nb_points_ready == block_size
            data = stream.read(n=block_size)
            expected = [[1.5, -1.5]] * block_size
90
            assert data.tolist() == expected