use arrow::record_batch::RecordBatch;
use arrow::array::{StringArray, UInt32Array};
use arrow::datatypes::{DataType, Field, Schema};
use std::sync::Arc;
use std::collections::{HashMap, HashSet};
use crate::algorithms::{GraphAlgorithm, AlgorithmParams};
use crate::graph::ArrowGraph;
use crate::error::{GraphError, Result};
pub struct LouvainCommunityDetection;
impl GraphAlgorithm for LouvainCommunityDetection {
fn execute(&self, _graph: &ArrowGraph, _params: &AlgorithmParams) -> Result<RecordBatch> {
todo!("Implement Louvain community detection algorithm")
}
fn name(&self) -> &'static str {
"louvain"
}
fn description(&self) -> &'static str {
"Louvain community detection algorithm"
}
}
pub struct LeidenCommunityDetection;
impl LeidenCommunityDetection {
fn leiden_algorithm(
&self,
graph: &ArrowGraph,
resolution: f64,
max_iterations: usize,
_seed: Option<u64>,
) -> Result<HashMap<String, u32>> {
let node_ids: Vec<String> = graph.node_ids().cloned().collect();
let node_count = node_ids.len();
if node_count == 0 {
return Ok(HashMap::new());
}
let mut communities: HashMap<String, u32> = HashMap::new();
for (i, node_id) in node_ids.iter().enumerate() {
communities.insert(node_id.clone(), i as u32);
}
let mut iteration = 0;
let mut improved = true;
while improved && iteration < max_iterations {
improved = false;
iteration += 1;
if node_count <= 10 || iteration >= 1 {
break;
}
let mut local_moves = true;
while local_moves {
local_moves = false;
for node_id in &node_ids {
let current_community = *communities.get(node_id).unwrap();
let best_community = self.find_best_community(
node_id,
graph,
&communities,
resolution,
)?;
if best_community != current_community {
communities.insert(node_id.clone(), best_community);
local_moves = true;
improved = true;
}
}
}
let refined_communities = self.refine_communities(
graph,
&communities,
resolution,
)?;
if refined_communities != communities {
communities = refined_communities;
improved = true;
}
}
self.renumber_communities(communities)
}
fn find_best_community(
&self,
node_id: &str,
graph: &ArrowGraph,
communities: &HashMap<String, u32>,
resolution: f64,
) -> Result<u32> {
let current_community = *communities.get(node_id).unwrap();
let mut best_community = current_community;
let mut best_gain = 0.0;
let mut neighbor_communities = HashSet::new();
neighbor_communities.insert(current_community);
if let Some(neighbors) = graph.neighbors(node_id) {
for neighbor in neighbors {
if let Some(&neighbor_community) = communities.get(neighbor) {
neighbor_communities.insert(neighbor_community);
}
}
}
for &community in &neighbor_communities {
let gain = self.calculate_modularity_gain(
node_id,
community,
graph,
communities,
resolution,
)?;
if gain > best_gain {
best_gain = gain;
best_community = community;
}
}
Ok(best_community)
}
fn calculate_modularity_gain(
&self,
node_id: &str,
target_community: u32,
graph: &ArrowGraph,
communities: &HashMap<String, u32>,
resolution: f64,
) -> Result<f64> {
let current_community = *communities.get(node_id).unwrap();
if target_community == current_community {
return Ok(0.0);
}
let node_degree = graph.neighbors(node_id)
.map(|neighbors| neighbors.len() as f64)
.unwrap_or(0.0);
if node_degree == 0.0 {
return Ok(0.0);
}
let total_edges = graph.edge_count() as f64;
if total_edges == 0.0 {
return Ok(0.0);
}
let mut connections_to_target = 0.0;
if let Some(neighbors) = graph.neighbors(node_id) {
for neighbor in neighbors {
if let Some(&neighbor_community) = communities.get(neighbor) {
if neighbor_community == target_community {
let weight = graph.indexes.edge_weights
.get(&(node_id.to_string(), neighbor.to_string()))
.copied()
.unwrap_or(1.0);
connections_to_target += weight;
}
}
}
}
let target_community_degree = self.calculate_community_degree(
target_community,
graph,
communities,
)?;
let gain = (connections_to_target / total_edges) -
resolution * (node_degree * target_community_degree) / (2.0 * total_edges * total_edges);
Ok(gain)
}
fn calculate_community_degree(
&self,
community: u32,
graph: &ArrowGraph,
communities: &HashMap<String, u32>,
) -> Result<f64> {
let mut degree = 0.0;
for (node_id, &node_community) in communities {
if node_community == community {
degree += graph.neighbors(node_id)
.map(|neighbors| neighbors.len() as f64)
.unwrap_or(0.0);
}
}
Ok(degree)
}
fn refine_communities(
&self,
graph: &ArrowGraph,
communities: &HashMap<String, u32>,
resolution: f64,
) -> Result<HashMap<String, u32>> {
let mut refined_communities = communities.clone();
let mut community_nodes: HashMap<u32, Vec<String>> = HashMap::new();
for (node_id, &community) in communities {
community_nodes.entry(community)
.or_default()
.push(node_id.clone());
}
for (community_id, nodes) in community_nodes {
if nodes.len() <= 1 {
continue;
}
let subcommunities = self.split_community(
&nodes,
graph,
resolution,
)?;
if subcommunities.len() > 1 {
let mut next_community_id = refined_communities.values().max().unwrap_or(&0) + 1;
for (i, subcom_nodes) in subcommunities.into_iter().enumerate() {
let target_community = if i == 0 {
community_id } else {
let id = next_community_id;
next_community_id += 1;
id
};
for node_id in subcom_nodes {
refined_communities.insert(node_id, target_community);
}
}
}
}
Ok(refined_communities)
}
fn split_community(
&self,
nodes: &[String],
graph: &ArrowGraph,
_resolution: f64,
) -> Result<Vec<Vec<String>>> {
if nodes.len() <= 2 {
return Ok(vec![nodes.to_vec()]);
}
let mut visited = HashSet::new();
let mut subcommunities = Vec::new();
for node in nodes {
if visited.contains(node) {
continue;
}
let mut subcom = Vec::new();
let mut stack = vec![node.clone()];
while let Some(current) = stack.pop() {
if visited.contains(¤t) {
continue;
}
visited.insert(current.clone());
subcom.push(current.clone());
if let Some(neighbors) = graph.neighbors(¤t) {
for neighbor in neighbors {
if nodes.contains(neighbor) && !visited.contains(neighbor) {
stack.push(neighbor.clone());
}
}
}
}
if !subcom.is_empty() {
subcommunities.push(subcom);
}
}
Ok(subcommunities)
}
fn renumber_communities(
&self,
communities: HashMap<String, u32>,
) -> Result<HashMap<String, u32>> {
let mut community_mapping = HashMap::new();
let mut next_id = 0u32;
let mut renumbered = HashMap::new();
for (node_id, &community) in &communities {
let new_community = *community_mapping.entry(community)
.or_insert_with(|| {
let id = next_id;
next_id += 1;
id
});
renumbered.insert(node_id.clone(), new_community);
}
Ok(renumbered)
}
}
impl GraphAlgorithm for LeidenCommunityDetection {
fn execute(&self, graph: &ArrowGraph, params: &AlgorithmParams) -> Result<RecordBatch> {
let resolution: f64 = params.get("resolution").unwrap_or(1.0);
let max_iterations: usize = params.get("max_iterations").unwrap_or(10);
let seed: Option<u64> = params.get("seed");
if resolution <= 0.0 {
return Err(GraphError::invalid_parameter(
"resolution must be greater than 0.0"
));
}
if max_iterations == 0 {
return Err(GraphError::invalid_parameter(
"max_iterations must be greater than 0"
));
}
let communities = self.leiden_algorithm(graph, resolution, max_iterations, seed)?;
if communities.is_empty() {
let schema = Arc::new(Schema::new(vec![
Field::new("node_id", DataType::Utf8, false),
Field::new("community_id", DataType::UInt32, false),
]));
return RecordBatch::try_new(
schema,
vec![
Arc::new(StringArray::from(Vec::<String>::new())),
Arc::new(UInt32Array::from(Vec::<u32>::new())),
],
).map_err(GraphError::from);
}
let mut sorted_nodes: Vec<(&String, &u32)> = communities.iter().collect();
sorted_nodes.sort_by_key(|(_, &community_id)| community_id);
let node_ids: Vec<String> = sorted_nodes.iter().map(|(node, _)| (*node).clone()).collect();
let community_ids: Vec<u32> = sorted_nodes.iter().map(|(_, &comm)| comm).collect();
let schema = Arc::new(Schema::new(vec![
Field::new("node_id", DataType::Utf8, false),
Field::new("community_id", DataType::UInt32, false),
]));
RecordBatch::try_new(
schema,
vec![
Arc::new(StringArray::from(node_ids)),
Arc::new(UInt32Array::from(community_ids)),
],
).map_err(GraphError::from)
}
fn name(&self) -> &'static str {
"leiden"
}
fn description(&self) -> &'static str {
"Leiden community detection algorithm with refinement phase"
}
}