Skip to content

Entity

Source code in Akatosh\entity.py 
 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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class Entity:

    def __init__(
        self,
        at: float | Event,
        till: float | Event,
        label: Optional[str] = None,
        priority: int = 0,
    ) -> None:
        """Create an entity with a creation and termination event.

        Args:
            at (float | Event): when the entity is created.
            till (float | Event): when the entity is terminated.
            label (Optional[str], optional): short description to the entity. Defaults to None.
            priority (int, optional): the priority of the entity, only impacts the creation and termination event. Defaults to 0.
        """
        self._label = label
        self._at = at
        self._till = till
        self._created = False
        self._terminated = False
        self._priority = priority

        # create an instant creation event
        self._creation = Event(
            at=at,
            till=till,
            action=self._create,
            label=f"{self} Creation",
            once=True,
            priority=self.priority,
        )
        self._termination = Event(
            at=till,
            till=till,
            action=self._terminate,
            label=f"{self} Termination",
            once=True,
            priority=self.priority,
        )

        # create a queue for engaged events
        self._events: List[Event] = list()

        # create a queue for acquired resources
        self._occupied_resources: List[Resource] = list()

    def __str__(self) -> str:
        """Return the label of the entity if it exists, otherwise return the id of the entity."""
        if self.label is None:
            return f"Entity {id(self)}"
        return self.label

    def _create(self):
        """Called when the entity is created."""
        self._created = True
        logger.debug(f"Entity {self} created.")

    def _terminate(self):
        """Called when the entity is terminated."""
        self._terminated = True
        for event in self.events:
            event.end()
        for resource in self.occupied_resources:
            resource.collect(self, inf)
        logger.debug(f"Entity {self} terminated.")

    def event(
        self,
        at: float | Event,
        till: float | Event,
        step: float = Mundus.time_step,
        label: Optional[str] = None,
        once: bool = False,
        priority: int = 0,
    ):
        """Decorator to add an event to the entity."""

        def _event(action: Callable):

            async def __event():

                if self.terminated:
                    logger.warn(f"Entity {self} already terminated.")
                    return

                while True:
                    if not self.created:
                        logger.warn(f"Entity {self} not created yet.")
                        await asyncio.sleep(0)
                    else:
                        break

                event = Event(
                    at=at,
                    till=till,
                    step=step,
                    action=action,
                    label=label,
                    once=once,
                    priority=priority,
                )
                self.events.append(event)
                logger.debug(f"Event {event} added to entity {self}.")

            Event(
                at=at, till=at, action=__event, label=f"{label} Engagement", once=True
            )

        return _event

    def acquire(self, resource: Resource, amount: float) -> bool:
        """Acquire a amount from the resource."""
        if resource.distribute(self, amount):
            return True
        else:
            return False

    def release(self, resource: Resource, amount: float) -> bool:
        """Release a amount from the resource."""
        if resource.collect(self, amount):
            return True
        else:
            return False

    @property
    def label(self):
        """Short description of the entity."""
        return self._label

    @property
    def created(self):
        """Return True if the entity is created, otherwise False."""
        return self._created

    @property
    def terminated(self):
        """Return True if the entity is terminated, otherwise False."""
        return self._terminated

    @property
    def events(self):
        """The events that the entity is engaged with."""
        return self._events

    @property
    def occupied_resources(self):
        """The resources that the entity is using."""
        return self._occupied_resources

    @property
    def priority(self):
        """The priority of the entity."""
        return self._priority

    @property
    def creation(self):
        """The creation event of the entity."""
        return self._creation

    @property
    def termination(self):
        """The termination event of the entity."""
        return self._termination

created property

Return True if the entity is created, otherwise False.

creation property

The creation event of the entity.

events property

The events that the entity is engaged with.

label property

Short description of the entity.

occupied_resources property

The resources that the entity is using.

priority property

The priority of the entity.

terminated property

Return True if the entity is terminated, otherwise False.

termination property

The termination event of the entity.

__init__(at, till, label=None, priority=0)

Create an entity with a creation and termination event.

Parameters:

Name Type Description Default
at float | Event

when the entity is created.

 required
till float | Event

when the entity is terminated.

 required
label Optional[str]

short description to the entity. Defaults to None.

 None
priority int

the priority of the entity, only impacts the creation and termination event. Defaults to 0.

 0
Source code in Akatosh\entity.py 
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
def __init__(
    self,
    at: float | Event,
    till: float | Event,
    label: Optional[str] = None,
    priority: int = 0,
) -> None:
    """Create an entity with a creation and termination event.

    Args:
        at (float | Event): when the entity is created.
        till (float | Event): when the entity is terminated.
        label (Optional[str], optional): short description to the entity. Defaults to None.
        priority (int, optional): the priority of the entity, only impacts the creation and termination event. Defaults to 0.
    """
    self._label = label
    self._at = at
    self._till = till
    self._created = False
    self._terminated = False
    self._priority = priority

    # create an instant creation event
    self._creation = Event(
        at=at,
        till=till,
        action=self._create,
        label=f"{self} Creation",
        once=True,
        priority=self.priority,
    )
    self._termination = Event(
        at=till,
        till=till,
        action=self._terminate,
        label=f"{self} Termination",
        once=True,
        priority=self.priority,
    )

    # create a queue for engaged events
    self._events: List[Event] = list()

    # create a queue for acquired resources
    self._occupied_resources: List[Resource] = list()

__str__()

Return the label of the entity if it exists, otherwise return the id of the entity.

Source code in Akatosh\entity.py 
63
64
65
66
67
def __str__(self) -> str:
    """Return the label of the entity if it exists, otherwise return the id of the entity."""
    if self.label is None:
        return f"Entity {id(self)}"
    return self.label

acquire(resource, amount)

Acquire a amount from the resource.

Source code in Akatosh\entity.py 
127
128
129
130
131
132
def acquire(self, resource: Resource, amount: float) -> bool:
    """Acquire a amount from the resource."""
    if resource.distribute(self, amount):
        return True
    else:
        return False

event(at, till, step=Mundus.time_step, label=None, once=False, priority=0)

Decorator to add an event to the entity.

Source code in Akatosh\entity.py 
 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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def event(
    self,
    at: float | Event,
    till: float | Event,
    step: float = Mundus.time_step,
    label: Optional[str] = None,
    once: bool = False,
    priority: int = 0,
):
    """Decorator to add an event to the entity."""

    def _event(action: Callable):

        async def __event():

            if self.terminated:
                logger.warn(f"Entity {self} already terminated.")
                return

            while True:
                if not self.created:
                    logger.warn(f"Entity {self} not created yet.")
                    await asyncio.sleep(0)
                else:
                    break

            event = Event(
                at=at,
                till=till,
                step=step,
                action=action,
                label=label,
                once=once,
                priority=priority,
            )
            self.events.append(event)
            logger.debug(f"Event {event} added to entity {self}.")

        Event(
            at=at, till=at, action=__event, label=f"{label} Engagement", once=True
        )

    return _event

release(resource, amount)

Release a amount from the resource.

Source code in Akatosh\entity.py 
134
135
136
137
138
139
def release(self, resource: Resource, amount: float) -> bool:
    """Release a amount from the resource."""
    if resource.collect(self, amount):
        return True
    else:
        return False