push-pull-anomaly-tracking/push_pull_manager.py at main · signetlabdei/push-pull-anomaly-tracking · GitHub

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
 #
 # This file is part of the Push-Pull Medium Access repository:
 # https://github.com/signetlabdei/push-pull-anomaly-tracking
 # Copyright (c) 2025:
 # Fabio Saggese (fabio.saggese@ing.unipi.it)
 # Federico Chiariotti (federico.chiariotti@unipd.it)
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, version 3.
 #
 # This program is distributed in the hope that it will be useful, but
 # WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 # General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program. If not, see <http://www.gnu.org/licenses/>.
 #

import numpy as np

class ResourceManager:
    manager_type = 0
    P = 0
    R = 0
    min_res = 0
    hysteresis = 0

    def __init__(self, manager_type: int, total_resources: int):
        """ Constructor of the class.

        :param manager_type: 0: fixed, 1: proportional allocation given by the risk, 2: slow adaptive
        :param total_resources: int, number of total resource elements in the system :math:`R`
        """
        self.manager_type = manager_type
        self.R = total_resources
        self.P = int(np.floor(total_resources / 2))
        self.min_res = 1
        self.hysteresis = 0

    def set_push_resources(self, resources: int):
        self.P = resources

    def set_min_threshold(self, min_res: int):
        """Set minimum resources for push and pull. Used by manager types 1 and 2"""
        self.min_res = min_res

    def set_hysteresis(self, hyst: float):
        """Set hysteresis. Used only by manager type 2"""
        self.hysteresis = hyst

    def allocate_resources(self, local_risk: float = 0., twin_risk: float = 0.) -> tuple:
        """Allocate resources for push and pull  based on the local risk and digital twin risks."""
        # Manager is parametric
        if self.manager_type == 0:
            return self.P, self.R - self.P
        # Set push ratio depending on the risks
        push_ratio = 1
        if local_risk + twin_risk > 0:
            push_ratio = local_risk / (local_risk + twin_risk)
        # Step by step allocation
        if self.manager_type == 1:
            self.P = int(np.floor(self.R * push_ratio))
        # Adaptive rate allocation
        if self.manager_type == 2:
            if local_risk != twin_risk:
                if local_risk > twin_risk + self.hysteresis:
                    self.P += 1
                if local_risk < twin_risk - self.hysteresis:
                    self.P -= 1
        # Check minimum resources
        if self.P < self.min_res:
            self.P = self.min_res
        if self.P > self.R - self.min_res:
            self.P = self.R - self.min_res
        return self.P, self.R - self.P