deimos 0.16.2

Control-loop and data pipeline for the Deimos data acquisition system
Documentation 
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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

"""
A showcase of drivers for imitating hardware from software
to test the software's interface with hardware.
"""

import os
import platform
import time
from contextlib import ExitStack
from pathlib import Path
from deimos import Controller, peripheral, socket, Termination, LoopMethod

HAS_UNIX_SOCKET = hasattr(socket, "UnixSocket") and hasattr(
    peripheral.HootlTransport, "unix_socket"
)


def _loopback_udp_socket() -> socket.UdpSocket:
    targets = socket.UdpSocket.possible_broadcast_targets()
    if not targets:
        raise RuntimeError("No UDP broadcast targets available for loopback test")
    return socket.UdpSocket.with_broadcast_targets([targets[0]])


def _should_retry_under_test() -> bool:
    """When running on MacOS in CI, we may need to retry several times
    due to issues with non-modifiable configuration of the CI runners."""
    return (
        os.environ.get("DEIMOS_TESTING", "").lower() == "true"
        and platform.system() == "Darwin"
    )


def _run_once() -> None:
    here = Path(__file__).parent.resolve()

    for loop_method in [LoopMethod.performant(), LoopMethod.efficient()]:
        print(f"Testing with loop method {loop_method}")

        # Set up HOOTL peripherals (drivers are attached via the controller)
        ctrl = Controller(op_name="mockup_demo", op_dir=str(here / "op"), rate_hz=20.0)
        ctrl.termination_criteria = Termination.timeout_s(2.0)
        ctrl.loop_method = loop_method

        ctrl.clear_sockets()
        ctrl.add_socket("mockup_chan", socket.ThreadChannelSocket("mockup_chan"))
        if HAS_UNIX_SOCKET:
            ctrl.add_socket("ctrl", socket.UnixSocket("ctrl"))
        ctrl.add_socket("udp", _loopback_udp_socket())  # Included by default, but cleared

        ctrl.add_peripheral("mock_thread", peripheral.DeimosDaqRev6(1))
        if HAS_UNIX_SOCKET:
            ctrl.add_peripheral("mock_unix", peripheral.DeimosDaqRev6(2))
        ctrl.add_peripheral("mock_udp", peripheral.DeimosDaqRev6(3))

        # Run
        with ExitStack() as stack:
            # Run the peripheral mockups, which will wait for the controller
            # to send a request to bind
            stack.enter_context(
                ctrl.attach_hootl_driver(
                    "mock_thread",
                    peripheral.HootlTransport.thread_channel("mockup_chan"),
                )
            )
            if HAS_UNIX_SOCKET:
                stack.enter_context(
                    ctrl.attach_hootl_driver(
                        "mock_unix",
                        peripheral.HootlTransport.unix_socket("mockup_unix"),
                    )
                )
            stack.enter_context(
                ctrl.attach_hootl_driver(
                    "mock_udp",
                    peripheral.HootlTransport.udp(),
                )
            )

            # Get list of inputs available to set manually.
            # This is also available from the RunHandle during operation.
            manual_inputs = ctrl.available_inputs()
            print("Manual inputs available:")
            for name in manual_inputs[:3]:  # Adjust slicing to list more input names
                print(f"    {name}")
            print(f"    ...and {len(manual_inputs) - 3} more")

            # Run the controller, which will bind the peripheral mockups
            start = time.perf_counter()
            handle = ctrl.run_nonblocking()

            try:
                time.sleep(0.2)  # Accommodate slow CI runners
                handle.write({"mock_thread.dac0": 0.0})

                # Make sure we had stable communication with all the peripheral mockups
                for k, v in handle.read().values.items():
                    if "loss_of_contact_counter" in k:
                        # When testing on macos runners, we're not able to set
                        # core affinity, which causes sporadic packet loss.
                        assert v < 2.0, f"Missed packet: {k} = {v:.0f}"
            except Exception:
                end = time.perf_counter()
                print(
                    "Sending termination signal to run handle"
                    f" from Python after {end - start:.2f}s."
                )
                handle.stop()
                raise
            finally:
                handle.join()


def main() -> None:
    """Run once under normal conditions, or retry up to 10 times on MacOS under test."""
    attempts = 10 if _should_retry_under_test() else 1
    last_exc: Exception | None = None

    for attempt in range(1, attempts + 1):
        try:
            if attempts > 1:
                print(f"macOS test run attempt {attempt}/{attempts}")
            _run_once()
            return
        except Exception as exc:
            last_exc = exc
            if attempt == attempts:
                raise
            print(
                f"macOS test run attempt {attempt}/{attempts} failed;"
                " retrying after cleanup."
            )
            time.sleep(0.2)

    if last_exc is not None:
        raise last_exc


if __name__ == "__main__":
    main()