raphtory 0.17.0

raphtory, a temporal graph library
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

use crate::{
    core::state::{accumulator_id::accumulators, compute_state::ComputeStateVec},
    db::{
        api::{
            state::{GenericNodeState, TypedNodeState},
            view::StaticGraphViewOps,
        },
        task::{
            context::{Context, GlobalState},
            node::eval_node::EvalNodeView,
            task::{ATask, Job, Step},
            task_runner::TaskRunner,
        },
    },
    prelude::*,
};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

#[derive(Clone, PartialEq, Serialize, Deserialize, Debug, Default)]
pub struct LabelPropState {
    #[serde(skip)]
    nbors: HashMap<usize, usize>,
    pub community_id: usize,
}

/// Computes components using a label propagation algorithm
///
/// # Arguments
///
/// - `g` - A reference to the graph
/// - `iter_count` - Number of iterations
/// - `seed` - (Optional) Array of 32 bytes of u8 which is set as the rng seed
/// - `threads` - (Optional) Number of threads to use
///
/// # Returns
///
/// A vector of hashsets each containing nodes
///
pub fn label_propagation<G>(
    g: &G,
    iter_count: usize,
    _seed: Option<[u8; 32]>,
    threads: Option<usize>,
) -> TypedNodeState<'static, LabelPropState, G>
where
    G: StaticGraphViewOps,
{
    let mut ctx: Context<G, ComputeStateVec> = g.into();
    let global_diff = accumulators::sum::<usize>(2);
    ctx.global_agg_reset(global_diff);

    let step1 = ATask::new(move |s| {
        let id = s.node.index();
        let state: &mut LabelPropState = s.get_mut();
        state.community_id = id;
        Step::Continue
    });

    let step2 = ATask::new(move |s: &mut EvalNodeView<_, LabelPropState>| {
        let prev_id = s.prev().community_id;
        let nbor_iter = s.neighbours();
        let state: &mut LabelPropState = s.get_mut();
        state.nbors = HashMap::from([(prev_id, 1)]);
        // get labels from neighbors
        for nbor in nbor_iter {
            let nbor_id = nbor.prev().community_id;
            state
                .nbors
                .insert(nbor_id, *state.nbors.get(&nbor_id).unwrap_or(&(0)) + 1);
        }
        // get max label (use usize ID to resolve tie)
        if let Some((&label, _)) = state
            .nbors
            .iter()
            .max_by(|(k1, v1), (k2, v2)| v1.cmp(v2).then(k1.cmp(k2)))
        {
            state.community_id = label;
        }
        if state.community_id != prev_id {
            s.global_update(&global_diff, 1);
        }
        Step::Continue
    });

    let step3 = Job::Check(Box::new(move |state: &GlobalState<ComputeStateVec>| {
        if state.read(&global_diff) > 0 {
            Step::Continue
        } else {
            Step::Done
        }
    }));

    let mut runner: TaskRunner<G, _> = TaskRunner::new(ctx);
    runner.run(
        vec![Job::new(step1)],
        vec![Job::new(step2), step3],
        None,
        |_, _, _, local| {
            TypedNodeState::new(GenericNodeState::new_from_eval(g.clone(), local, None))
        },
        threads,
        iter_count,
        None,
        None,
    )
}