Skip to content

Commit

Permalink
Added process_attributes function (#49)
Browse files Browse the repository at this point in the history 
Add process_switches function, convert exist list to PossibleExits class, and add error checking to Attributes class
  • Loading branch information
@GeorgeEfstathiadis
GeorgeEfstathiadis authored Nov 19, 2021
1 parent e03780d commit 9a57d71
Show file tree
Hide file tree
Showing 2 changed files with 224 additions and 91 deletions.
174 changes: 128 additions & 46 deletions forest/bonsai/simulate_gps_data.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
from enum import Enum
import os
import time
from typing import Dict, List, Tuple, Union
from typing import Dict, List, Optional, Tuple, Union

import numpy as np
import openrouteservice
Expand All @@ -18,6 +18,19 @@
from forest.poplar.legacy.common_funcs import datetime2stamp, stamp2datetime

R = 6.371*10**6
ACTIVE_STATUS_LIST = range(11)
TRAVELLING_STATUS_LIST = range(11)


class PossibleExits(Enum):
"""This class enumerates possible exits for attributes"""
CAFE = "cafe"
BAR = "bar"
RESTAURANT = "restaurant"
PARK = "park"
CINEMA = "cinema"
DANCE = "dance"
FITNESS = "fitness"


class Vehicle(Enum):
Expand All @@ -30,7 +43,7 @@ class Vehicle(Enum):

class Occupation(Enum):
"""This class enumerates occupation for attributes"""
NONE = ""
NONE = "none"
WORK = "office"
SCHOOL = "university"

Expand Down Expand Up @@ -170,27 +183,79 @@ def bounding_box(center: Tuple[float, float], radius: int) -> Tuple:
return lat - lat_const, lon - lon_const, lat + lat_const, lon + lon_const


@dataclass
class Attributes:
"""This class holds the attributes needed to create an instance of a person
"""This class holds the attributes needed to create an instance of a
Person class"""

Args:
vehicle used for distances and time of flights
main_occupation used for routine action in weekdays
active_status = 0-10
used for probability in free time to take an action
or stay home
travelling status = 0-10
used to derive amount of distance travelled
preferred_places = [x1, x2, x3]
used to sample action when free time
where x1-x3 are amenities (str)
"""
vehicle: Vehicle
main_occupation: Occupation
active_status: int
travelling_status: int
preferred_places: list
def __init__(self,
vehicle: Optional[str] = None,
main_employment: Optional[str] = None,
active_status: Optional[int] = None,
travelling_status: Optional[int] = None,
preferred_places: Optional[List[str]] = None,
**kwargs):
"""Error check and generate missing data for attributes
Args:
vehicle: used for distances and time of flights
main_occupation: used for routine action in weekdays
active_status: used for probability in free time to take an action
or stay home
travelling status: used to derive amount of distance travelled
preferred_places :used to sample action when free time
where x1-x3 are amenities (str)
Raises:
ValueError: for incorrect vehicle type
ValueError: for incorrect main_employment type
ValueError: if active_status is not between 0 and 10
ValueError: if travelling_status is not between 0 and 10
ValueError: if an exit from possible_exits is not correct type
"""
if vehicle is not None:
self.vehicle = Vehicle(vehicle)
else:
# exclude bus
self.vehicle = np.random.choice(list(Vehicle)[1:])

if main_employment is not None:
self.main_occupation = Occupation(main_employment)
else:
self.main_occupation = np.random.choice(list(Occupation))

if active_status is not None:
if active_status not in ACTIVE_STATUS_LIST:
raise ValueError("active_status must be between 0 and 10")
self.active_status = int(active_status)
else:
self.active_status = np.random.choice(ACTIVE_STATUS_LIST)

if travelling_status is not None:
if travelling_status not in TRAVELLING_STATUS_LIST:
raise ValueError("travelling_status must be between 0 and 10")
self.travelling_status = int(travelling_status)
else:
self.travelling_status = np.random.choice(TRAVELLING_STATUS_LIST)

if preferred_places is not None:
self.preferred_places = []
for possible_exit in preferred_places:
possible_exit2 = PossibleExits(possible_exit)
self.preferred_places.append(possible_exit2)

possible_exits2 = [
x for x in PossibleExits
if x not in self.preferred_places
]

random_exits = np.random.choice(
possible_exits2, 3 - len(self.preferred_places), replace=False
).tolist()
for choice in random_exits:
self.preferred_places.append(choice)
else:
self.preferred_places = np.random.choice(
list(PossibleExits), 3, replace=False
).tolist()


@dataclass
Expand Down Expand Up @@ -271,15 +336,7 @@ def __init__(self,
self.office_days = np.array([])

# define favorite places
self.possible_destinations = [
"cafe",
"bar",
"restaurant",
"park",
"cinema",
"dance",
"fitness",
]
self.possible_destinations = list(PossibleExits)

# for a certain venue select 3 locations for each venue randomly
# these will be considered the 3 favorite places to go
Expand All @@ -290,46 +347,48 @@ def __init__(self,
for possible_exit in self.possible_destinations:
# if there are more than 3 sets of coordinates for an venue
# select 3 at random, else select all of them as preferred
if len(local_places[possible_exit]) > 3:
if len(local_places[possible_exit.value]) > 3:
random_places = np.random.choice(
range(len(local_places[possible_exit])), 3, replace=False
range(len(local_places[possible_exit.value])),
3, replace=False,
).tolist()
places_selected = [
tuple(place)
for place in np.array(local_places[possible_exit])[
for place in np.array(local_places[possible_exit.value])[
random_places
]
if tuple(place) != home_coordinates
]
setattr(self, possible_exit + "_places", places_selected)
setattr(self, possible_exit.value + "_places", places_selected)
else:
setattr(
self,
possible_exit + "_places",
possible_exit.value + "_places",
[
tuple(place) for place in local_places[possible_exit]
tuple(place) for place
in local_places[possible_exit.value]
if tuple(place) != home_coordinates
],
)
# calculate distances of selected places from home
# create a list of the locations ordered by distance
distances = [
great_circle_dist(*home_coordinates, *place)
for place in getattr(self, possible_exit + "_places")
for place in getattr(self, possible_exit.value + "_places")
]
order = np.argsort(distances)
setattr(
self,
possible_exit + "_places_ordered",
np.array(getattr(self, possible_exit + "_places"))[
possible_exit.value + "_places_ordered",
np.array(getattr(self, possible_exit.value + "_places"))[
order
].tolist(),
)

# remove all exits which have no places nearby
possible_destinations2 = self.possible_destinations.copy()
for act in possible_destinations2:
if len(getattr(self, act + "_places")) == 0:
if len(getattr(self, act.value + "_places")) == 0:
self.possible_destinations.remove(act)

# order preferred places by travelling_status
Expand All @@ -340,15 +399,15 @@ def __init__(self,
+ (10 - self.attributes.travelling_status) ** 2
)
for act in self.possible_destinations:
act_places = getattr(self, act + "_places_ordered").copy()
act_places = getattr(self, act.value + "_places_ordered").copy()

places = []
for i in range(len(act_places) - 1, -1, -1):
index = np.random.binomial(i, travelling_status_norm)
places.append(act_places[index])
del act_places[index]

setattr(self, act + "_places", places)
setattr(self, act.value + "_places", places)

def set_travelling_status(self, travelling_status: int):
"""Update preferred locations of exits
Expand All @@ -364,15 +423,15 @@ def set_travelling_status(self, travelling_status: int):
travelling_status ** 2 + (10 - travelling_status) ** 2
)
for act in self.possible_destinations:
act_places = getattr(self, act + "_places_ordered").copy()
act_places = getattr(self, act.value + "_places_ordered").copy()

places = []
for i in range(len(act_places) - 1, -1, -1):
index = np.random.binomial(i, travelling_status_norm)
places.append(act_places[index])
del act_places[index]

setattr(self, act + "_places", places)
setattr(self, act.value + "_places", places)

def set_active_status(self, active_status: int):
"""Update active status.
Expand All @@ -393,7 +452,7 @@ def set_active_status(self, active_status: int):
)
self.office_days.sort()

def update_preferred_places(self, exit_code: str):
def update_preferred_places(self, exit_code: PossibleExits):
"""This function updates the set of preferred exits for the day,
after an action has been performed.
Expand Down Expand Up @@ -496,7 +555,7 @@ def choose_preferred_exit(self, current_time: float,

# after venue has been selected, a location for that venue
# needs to be selected as well.
action_locations = getattr(self, selected_action + "_places")
action_locations = getattr(self, selected_action.value + "_places")
ratios2 = ratios[: len(action_locations)]
probabilities2 = np.array(ratios2)
probabilities2 = probabilities2 / sum(probabilities2)
Expand Down Expand Up @@ -1035,6 +1094,29 @@ def prepare_data(
)


def process_switches(
attributes: Dict[str, Dict], key: str,
) -> Dict[str, int]:
"""Preprocesses the attributes of each person.
Args:
attributes: (dictionary) contains attributes of each person,
loaded from json file.
key: (str) a key from attributes.keys()
Returns:
switches: (dictionary) contains possible changes of attributes
in between of simulation
"""
switches = {}

for x in attributes[key].keys():
key_list = x.split("-")
if len(key_list) == 2:
switches[x] = attributes[key][x]

return switches


def sim_GPS_data(cycle,p,data_folder):
## only two parameters
## cycle is the sum of on-cycle and off_cycle, unit is minute
Expand Down
Loading

0 comments on commit 9a57d71

Please sign in to comment.