Custom degradation and decision-making classes are served in a more pythonic way.

Author: aslan-ng

examples/custom-decision-making/decision_making.py

@@ -7,28 +7,30 @@ class CustomDecisionMaking(DecisionMaking):
     Methods related to agents' decision-making
     """
 
+    is_custom = True
+
     def decide_destination(self, agent_id: int, duration: float) -> None:
         """
         Decide the destination
         """
         destination_id = None
         critical_stay_length = duration
-        action_queue = self.get_action_queue(id=agent_id)
+        action_queue = self.society.get_action_queue(id=agent_id)
         # Find suitable market
         destinations = self.preasssumed_destinations(agent_id=agent_id)
         destinations = sorted(destinations, key=lambda x: x["score"], reverse=True)
         suitable_destination_found = False
         for destination in destinations:
-            path = self.infrastructure.path(
-                id_start=self.get_current_node(id=agent_id), id_end=destination["id"]
+            path = self.society.infrastructure.path(
+                id_start=self.society.get_current_node(id=agent_id), id_end=destination["id"]
             )
             move_go = Move(
                 action_queue=action_queue,
                 path=path,
-                usage=self.transportation_degradation,
+                usage=self.society.transportation_degradation,
             )
             # Stay (at the destination)
-            stay_length = self.max_time_outside - (2 * move_go.total_duration)
+            stay_length = self.society.max_time_outside - (2 * move_go.total_duration)
             if stay_length > critical_stay_length:
                 suitable_destination_found = True
                 destination_id = destination["id"]
@@ -39,17 +41,17 @@ def decide_destination(self, agent_id: int, duration: float) -> None:
             destinations = self.search_destinations(agent_id=agent_id)
             destinations = sorted(destinations, key=lambda x: x["score"], reverse=True)
             for destination in destinations:
-                path = self.infrastructure.path(
-                    id_start=self.get_current_node(id=agent_id),
+                path = self.society.infrastructure.path(
+                    id_start=self.society.get_current_node(id=agent_id),
                     id_end=destination["id"],
                 )
                 move_go = Move(
                     action_queue=action_queue,
                     path=path,
-                    usage=self.transportation_degradation,
+                    usage=self.society.transportation_degradation,
                 )
                 # Stay (at the destination)
-                stay_length = self.max_time_outside - (2 * move_go.total_duration)
+                stay_length = self.society.max_time_outside - (2 * move_go.total_duration)
                 if stay_length > critical_stay_length:
                     suitable_destination_found = True
                     destination_id = destination["id"]

examples/custom-decision-making/main.py

@@ -1,5 +1,6 @@
 import os
 import piperabm as pa
+from decision_making import CustomDecisionMaking
 
 
 path = os.path.dirname(os.path.realpath(__file__))
@@ -23,6 +24,7 @@
 model.infrastructure.bake()
 
 # Set up the society
+model.society.set_decision_making(CustomDecisionMaking)  # Set the custom decision-making class
 model.society.average_income = 1
 agent_id = 1
 home_id = model.infrastructure.homes[0]

examples/custom-decision-making/result/animation.mp4

@@ -1,3 +1,25 @@
 # Custom Degradation
 
-In this example, we see how the user can customize degradation behavior of the infrastrcture elements of the model by providing their own `degradation.py` file.
+In this example, we see how the user can customize degradation behavior of the infrastrcture elements of the model by providing a custom degradation class. Then, the impact of this change is compared to default model behavior.
+
+This is how the custom degradation file is being served:
+
+```python
+from piperabm.infrastructure.degradation import Degradation
+
+class CustomDegradation(Degradation):
+
+    def calculate_adjustment_factor(
+        self, usage_impact: float, age_impact: float
+    ) -> float:
+        """
+        Calculate adjustment factor using a custom formula.
+        """
+        return (
+            1 + (self.infrastructure.coeff_usage * usage_impact) + (self.infrastructure.coeff_age * (age_impact ** 2))
+        )
+        
+model.infrastructure.set_degradation(CustomDegradation)
+```
+
+When edges degrade, they will become less efficient to be used. It is equivalent of having longer length, since it will take longer (and require more fuel) for the agents to pass through. This is called `adjusted_length` and using custom degradation, the formula to calculate it is customized.

examples/custom-decision-making/result/final.json

@@ -1,45 +1,69 @@
-import os
 import piperabm as pa
+from piperabm.infrastructure.degradation import Degradation
 
 
-path = os.path.dirname(os.path.realpath(__file__))
-model = pa.Model(path=path, seed=2)
-
-# Set up the infrastructure
-model.infrastructure.coeff_usage = 1
-model.infrastructure.coeff_age = 1
-model.infrastructure.add_street(pos_1=[0, 0], pos_2=[-60, 40], name="road")
-model.infrastructure.add_home(pos=[5, 0], id=1, name="home")
-model.infrastructure.add_market(
-    pos=[-60, 45],
-    id=2,
-    name="market",
-    resources={
-        "food": 100,
-        "water": 100,
-        "energy": 100,
-    },
-)
-model.infrastructure.bake()
-
-# Set up the society
-model.society.average_income = 1
-agent_id = 1
-home_id = model.infrastructure.homes[0]
-model.society.add_agent(
-    id=agent_id,
-    home_id=home_id,
-    socioeconomic_status=1,
-    resources={
-        "food": 1,
-        "water": 1,
-        "energy": 1,
-    },
-    balance=100,
-)
-
-# Run the simulation
-model.run(n=80, step_size=1, save=True)
-
-# Animate the result
-model.animate()
+class CustomDegradation(Degradation):
+
+    def calculate_adjustment_factor(
+        self, usage_impact: float, age_impact: float
+    ) -> float:
+        """
+        Calculate adjustment factor using a custom formula.
+        """
+        return (
+            1 + (self.infrastructure.coeff_usage * usage_impact) + (self.infrastructure.coeff_age * (age_impact ** 2))
+        )
+
+
+def setup_model():
+    model = pa.Model(seed=2)
+
+    # Set up the infrastructure
+    model.infrastructure.coeff_usage = 1
+    model.infrastructure.coeff_age = 1
+    model.infrastructure.add_street(pos_1=[0, 0], pos_2=[-60, 40], name="road")
+    model.infrastructure.add_home(pos=[5, 0], id=1, name="home")
+    model.infrastructure.add_market(
+        pos=[-60, 45],
+        id=2,
+        name="market",
+        resources={
+            "food": 100,
+            "water": 100,
+            "energy": 100,
+        },
+    )
+    model.infrastructure.bake()
+
+    # Set up the society
+    model.society.average_income = 1
+    agent_id = 1
+    home_id = model.infrastructure.homes[0]
+    model.society.add_agent(
+        id=agent_id,
+        home_id=home_id,
+        socioeconomic_status=1,
+        resources={
+            "food": 1,
+            "water": 1,
+            "energy": 1,
+        },
+        balance=100,
+    )
+
+    return model
+
+edge = (8042686386972756495, 478254495130285640)
+
+# Run the simulation with default degradation
+model_default = setup_model()
+model_default.run(n=80, step_size=1)
+
+# Run the simulation with custom degradation
+model_custom = setup_model()
+model_custom.infrastructure.set_degradation(CustomDegradation)  # Set the custom degradation class
+model_custom.run(n=80, step_size=1)
+
+# Compare the results
+print("Default adjusted length formula:", model_default.infrastructure.get_adjusted_length(ids=edge))
+print("Custom adjusted length formula:",model_custom.infrastructure.get_adjusted_length(ids=edge))