test add test for cpp_temporal_reduciton w. io

test/CMakeLists.txt

@@ -1023,4 +1023,15 @@ teca_add_test(test_cpp_temporal_reduction_inmem
     SOURCES test_cpp_temporal_reduction_inmem.cpp
     LIBS teca_core teca_data teca_io teca_alg ${teca_test_link}
     COMMAND test_cpp_temporal_reduction_inmem
-    360 180 5 24 1 1 1 monthly average 1 test_cpp_temporal_reduction_%t%.nc yearly 1 1 1 1)
+    360 180 5 24 1 ${TEST_CORES} ${TEST_CORES} monthly average 1 #TODO -- more than 1 step per req
+    test_cpp_temporal_reduction_inmem_%t%.nc yearly ${TEST_CORES} 1 1 1)
+
+teca_add_test(test_cpp_temporal_reduction_io
+    SOURCES test_cpp_temporal_reduction_io.cpp
+    LIBS teca_core teca_data teca_io teca_alg ${teca_test_link}
+    COMMAND test_cpp_temporal_reduction_io
+    ${TECA_DATA_ROOT}/prw_hus_day_MRI-CGCM3_historical_r1i1p1_19500101-19501231\.nc
+    lon lat . time prw 0 -1 ${TEST_CORES} ${TEST_CORES} ${TEST_CORES}
+    1 0 monthly average 1 # TODO -- more than 1 step per req.
+    ./test_cpp_temporal_reduction_io_%t%.nc yearly
+    ${TEST_CORES} ${TEST_CORES} ${TEST_CORES})

test/test_cpp_temporal_reduction_io.cpp

@@ -0,0 +1,110 @@
+#include "teca_config.h"
+#include "teca_metadata.h"
+#include "teca_cf_reader.h"
+#include "teca_temporal_reduction.h"
+#include "teca_cf_writer.h"
+#include "teca_mpi_manager.h"
+#include "teca_system_interface.h"
+
+#include <iostream>
+#include <vector>
+#include <tuple>
+#include <cmath>
+#include <cstring>
+
+int main(int argc, char **argv)
+{
+    teca_mpi_manager mpi_man(argc, argv);
+    int rank = mpi_man.get_comm_rank();
+    int n_ranks = mpi_man.get_comm_size();
+
+    teca_system_interface::set_stack_trace_on_error();
+    teca_system_interface::set_stack_trace_on_mpi_error();
+
+    if (argc != 22)
+    {
+        std::cerr << "test_temporal_reduction [files regex]"
+            " [x axis var] [y axis var] [z axis var] [t axis var]"
+            " [red var] [first step] [last step]"
+            " [red threads] [red threads per dev] [red ranks per dev] [red bind threads] [red prop dev]"
+            " [reduction interval] [reduction operator] [steps per request]"
+            " [out file] [file layout] [wri threads] [wri threads per dev] [wri ranks per dev]" << std::endl;
+        return -1;
+    }
+
+    std::string files_regex = argv[1];
+    std::string x_axis_var = argv[2];
+    std::string y_axis_var = argv[3];
+    std::string z_axis_var = argv[4];
+    std::string t_axis_var = argv[5];
+    std::string red_var = argv[6];
+    int first_step = atoi(argv[7]);
+    int last_step = atoi(argv[8]);
+
+    int n_red_threads = atoi(argv[9]);
+    int red_threads_per_dev = atoi(argv[10]);
+    int red_ranks_per_dev = atoi(argv[11]);
+    int red_bind_threads = atoi(argv[12]);
+    int red_prop_dev_id = atoi(argv[13]);
+
+    std::string red_int = argv[14];
+    std::string red_op = argv[15];
+    int steps_per_req = atoi(argv[16]);
+
+    std::string ofile_name = argv[17];
+    std::string layout = argv[18];
+    int n_wri_threads = atoi(argv[19]);
+    int wri_threads_per_dev = atoi(argv[20]);
+    int wri_ranks_per_dev = atoi(argv[21]);
+
+
+    if (rank == 0)
+    std::cerr << "n_ranks=" << n_ranks
+        << "  n_red_threads=" << n_red_threads << "  red_threads_per_dev=" << red_threads_per_dev
+        << "  red_ranks_per_dev=" << red_ranks_per_dev << "  red_bind_threads=" << red_bind_threads
+        << "  red_pro_dev_id=" << red_prop_dev_id << "  red_int=" << red_int << "  red_op=" << red_op
+        << "  steps_per_req=" << steps_per_req
+        << "  layout=" << layout << "  n_wri_threads=" << n_wri_threads
+        << "  wri_threads_per_dev=" << wri_threads_per_dev
+        << "  wri_ranks_per_dev=" << wri_ranks_per_dev
+        << std::endl;
+
+    // reader
+    auto cf_reader = teca_cf_reader::New();
+    cf_reader->set_x_axis_variable(x_axis_var);
+    cf_reader->set_y_axis_variable(y_axis_var);
+    cf_reader->set_z_axis_variable(z_axis_var == "." ? std::string() : z_axis_var);
+    cf_reader->set_t_axis_variable(t_axis_var);
+    cf_reader->set_files_regex(files_regex);
+
+    // temporal reduction
+    auto reduc = teca_cpp_temporal_reduction::New();
+    reduc->set_input_connection(cf_reader->get_output_port());
+    reduc->set_verbose(1);
+    reduc->set_threads_per_device(red_threads_per_dev);
+    reduc->set_ranks_per_device(red_ranks_per_dev);
+    reduc->set_bind_threads(red_bind_threads);
+    reduc->set_propagate_device_assignment(red_prop_dev_id);
+    reduc->set_thread_pool_size(n_red_threads);
+    reduc->set_interval(red_int);
+    reduc->set_operation(red_op);
+    reduc->set_point_arrays({red_var});
+    reduc->set_steps_per_request(steps_per_req);
+
+    // writer
+    auto cfw = teca_cf_writer::New();
+    cfw->set_input_connection(reduc->get_output_port());
+    cfw->set_verbose(1);
+    cfw->set_threads_per_device(wri_threads_per_dev);
+    cfw->set_ranks_per_device(wri_ranks_per_dev);
+    cfw->set_thread_pool_size(n_wri_threads);
+    cfw->set_file_name(ofile_name);
+    cfw->set_layout(layout);
+    cfw->set_point_arrays({red_var});
+    cfw->set_first_step(first_step);
+    cfw->set_last_step(last_step);
+
+    cfw->update();
+
+    return 0;
+}