Bugfix/root chain point labels

script/install_tpls.py

@@ -0,0 +1,116 @@
+"""
+This script installs third-party libraries (TPLs) for running HPC-scale clustering.
+The script clones and builds a YGM-based partition comparison
+
+Usage:
+    # Assuming we are in 'build' directory
+    python ../script/install_tpls.py [-j number_of_jobs] [-d destination]
+"""
+
+
+import os
+import subprocess
+import argparse
+from datetime import datetime
+
+# Execute a shell command
+# out_file: output file to write the stdout
+# err_file: output file to write the stderr
+# command: command to execute
+# cwd: working directory
+def execute_cmd(log_file, err_log_file, command, cwd='./'):
+    print(f'In {os.path.abspath(cwd)}')
+    print(f'Command: {command}')
+    result = subprocess.run(f'{command}', shell=True,
+                            cwd=cwd,
+                            stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE)
+
+    with open(log_file, 'a') as f:
+        f.write(f'Command: {command}')
+        f.write(result.stdout.decode("utf8"))
+    with open(err_log_file, 'a') as f:
+        f.write(result.stderr.decode("utf8"))
+
+    if result.returncode == 0:
+        print("Command executed successfully.\n")
+    else:
+        print(f"Error executing command. See {err_log_file}.\n")
+        exit(1)
+
+
+def parse_options():
+    parser = argparse.ArgumentParser(
+        description='Install third-party libraries (TPLs) for'
+                    'running HPC-scale clustering')
+
+    # Install destination directory
+    parser.add_argument('-d', '--destination',
+                        dest='destination',
+                        required=False, action='store', type=str,
+                        default='./tpls',
+                        help='Destination directory to install TPLs')
+
+    # #of parallel jobs for make
+    parser.add_argument('-j', '--jobs',
+                        dest='jobs',
+                        required=False, action='store', type=int,
+                        default=1,
+                        help='#of parallel jobs for make')
+
+    options = parser.parse_args()
+    return options
+
+
+def main():
+
+    # Parse the command line arguments
+    opts = parse_options()
+    install_dir = os.path.abspath(opts.destination)
+    print(f"Install destination directory: {install_dir}\n")
+
+    # get timestamp to use it log file name
+    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+
+    out_log = f'out_install_tpls_{timestamp}.log'
+    with open(out_log, 'w') as f:
+        pass
+    print(f"Log file: {out_log}")
+
+    err_log = f'err_install_tpls_{timestamp}.log'
+    with open(err_log, 'w') as f:
+        pass
+    print(f"Error file: {err_log}\n")
+
+    # Check if the destination directory exists
+    if os.path.exists(install_dir):
+        # Ask the user if it is okay to delete the directory
+        response = input(
+            f"Directory {install_dir} exists. Do you want to delete it? (y/n): ")
+        if response.lower() == 'y':
+            print(f"Deleting directory {install_dir}")
+            execute_cmd(out_log, err_log, f'rm -rf {install_dir}')
+            execute_cmd(out_log, err_log, f'mkdir -p {install_dir}')
+        # else:
+        #     print("Exiting...")
+        #     exit(1)
+    else:
+        # Create the destination directory
+        execute_cmd(out_log, err_log, f'mkdir -p {install_dir}')
+
+    # Partition comparison with YGM
+    print('Cloning and building YGM-based partition-comparison...')
+    execute_cmd(out_log, err_log,
+                'git clone https://github.com/llnl/clams-cc.git', cwd=f'{install_dir}')
+    execute_cmd(out_log, err_log,
+                f'mkdir -p {install_dir}/clams-cc/build')
+    execute_cmd(out_log, err_log, 'cmake -DCMAKE_BUILD_TYPE=Release ..',
+                cwd=f'{install_dir}/clams-cc/build')
+    execute_cmd(out_log, err_log, f'make -j {opts.jobs}',
+                cwd=f'{install_dir}/clams-cc/build')
+
+    print("Installation complete.")
+
+
+if __name__ == "__main__":
+    main()

src/clustering/distributed_clustering_hdbscan.hpp

@@ -2432,18 +2432,24 @@ void get_valid_cluster_parent_child_relations(
 }
 
 /**
- * @brief Writes information on valid clusters to file, disregarding
- * min_cluster_size.
+ * @brief Writes information on valid clusters to file.
  *
  * @param cluster_file_path File to write cluster data to. Each rank must
  * receive its own file name (e.g., path_to_cluster_data/rank.csv).
  * @param valid_cluster_map YGM map of valid clusters mapping cluster name ->
  * valid cluster info.
+ *
+ * @return The sum of sizes of all selected clusters.
+ * This is a sanity check against the number of points clustered.
  */
-void write_valid_clusters_to_file(
+id_t write_valid_clusters_to_file(
     std::filesystem::path cluster_file_path,
     ygm::container::map<cluster_name_t, full_valid_cluster_info>
         &valid_cluster_map) {
+  ygm::comm &comm = valid_cluster_map.comm();
+
+  id_t num_points_clustered = 0;
+
   // Create output file stream
   std::ofstream ofs(cluster_file_path);
   if (!ofs.is_open()) {
@@ -2472,9 +2478,18 @@ void write_valid_clusters_to_file(
        << "\"" << cluster_info.children[1] << "\""
        << "," << cluster_info.num_points_added;
     ofs << ss.str() << "\n";
+
+    if (cluster_info.selected) {
+      num_points_clustered += cluster_info.size;
+    }
   }
   ofs.flush();
   ofs.close();
+  comm.barrier();
+
+  id_t total_num_points_clustered = ygm::sum(num_points_clustered, comm);
+
+  return total_num_points_clustered;
 }
 
 }  // namespace clams::clustering

src/clustering/run_distributed_hdbscan_clustering.cpp

@@ -1084,13 +1084,22 @@ int main(int argc, char *argv[]) {
     {
       sw_step.reset();
 
+      // Traverse down the child chains for cluster selection from all root
+      // chain clusters above the selected one (if there is one)
+      local_selected_clusters = traverse_down_hierarchy_and_select_clusters(
+          root_chain_cluster_map, chain_map, leaf_cluster_map,
+          root_chain_second_child, min_cluster_size, root_chain_supernode,
+          selected_root_chain_cluster_edge_id);
+      world.barrier();
+
       // If we selected a root chain cluster as a flat cluster, set it as
       // selected and add it to local selected clusters on rank 0. This is the
       // first cluster selected, so it will have cluster id 0.
       if (world.rank() == 0 && selected_root_chain_cluster_edge_id > 0) {
         cluster_name_t cluster_name = std::make_pair(
             root_chain_supernode, selected_root_chain_cluster_edge_id);
-        local_selected_clusters.push_back(cluster_name);
+        local_selected_clusters.insert(local_selected_clusters.begin(),
+                                       cluster_name);
         root_chain_cluster_map.async_visit(
             selected_root_chain_cluster_edge_id,
             []([[maybe_unused]] const id_t &cluster_edge_id,
@@ -1100,13 +1109,6 @@ int main(int argc, char *argv[]) {
       }
       world.barrier();
 
-      // Traverse down the child chains for cluster selection from all root
-      // chain clusters above the selected one (if there is one)
-      local_selected_clusters = traverse_down_hierarchy_and_select_clusters(
-          root_chain_cluster_map, chain_map, leaf_cluster_map,
-          root_chain_second_child, min_cluster_size, root_chain_supernode,
-          selected_root_chain_cluster_edge_id);
-
       select_clusters_time += sw_step.elapsed().count();
 
       id_t num_clusters       = local_selected_clusters.size();
@@ -1178,6 +1180,8 @@ int main(int argc, char *argv[]) {
   {
     sw_step.reset();
 
+    id_t num_points_clustered = 0;
+
     int num_files = world.size();
     if (world.rank() == 0) {
       spdlog::info("Writing {} files containing cluster labels to: {}",
@@ -1198,12 +1202,18 @@ int main(int argc, char *argv[]) {
     label_ofs << "# point_id\tcluster_id\n";
     for (const auto &[point_id, cluster_id] : point_to_cluster_id_map) {
       label_ofs << point_id << "\t" << cluster_id << "\n";
+      if (cluster_id >= 0) {
+        ++num_points_clustered;
+      }
     }
     label_ofs.flush();
     label_ofs.close();
     world.barrier();
 
+    id_t total_num_points_clustered = ygm::sum(num_points_clustered, world);
     if (world.rank() == 0) {
+      spdlog::info("Number of points clustered: {}",
+                   total_num_points_clustered);
       spdlog::info("  Time to write point labels (s): {:.3f}", sw_step);
     }
   }
@@ -1259,8 +1269,14 @@ int main(int argc, char *argv[]) {
       }
       sw_step.reset();
 
-      write_valid_clusters_to_file(cluster_file_path, valid_cluster_map);
+      id_t total_num_points_clustered =
+          write_valid_clusters_to_file(cluster_file_path, valid_cluster_map);
       world.barrier();
+
+      if (world.rank() == 0) {
+        spdlog::info("  Sum of sizes of selected clusters: {}",
+                     total_num_points_clustered);
+      }
     }
 
     if (world.rank() == 0) {

src/mst/build_amst.cpp

@@ -208,7 +208,8 @@ std::vector<edge_t> approx_mst(const std::vector<edge_t> &edges,
   c.barrier();
   double amst_time = amst_timer.elapsed();
 
-  c.cout0() << "Trees: " << amst_builder.num_trees()
+  c.cout0() << "Approx bound: " << approx_bound
+            << "\nTrees: " << amst_builder.num_trees()
             << "\nEdges: " << sum(amst_builder.get_result().size(), c)
             << "\nTime: " << amst_time << std::endl;
 
@@ -398,6 +399,9 @@ int main(int argc, char **argv) {
                                   ? read_dnnd_output(txt_input_filenames, world)
                                   : read_pm_knng_output(pm_input_path, world);
   world.cout0("Edge reading time (s): ", read_timer.elapsed());
+  world.cout0() << "Total number of edges read: "
+                << ygm::sum(edges.size(), world) << std::endl;
+  world.barrier();
 
   auto amst_edges = approx_mst(edges, amst_approx_bound, world);
   world.cout0() << "AMST edges: " << sum(amst_edges.size(), world) << std::endl;