host: add octoclock python api

Co-authored-by: Joerg Hofrichter <joerg.hofrichter@ni.com>

host/lib/usrp_clock/multi_usrp_clock_python.cpp

@@ -0,0 +1,35 @@
+//
+// Copyright 2024 Ettus Research, a National Instruments Brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+
+#include <pybind11/complex.h>
+#include <pybind11/pybind11.h>
+#include <pybind11/stl.h>
+
+namespace py = pybind11;
+
+#include "multi_usrp_clock_python.hpp"
+#include <uhd/usrp_clock/multi_usrp_clock.hpp>
+
+void export_multi_usrp_clock(py::module& m)
+{
+    using multi_usrp_clock = uhd::usrp_clock::multi_usrp_clock;
+
+    // clang-format off
+    py::class_<multi_usrp_clock, multi_usrp_clock::sptr>(m, "multi_usrp_clock")
+
+        // Factory
+        .def(py::init(&multi_usrp_clock::make))
+
+        // General multi_usrp_clock methods
+        .def("get_pp_string"           , &multi_usrp_clock::get_pp_string)
+        .def("get_num_boards"          , &multi_usrp_clock::get_num_boards)
+        .def("get_time"                , &multi_usrp_clock::get_time, py::arg("board") = 0)
+        .def("get_sensor"              , &multi_usrp_clock::get_sensor, py::arg("name"), py::arg("board") = 0)
+        .def("get_sensor_names"        , &multi_usrp_clock::get_sensor_names, py::arg("board") = 0)
+        // clang-format off
+        ;
+    // clang-format on
+}

host/lib/usrp_clock/multi_usrp_clock_python.hpp

@@ -0,0 +1,9 @@
+//
+// Copyright 2024 Ettus Research, a National Instruments Brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+
+#pragma once
+
+void export_multi_usrp_clock(py::module& m);

host/python/CMakeLists.txt

@@ -86,6 +86,7 @@ add_library(pyuhd SHARED
     ${UHD_SOURCE_DIR}/lib/property_tree_python.cpp
     ${UHD_SOURCE_DIR}/lib/device_python.cpp
     ${UHD_SOURCE_DIR}/lib/usrp/multi_usrp_python.cpp
+    ${UHD_SOURCE_DIR}/lib/usrp_clock/multi_usrp_clock_python.cpp
 )
 
 # python expects extension modules with a particular suffix

host/python/pyuhd.cpp

@@ -45,6 +45,7 @@ namespace py = pybind11;
 #include "usrp/fe_connection_python.hpp"
 #include "usrp/multi_usrp_python.hpp"
 #include "usrp/subdev_spec_python.hpp"
+#include "usrp_clock/multi_usrp_clock_python.hpp"
 #include "utils/paths_python.hpp"
 #include "utils/utils_python.hpp"
 #include "version_python.hpp"
@@ -93,6 +94,10 @@ PYBIND11_MODULE(libpyuhd, m)
     export_fe_connection(usrp_module);
     export_stream(usrp_module);
 
+    // Register usrp clock submodule
+    auto usrp_clock_module = m.def_submodule("usrp_clock", "USRP Clock Objects");
+    export_multi_usrp_clock(usrp_clock_module);
+
     // Register filters submodule
     auto filters_module = m.def_submodule("filters", "Filter Submodule");
     export_filters(filters_module);

host/python/uhd/__init__.py.in

@@ -27,7 +27,7 @@ if os.name == "nt" and hasattr(os, "add_dll_directory"):
                 "Failed to add uhd.dll path to DLL search, module may not import properly, use os.add_dll_directory to add the correct path manually."
             )
 
-from . import chdr, dsp, filters, rfnoc, types, usrp, usrpctl
+from . import chdr, dsp, filters, rfnoc, types, usrp, usrp_clock, usrpctl
 from .libpyuhd import find, get_abi_string, get_component, get_version_string
 from .libpyuhd.paths import *  # noqa: F403
 from .property_tree import PropertyTree

host/python/uhd/usrp_clock/__init__.py

@@ -0,0 +1,10 @@
+#
+# Copyright 2024 Ettus Research, a National Instruments Brand
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+""" @package usrp_clock
+Python UHD module containing the MultiUSRPClock
+"""
+
+from .multi_usrp_clock import MultiUSRPClock

host/python/uhd/usrp_clock/multi_usrp_clock.py

@@ -0,0 +1,20 @@
+#
+# Copyright 2024 Ettus Research, a National Instruments Brand
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+""" @package usrp_clock
+Python UHD module containing the MultiUSRPClock
+"""
+
+from .. import libpyuhd as lib
+
+
+class MultiUSRPClock(lib.usrp_clock.multi_usrp_clock):
+    """
+    MultiUSRPClock object for controlling devices
+    """
+    def __init__(self, args=""):
+        """MultiUSRPClock constructor"""
+        super(MultiUSRPClock, self).__init__(args)
+

host/tests/devtest/multi_usrp_clock_test.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+#
+# Copyright 2024 Ettus Research, a National Instruments Brand
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+""" Test all python API functions for the connected device. """
+
+import sys
+import argparse
+import uhd
+
+from multi_usrp_test import MethodExecutor
+
+
+# pylint: disable=broad-except
+
+def time_test(clock, num_mboards):
+    """
+    Test if each Octoclock (which is equipped with GPS) returns a time value.
+
+    The test is skipped on Octoclocks that are not equipped with GPS.
+
+    clock -- Device object to run tests on.
+    num_mboards -- Integer value of number of Octoclocks.
+    """
+    for mboard in range(0, num_mboards):
+        sensor_names = clock.get_sensor_names(mboard)
+        if 'gps_time' not in sensor_names:
+            print(f'[Octo {mboard}] gps_time sensor is not available, skipping time_test')
+            continue
+        time = clock.get_time(mboard)
+        if time is None:
+            raise Exception(f'get_time() function on Octoclock {mboard} returns None')
+        else:
+            print(f'[Octo {mboard}] Current time: {time}')
+    return True
+
+
+def test_sensor_api(clock, num_mboards):
+    """
+    Test the sensor API. It consists of two API calls ("get_sensor_names" and
+    "get_sensor"). It will read out all the sensors and print their values
+    for each octoclock.
+
+    clock -- Device object to run tests on.
+    num_mboards -- Integer value of number of Octoclocks.
+    """
+
+    for mboard in range(num_mboards):
+        sensor_names = clock.get_sensor_names(mboard)
+        print(f'[Octo {mboard}] Got {len(sensor_names)} sensors.')
+
+        for i, sensor_name in enumerate(sensor_names):
+            sensor_value = clock.get_sensor(sensor_name, mboard)
+            print(f'[{i}/{len(sensor_names)}] ({sensor_name}) {str(sensor_value)}')
+
+
+def run_api_test(clock):
+    """
+    Name functions to be tested.
+    clock -- device object to run tests on
+    """
+    num_mboards = clock.get_num_boards()
+
+    method_executor = MethodExecutor()
+
+    method_executor.tested.extend(('make',
+                                   'get_num_mboards', #  Got called above
+    ))
+
+    actual_tests = [
+        (['get_pp_string'], clock.get_pp_string),
+        (['get_sensor_names', 'get_sensor'],
+         lambda: test_sensor_api(clock, num_mboards)),
+        (['get_time'],
+         lambda: time_test(clock, num_mboards)),
+    ]
+
+    success = True
+    # ...then we can run them all through the executor like this:
+    for method_names, test_callback in actual_tests:
+        if not method_executor.execute_methods(method_names, test_callback):
+            success = False
+
+    if method_executor.failed:
+        print("The following API calls caused failures:")
+        print("* " + "\n* ".join(method_executor.failed))
+
+    return success
+
+
+def parse_args():
+    """
+    Parse args.
+    """
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        '--args', default='addr=192.168.10.3',
+    )
+    return parser.parse_args()
+
+
+def main():
+    """
+    Returns True on Success
+    """
+    args = parse_args()
+    clock = uhd.usrp_clock.MultiUSRPClock(args.args)
+    ret_val = run_api_test(clock)
+
+    if ret_val != 1:
+        raise Exception("Python API Tester Received Errors")
+    return ret_val
+
+if __name__ == "__main__":
+    sys.exit(not main())