matsim_plan_gen/plan_gen.py

#!/usr/bin/env python3
''' Python generator for MATSim plans. '''

import sys
import time
import numpy as np
import lxml.etree as etree

# constants
MIN_DEPARTURE_TIME = '08:00:00'
MAX_DEPARTURE_TIME = '09:00:00'
WORK_DURATION = '04:00:00'

def read_nodes(input_network):
    ''' returns all network nodes as a list '''
    if not input_network:
        return None
    tree = etree.parse(input_network)
    return [node for node in tree.xpath("/network/nodes/node")]

def get_seconds(time_str):
    ''' returns seconds from a time string '''
    h, m, s = time_str.split(':')
    return int(h) * 3600 + int(m) * 60 + int(s)

def rand_time(low, high):
    ''' returns a random time between low and high bounds '''
    low_s = get_seconds(low)
    high_s = get_seconds(high)
    delta = np.random.randint(high_s - low_s)
    return time.strftime('%H:%M:%S', time.gmtime(low_s + delta))

def rand_node_xy(nodes, clusters):
    ''' returns a random node coordinates from a list of nodes '''
    used_nodes = nodes
    if any(clusters):
        cluster = np.random.randint(len(clusters))
        used_nodes = clusters[cluster]
    node = used_nodes[np.random.randint(len(used_nodes))]
    return (node.get('x'), node.get('y'))

def rand_person(nodes, home_clusters, work_clusters):
    ''' returns a person as a dictionnary of random parameters '''
    home_xy = rand_node_xy(nodes, home_clusters)
    work_xy = rand_node_xy(nodes, work_clusters)
    home_departure = rand_time(MIN_DEPARTURE_TIME, MAX_DEPARTURE_TIME)
    return {'home': home_xy, 'work': work_xy, 'home_departure': home_departure}

def create_child(parent_node, child_name, child_attrs=None):
    ''' creates an xml child element and set its attributes '''
    child = etree.SubElement(parent_node, child_name)
    if child_attrs is None:
        return child
    for attr, value in child_attrs.items():
        child.set(attr, value)
    return child

def make_plans(persons):
    ''' makes xml tree of plans based on persons list '''
    plans = etree.Element('plans')
    for n, p in enumerate(persons):
        person = create_child(plans, 'person', {'id': str(n+1)})
        plan = create_child(person, 'plan')
        # plan
        create_child(plan, 'act', {'type': 'h', 'x': p['home'][0], 'y': p['home'][1], 'end_time': p['home_departure']})
        create_child(plan, 'leg', {'mode': 'car'})
        create_child(plan, 'act', {'type': 'w', 'x': p['work'][0], 'y': p['work'][1], 'dur': WORK_DURATION})
        create_child(plan, 'leg', {'mode': 'car'})
        create_child(plan, 'act', {'type': 'h', 'x': p['home'][0], 'y': p['home'][1]})
    return plans

if __name__ == '__main__':

    # command line arguments
    if len(sys.argv) != 5:
        print('usage:', sys.argv[0], '<nb_persons> <input_network> <input_home_clusters> <input_work_clusters>')
        print('type "" if you don\'t want to use specific clusters')
        sys.exit(-1)
    NB_PERSONS = int(sys.argv[1])
    INPUT_NETWORK = sys.argv[2]
    INPUT_HOME_CLUSTERS = sys.argv[3].split(',')
    INPUT_WORK_CLUSTERS = sys.argv[4].split(',')

    # get data
    NODES = read_nodes(INPUT_NETWORK)
    HOME_CLUSTERS = [read_nodes(cluster) for cluster in INPUT_HOME_CLUSTERS]
    WORK_CLUSTERS = [read_nodes(cluster) for cluster in INPUT_WORK_CLUSTERS]

    # make xml
    PERSONS = [rand_person(NODES, HOME_CLUSTERS, WORK_CLUSTERS) for _ in range(NB_PERSONS)]
    PLANS = make_plans(PERSONS)

    # print XML
    print('<?xml version="1.0" ?>')
    print('<!DOCTYPE plans SYSTEM "http://www.matsim.org/files/dtd/plans_v4.dtd">')
    print(etree.tostring(PLANS, pretty_print=True).decode('utf-8'))