<commit_before>from time import sleep

import numpy as np
from pytest import approx, raises
from tests.input.mockdevices import Dummy, DummyList, SingleResp, Timebomb

from toon.input.clock import mono_clock
from toon.input.mpdevice import MpDevice

# bump up the sampling frequency for tests
Dummy.sampling_frequency = 1000
DummyList.sampling_frequency = 1000


def test_device_single():
    # single device with two data sources
    dev = MpDevice(Dummy())
    dev.start()
    sleep(0.2)
    res = dev.read()
    dev.stop()
    assert(len(res.num1.time) > 5)
    assert(len(res.num1.time) == res.num1.shape[0])
    assert(res.num1.shape[1] == 5)
    assert(res.num2.dtype == np.int32)


def test_single_resp():
    dev = MpDevice(SingleResp())
    dev.start()
    sleep(0.2)
    res = dev.read()
    dev.stop()
    assert(isinstance(res, np.ndarray))
    assert(type(res.time) is np.ndarray)


def test_ringbuffer():
    original_fs = SingleResp.sampling_frequency
    SingleResp.sampling_frequency = 10
    dev = MpDevice(SingleResp())
    with dev:
        sleep(5)
        res = dev.read()
    assert(all(np.diff(res) == 1))
    assert(all(np.diff(res.time) > 0))
    SingleResp.sampling_frequency = original_fs


def test_have_all_data():
    dev = MpDevice(SingleResp())
    datae = []
    times = []
    with dev:
        for i in range(100):
            sleep(0.2)
            data = dev.read()
            if data is not None:
                datae.append(np.copy(data))
                times.append(np.copy(data.time))
    times = np.hstack(times)
    datae = np.hstack(datae)
    # check time is always increasing
    assert((np.diff(times) > 0).all())
    # check data (should be monotonically increasing)
    assert((np.diff(datae) == 1).all())


def test_device_list():
    # two observations per read on the device
    dev = MpDevice(DummyList())
    dev.start()
    sleep(0.2)
    res = dev.read()
    dev.stop()
    assert(len(res.num1.time) > 5)
    assert(len(res.num1.time) == res.num1.shape[0])
    assert(res.num1.shape[1] == 5)
    assert(res.num2.dtype == np.int32)
    assert(res.num1.shape[0] > res.num2.shape[1])


def test_restart():
    # start & stop device
    dev = MpDevice(Dummy())
    dev.start()
    sleep(0.2)
    res = dev.read()
    dev.stop()
    dev.start()
    sleep(0.2)
    res2 = dev.read()
    dev.stop()
    assert(res.any() and res2.any())


def test_context():
    # device as context manager
    dev = MpDevice(Dummy())
    with dev:
        sleep(0.2)
        res = dev.read()
    assert(len(res.num1.time) > 5)
    assert(len(res.num1.time) == res.num1.shape[0])
    assert(res.num1.shape[1] == 5)
    assert(res.num2.dtype == np.int32)


def test_multi_devs():
    # 2+ devices at once (each gets own process)
    dev1 = MpDevice(Dummy())
    dev2 = MpDevice(Dummy())
    with dev1, dev2:
        sleep(0.1)
        res1 = dev1.read()
        res2 = dev2.read()

    assert(res1.num1.time is not None)
    assert(res2.num1.time is not None)


def test_remote_clock():
    # does the clock origin match?<commit_msg>Test for the issue addressed in the previous commit<commit_after>from time import sleep

import numpy as np
from pytest import approx, raises
from tests.input.mockdevices import Dummy, DummyList, SingleResp, Timebomb

from toon.input.clock import mono_clock
from toon.input.mpdevice import MpDevice

# bump up the sampling frequency for tests
Dummy.sampling_frequency = 1000
DummyList.sampling_frequency = 1000


def test_device_single():
    # single device with two data sources
    dev = MpDevice(Dummy())
    dev.start()
    sleep(0.2)
    res = dev.read()
    dev.stop()
    assert(len(res.num1.time) > 5)
    assert(len(res.num1.time) == res.num1.shape[0])
    assert(res.num1.shape[1] == 5)
    assert(res.num2.dtype == np.int32)


def test_single_resp():
    dev = MpDevice(SingleResp())
    dev.start()
    sleep(0.2)
    res = dev.read()
    dev.stop()
    assert(isinstance(res, np.ndarray))
    assert(type(res.time) is np.ndarray)


def test_ringbuffer():
    original_fs = SingleResp.sampling_frequency
    SingleResp.sampling_frequency = 10
    dev = MpDevice(SingleResp())
    with dev:
        sleep(5)
        res = dev.read()
    assert(all(np.diff(res) == 1))
    assert(all(np.diff(res.time) > 0))
    SingleResp.sampling_frequency = original_fs


def test_have_all_data():
    dev = MpDevice(SingleResp())
    datae = []
    times = []
    with dev:
        for i in range(100):
            sleep(0.2)
            data = dev.read()
            if data is not None:
                datae.append(np.copy(data))
                times.append(np.copy(data.time))
    times = np.hstack(times)
    datae = np.hstack(datae)
    # check time is always increasing
    assert((np.diff(times) > 0).all())
    # check data (should be monotonically increasing)
    assert((np.diff(datae) == 1).all())


def test_device_list():
    # two observations per read on the device
    dev = MpDevice(DummyList())
    dev.start()
    sleep(0.2)
    res = dev.read()
    dev.stop()
    assert(len(res.num1.time) > 5)
    assert(len(res.num1.time) == res.num1.shape[0])
    assert(res.num1.shape[1] == 5)
    assert(res.num2.dtype == np.int32)
    assert(res.num1.shape[0] > res.num2.shape[1])


def test_context():
    # device as context manager
    dev = MpDevice(Dummy())
    with dev:
        sleep(0.2)
        res = dev.read()
    assert(len(res.num1.time) > 5)
    assert(len(res.num1.time) == res.num1.shape[0])
    assert(res.num1.shape[1] == 5)
    assert(res.num2.dtype == np.int32)


def test_restart():
    # start & stop device
    dev = MpDevice(Dummy())
    with dev:
        sleep(0.2)
        res = dev.read()
    with dev:
        sleep(0.2)
        res2 = dev.read()
    assert(res.any() and res2.any())


def test_reuse():
    local_dev = Dummy()
    dev = MpDevice(local_dev)
    # device is exclusive to remote
    with dev:
        sleep(0.2)
        res = dev.read()
    # should be able to use locally now
    with local_dev:
        res2 = local_dev.do_read()
    # lock again
    with dev:
        sleep(0.2)
        res3 = dev.read()
    print(res)
    print(res2)
    print(res3)
    assert(res.any())
    assert(res2.any())
    assert(res3.any())


def test_multi_devs():
    # 2+ devices at once (each gets own process)
    dev1 = MpDevice(Dummy())
    dev2 = MpDevice(Dummy())
    with dev1, dev2:
        sleep(0.1)
        res1 = dev1.read()
        res2 = dev2.read()

    assert(res1.num1.time is not None)
    assert(res2.num1.time is not None)


def test_remote_clock():
    # does the clock origin match?
{'fraction_non_alphanumeric': 0.13524962178517397, 'fraction_numerical': 0.027534039334341905, 'mean_word_length': 2.6224657534246574, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3489847', 'n_tokens_mistral': 2433, 'n_tokens_neox': 2337, 'n_words': 527}