131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
"""This module contains the classes describing the population of a simulation.
|
|
"""
|
|
# Copyright Lucas Javaudin - All Rights Reserved
|
|
# Unauthorized copying of this file, via any medium is strictly prohibited
|
|
# Proprietary and confidential
|
|
# Written by Lucas Javaudin <me@lucasjavaudin.com>, 2021
|
|
|
|
import os
|
|
import json
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from metrosim.exceptions import (
|
|
MetroIOError, MetroInputError, MetroUnsupportedError
|
|
)
|
|
from metrosim.models.schedule_utility import AlphaBetaGammaModel
|
|
from metrosim.models.mode_choice import (
|
|
DeterministicModeChoice, StochasticModeChoice
|
|
)
|
|
|
|
|
|
class Agent(object):
|
|
|
|
def __init__(self, agent_id, schedule_utility, mode_choice):
|
|
self.id = agent_id
|
|
self.schedule_utility = schedule_utility
|
|
self.mode_choice = mode_choice
|
|
|
|
def from_dict(data):
|
|
if 'id' not in data:
|
|
raise MetroInputError("Each Agent must have a valid id")
|
|
|
|
schedule_utility_data = data.get('schedule_utility', dict())
|
|
schedule_utility = super()._load_schedule_utility(
|
|
schedule_utility_data)
|
|
|
|
mode_choice_data = data.get('mode_choice', dict())
|
|
mode_choice = super()._load_mode_choice(mode_choice_data)
|
|
|
|
return Agent(data['id'], schedule_utility, mode_choice)
|
|
|
|
def _load_schedule_utility(data):
|
|
utility_type = data.get('type', None)
|
|
parameters = data.get('parameters', dict())
|
|
if utility_type == 'alpha-beta-gamma':
|
|
return AlphaBetaGammaModel.from_dict(parameters)
|
|
elif utility_type:
|
|
msg = "Unsupported schedule-utility type: {}"
|
|
raise MetroUnsupportedError(msg.format(utility_type))
|
|
else:
|
|
raise MetroInputError("Each Agent must have a schedule utility")
|
|
|
|
def _load_mode_choice(data):
|
|
choice_type = data.get('type', None)
|
|
modes = data.get('modes', list())
|
|
if choice_type == 'deterministic':
|
|
return DeterministicModeChoice.from_dict(modes)
|
|
elif choice_type == 'stochastic':
|
|
return StochasticModeChoice.from_dict(modes)
|
|
elif choice_type:
|
|
msg = "Unsupported mode-choice type: {}"
|
|
raise MetroUnsupportedError(msg.format(choice_type))
|
|
else:
|
|
raise MetroInputError("Each Agent must have a mode choice")
|
|
|
|
def get_pre_day_choice(self):
|
|
# Compute the expected utility for each continuous mode.
|
|
best_utility = -np.inf
|
|
best_mode = None
|
|
for mode in self.cont_modes:
|
|
results = mode.get_exp_utility(self)
|
|
self.mode_results[mode] = results
|
|
if results['exp_utility'] > best_utility:
|
|
best_utility = results['exp_utility']
|
|
best_mode = mode
|
|
|
|
for mode in self.det_modes:
|
|
results = mode.get_exp_utility(self, threshold=best_utility)
|
|
self.mode_results[mode] = results
|
|
|
|
|
|
class RawPopulation(object):
|
|
|
|
def __init__(self, segments, random_seed=None):
|
|
if not segments:
|
|
raise MetroInputError("A RawPopulation must have at least one "
|
|
"PopulationSegment")
|
|
self.segments = segments
|
|
self.random_seed = random_seed
|
|
|
|
def from_file(filename):
|
|
if not isinstance(filename, str):
|
|
raise TypeError("filename must be a str, got a {}"
|
|
.format(type(filename)))
|
|
if not os.path.isfile(filename):
|
|
raise MetroIOError("File not found: {}".format(filename))
|
|
|
|
try:
|
|
with open(filename, 'r') as f:
|
|
data = json.load(f)
|
|
except json.decoder.JSONDecodeError:
|
|
raise MetroIOError("Invalid JSON file: {}".format(filename))
|
|
|
|
segments = list()
|
|
for segment in data.get('segments', []):
|
|
segments.append(PopulationSegment(**segment))
|
|
|
|
random_seed = data.get('random_seed', None)
|
|
|
|
return RawPopulation(segments, random_seed)
|
|
|
|
|
|
class PopulationSegment(object):
|
|
|
|
def __init__(self, id, od_matrix):
|
|
self.id = id
|
|
self.od_matrix = ODMatrix.from_file(od_matrix)
|
|
|
|
|
|
class ODMatrix(object):
|
|
|
|
def from_file(filename):
|
|
if not isinstance(filename, str):
|
|
raise TypeError("filename must be a str, got a {}"
|
|
.format(type(filename)))
|
|
if not os.path.isfile(filename):
|
|
raise MetroIOError("File not found: {}".format(filename))
|
|
|
|
df = pd.read_csv(filename)
|