Add tests for signal generator

Author: callumforrester

examples/configs/signal-generator/signal-generator.yaml

@@ -1,4 +1,4 @@
-- type: tickit.devices.signal_generator.EpicsSignalGenerator
+- type: tickit_devices.signal_generator.signal_generator.EpicsSignalGenerator
   name: gen
   inputs: {}
 - type: tickit.devices.sink.Sink

src/tickit_devices/signal_generator/signal_generator.py

@@ -6,7 +6,6 @@
 import pydantic.v1.dataclasses
 from pydantic.v1 import Field
 from softioc import builder
-
 from tickit.adapters.epics import EpicsAdapter
 from tickit.adapters.io import EpicsIo
 from tickit.core.adapter import AdapterContainer
@@ -82,7 +81,7 @@ def set_amplitude(self, amplitude: float) -> None:
         self._wave.amplitude = amplitude
 
     def get_amplitude_offset(self) -> float:
-        return self._wave.amplitude
+        return self._wave.amplitude_offset
 
     def set_amplitude_offset(self, amplitude_offset: float) -> None:
         self._wave.amplitude_offset = amplitude_offset
@@ -119,10 +118,10 @@ def _compute_wave(self, time: SimTime) -> float:
 
     def _sine(self, time: SimTime) -> float:
         return self._wave.amplitude_offset + (
-            self._wave.amplitude * self._sinosoid(time)
+            self._wave.amplitude * self._sinusoid(time)
         )
 
-    def _sinosoid(self, time: SimTime) -> float:
+    def _sinusoid(self, time: SimTime) -> float:
         time_seconds = time * 1e-9
         return math.sin(2 * math.pi * self._wave.frequency * time_seconds)
 

tests/signal_generator/__init__.py

@@ -0,0 +1,102 @@
+from dataclasses import dataclass
+from typing import List
+
+import pytest
+from tickit.core.typedefs import SimTime
+
+from tickit_devices.signal_generator.signal_generator import (
+    SignalGeneratorDevice,
+    WaveConfig,
+)
+
+
+@pytest.fixture
+def signal_generator() -> SignalGeneratorDevice:
+    config = WaveConfig()
+    return SignalGeneratorDevice(config)
+
+
+TIMES = [
+    SimTime(0),
+    SimTime(0.25e9),
+    SimTime(0.5e9),
+    SimTime(0.75e9),
+    SimTime(1e9),
+    SimTime(1.25e9),
+]
+
+EXPECTED_SINE_VALUES = [
+    1.0,
+    2.0,
+    1.0,
+    0.0,
+    1.0,
+    2.0,
+]
+
+EXPECTED_GATE_STATES = [
+    True,
+    True,
+    True,
+    False,
+    True,
+    True,
+]
+
+EXPECTED_OUTPUTS = [
+    SignalGeneratorDevice.Outputs(value=1.0, gate=True),
+    SignalGeneratorDevice.Outputs(value=2.0, gate=True),
+    SignalGeneratorDevice.Outputs(value=1.0, gate=True),
+    SignalGeneratorDevice.Outputs(value=0.0, gate=False),
+    SignalGeneratorDevice.Outputs(value=1.0, gate=True),
+    SignalGeneratorDevice.Outputs(value=2.0, gate=True),
+]
+
+
+def test_signal_generator_produces_sine_wave(
+    signal_generator: SignalGeneratorDevice,
+):
+    sine_values = []
+    for sim_time in TIMES:
+        signal_generator.update(sim_time, {})
+        sine_values.append(signal_generator.get_value())
+    assert sine_values == pytest.approx(EXPECTED_SINE_VALUES)
+
+
+def test_signal_generator_can_be_disabled(
+    signal_generator: SignalGeneratorDevice,
+):
+    sine_values = []
+    signal_generator.set_enabled(False)
+    for sim_time in TIMES:
+        signal_generator.update(sim_time, {})
+        sine_values.append(signal_generator.get_value())
+    assert sine_values == [0.0] * 6
+
+
+def test_signal_generator_trips_gate(
+    signal_generator: SignalGeneratorDevice,
+):
+    gate_states = []
+    for sim_time in TIMES:
+        signal_generator.update(sim_time, {})
+        gate_states.append(signal_generator.is_gate_open())
+    assert gate_states == EXPECTED_GATE_STATES
+
+
+def test_signal_generator_produces_outputs(
+    signal_generator: SignalGeneratorDevice,
+):
+    outputs = []
+    for sim_time, expected_output in zip(TIMES, outputs):
+        output = signal_generator.update(sim_time, {})
+        assert output["gate"] == expected_output["gate"]
+        assert output["value"] == pytest.approx(expected_output["value"])
+
+
+def test_config_passes(signal_generator: SignalGeneratorDevice):
+    default_wave_config = WaveConfig()
+    signal_generator.get_amplitude() == default_wave_config.amplitude
+    signal_generator.get_amplitude_offset() == default_wave_config.amplitude_offset
+    signal_generator.get_frequency() == default_wave_config.frequency
+    signal_generator.get_gate_threshold() == 0.5

tests/signal_generator/test_signal_generator.py

@@ -0,0 +1,51 @@
+import asyncio
+from dataclasses import dataclass
+from typing import Any
+
+import aioca
+import pytest
+
+PV_PREFIX = "SIGNALGEN"
+AMPLITUDE = f"{PV_PREFIX}:Amplitude"
+AMPLITUDE_RBV = f"{AMPLITUDE}_RBV"
+OFFSET = f"{PV_PREFIX}:Offset"
+OFFSET_RBV = f"{OFFSET}_RBV"
+FREQUENCY = f"{PV_PREFIX}:Frequency"
+FREQUENCY_RBV = f"{FREQUENCY}_RBV"
+GATE_THRESHOLD = f"{PV_PREFIX}:GateThreshold"
+GATE_THRESHOLD_RBV = f"{GATE_THRESHOLD}_RBV"
+ENABLED = f"{PV_PREFIX}:Enabled"
+ENABLED_RBV = f"{ENABLED}_RBV"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "tickit_process",
+    ["examples/configs/signal-generator/signal-generator.yaml"],
+    indirect=True,
+)
+async def test_signal_generator_system(tickit_process):
+    await set_and_test(AMPLITUDE, AMPLITUDE_RBV, 1.0, 2.0)
+    await set_and_test(OFFSET, OFFSET_RBV, 1.0, 2.0)
+    await set_and_test(FREQUENCY, FREQUENCY, 1.0, 2.0)
+    await set_and_test(GATE_THRESHOLD, GATE_THRESHOLD_RBV, 0.5, 0.25)
+    await set_and_test(ENABLED, ENABLED_RBV, True, False)
+
+
+async def set_and_test(pv: str, rbv: str, expected_initial_value: Any, new_value: Any):
+    initial_value = await aioca.caget(rbv)
+    await aioca.caput(pv, new_value, wait=True)
+    await asyncio.sleep(1.0)
+    actual_demand_value, actual_new_value = await asyncio.gather(
+        *[aioca.caget(pv), aioca.caget(rbv)]
+    )
+
+    assert initial_value == expected_initial_value, (
+        f"Initially caget({pv}) was {initial_value}, expected {expected_initial_value}",
+    )
+    assert actual_demand_value == new_value, (
+        f"After caput({pv}, {new_value}), {pv} was actually {actual_new_value}"
+    )
+    assert actual_new_value == new_value, (
+        f"After caput({pv}, {new_value}), {rbv} was actually {actual_new_value}"
+    )