Skip to content

process

Base classes for simulation of dynamical processes

BaseProcess

Abstract base class for all implementations of discrete-time dynamical processes.

Source code in src/pathpyG/processes/process.py 
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
class BaseProcess:
    """Abstract base class for all implementations of discrete-time dynamical processes.
    """

    def __init__(self, network: Graph):
        """initialize process."""
        self._network = network
        self.init(self.random_seed())

    @property
    def network(self) -> Graph:
        return self._network

    @abc.abstractmethod
    def init(self, seed: Any) -> None:
        """Abstract method to initialize the process with a given seed state."""

    @abc.abstractmethod
    def random_seed(self) -> Any:
        """Abstract method to generate a random seed state for the process."""

    @abc.abstractmethod
    def step(self) -> Iterable[str]:
        """Abstract method to simulate a single step of the process. Returns 
        an iterable of node uids whose state has been changed in this step."""

    @abc.abstractproperty
    def time(self) -> int:
        """Abstract property returning the current time."""

    @abc.abstractmethod
    def state_to_color(self, Any) -> Union[Tuple[int, int, int], str]:
        """Abstract method mapping node states to RGB colors or color names."""

    @abc.abstractmethod
    def node_state(self, v: str) -> Any:
        """Abstract method returning the current state of a given node."""

    def simulation_run(self, steps: int, seed: Optional[Any] = None) -> Tuple[int, Set[str]]:
        """Abstract generator method that initializes the process, runs a number of steps and yields a tuple consisting of the current time and the set of nodes whose state has changed in each step."""
        if seed == None:
            self.init(self.random_seed())
        else:
            self.init(seed)
        for _ in range(steps):
            ret = self.step()
            if ret is not None:
                yield self.time, ret
            else:
                return None

    def run_experiment(self, steps: int, runs: Optional[Union[int, Iterable[Any]]] = 1) -> DataFrame:
        """Perform one or more simulation runs of the process with a given number of steps."""

        # Generate initializations for different runs
        seeds: List = list()
        if type(runs) == int:
            for s in range(runs):
                seeds.append(self.random_seed())
        else:
            for s in runs:
                seeds.append(s)

        results = list()
        run_id: int = 0
        for seed in tqdm(seeds):

            # initialize seed state and record initial state
            self.init(seed)
            for v in self.network.nodes:
                results.append({'run_id': run_id, 'seed': seed,
                               'time': self.time, 'node': v, 'state': self.node_state(v)})

            # simulate the given number of steps
            for time, updated_nodes in self.simulation_run(steps, seed):
                # print(updated_nodes)
                # record the new state of each changed node
                for v in updated_nodes:
                    results.append({'run_id': run_id, 'seed': seed,
                                   'time': time, 'node': v, 'state': self.node_state(v)})
            run_id += 1

        return DataFrame.from_dict(results)

__init__

initialize process.

Source code in src/pathpyG/processes/process.py 
16
17
18
19
def __init__(self, network: Graph):
    """initialize process."""
    self._network = network
    self.init(self.random_seed())

init abstractmethod

Abstract method to initialize the process with a given seed state.

Source code in src/pathpyG/processes/process.py 
25
26
27
@abc.abstractmethod
def init(self, seed: Any) -> None:
    """Abstract method to initialize the process with a given seed state."""

node_state abstractmethod

Abstract method returning the current state of a given node.

Source code in src/pathpyG/processes/process.py 
46
47
48
@abc.abstractmethod
def node_state(self, v: str) -> Any:
    """Abstract method returning the current state of a given node."""

random_seed abstractmethod

Abstract method to generate a random seed state for the process.

Source code in src/pathpyG/processes/process.py 
29
30
31
@abc.abstractmethod
def random_seed(self) -> Any:
    """Abstract method to generate a random seed state for the process."""

run_experiment

Perform one or more simulation runs of the process with a given number of steps.

Source code in src/pathpyG/processes/process.py 
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
def run_experiment(self, steps: int, runs: Optional[Union[int, Iterable[Any]]] = 1) -> DataFrame:
    """Perform one or more simulation runs of the process with a given number of steps."""

    # Generate initializations for different runs
    seeds: List = list()
    if type(runs) == int:
        for s in range(runs):
            seeds.append(self.random_seed())
    else:
        for s in runs:
            seeds.append(s)

    results = list()
    run_id: int = 0
    for seed in tqdm(seeds):

        # initialize seed state and record initial state
        self.init(seed)
        for v in self.network.nodes:
            results.append({'run_id': run_id, 'seed': seed,
                           'time': self.time, 'node': v, 'state': self.node_state(v)})

        # simulate the given number of steps
        for time, updated_nodes in self.simulation_run(steps, seed):
            # print(updated_nodes)
            # record the new state of each changed node
            for v in updated_nodes:
                results.append({'run_id': run_id, 'seed': seed,
                               'time': time, 'node': v, 'state': self.node_state(v)})
        run_id += 1

    return DataFrame.from_dict(results)

simulation_run

Abstract generator method that initializes the process, runs a number of steps and yields a tuple consisting of the current time and the set of nodes whose state has changed in each step.

Source code in src/pathpyG/processes/process.py 
50
51
52
53
54
55
56
57
58
59
60
61
def simulation_run(self, steps: int, seed: Optional[Any] = None) -> Tuple[int, Set[str]]:
    """Abstract generator method that initializes the process, runs a number of steps and yields a tuple consisting of the current time and the set of nodes whose state has changed in each step."""
    if seed == None:
        self.init(self.random_seed())
    else:
        self.init(seed)
    for _ in range(steps):
        ret = self.step()
        if ret is not None:
            yield self.time, ret
        else:
            return None

state_to_color abstractmethod

Abstract method mapping node states to RGB colors or color names.

Source code in src/pathpyG/processes/process.py 
42
43
44
@abc.abstractmethod
def state_to_color(self, Any) -> Union[Tuple[int, int, int], str]:
    """Abstract method mapping node states to RGB colors or color names."""

step abstractmethod

Abstract method to simulate a single step of the process. Returns an iterable of node uids whose state has been changed in this step.

Source code in src/pathpyG/processes/process.py 
33
34
35
36
@abc.abstractmethod
def step(self) -> Iterable[str]:
    """Abstract method to simulate a single step of the process. Returns 
    an iterable of node uids whose state has been changed in this step."""

time

Abstract property returning the current time.

Source code in src/pathpyG/processes/process.py 
38
39
40
@abc.abstractproperty
def time(self) -> int:
    """Abstract property returning the current time."""