data.py

"""
Get OSM Data and merge with charging stops.
"""
import argparse
from typing import Tuple

import pandas as pd

long = float
lat = float
coordinate = Tuple[lat, long]


def parse_charging_stations_csv(file: str, nw_corner: coordinate = None, se_corner: coordinate = None):
    def float_converter(s: str):
        if not s:
            return 0.0
        return float(s.replace(',', '.'))

    converters = {
        'lon': lambda s: float(s.replace(',', '.')),
        'lat': lambda s: float(s.replace(',', '.')),
        'p1': float_converter,
        'p2': float_converter,
        'p3': float_converter,
        'p4': float_converter
    }

    df: pd.DataFrame = pd.read_csv(
        file,
        sep='\t',
        skiprows=6,
        header=None,
        names=converters.keys(),
        converters=converters,
        usecols=[4, 5, 11, 14, 17, 20]
    )

    # Store only information about maximum power
    df['power'] = df[['p1', 'p2', 'p3', 'p4']].max(axis=1)
    del df['p1']
    del df['p2']
    del df['p3']
    del df['p4']

    if nw_corner is not None and se_corner is not None:
        lat_max, lon_min = nw_corner
        lat_min, lon_max = se_corner

        df = df[(df['lon'] <= lon_max)
                & (df['lon'] >= lon_min)
                & (df['lat'] <= lat_max)
                & (df['lat'] >= lat_min)
                ]
    return df


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description="Merge OSM map data with"
                                                 "charging station data.")
    parser.add_argument(
        'input_path',
        metavar='source',
        type=argparse.FileType('r'),
        help='Path to the csv file.'
    )

    parser.add_argument(
        'dest_path',
        metavar='dest',
        help='Path for destination of Charging Stations json.'
    )

    parser.add_argument(
        '--cut',
        type=float,
        nargs=4,
        metavar=('lat_nw', 'lon_nw', 'lat_se', 'lon_se'),
        help='Use only those charging stations that are located with in a '
             'rectangle defined by the north-west and south-east coordinates.'
    )

    args = parser.parse_args()

    nw_corner = None
    se_corner = None
    if args.cut:
        nw_corner = args.cut[:2]
        se_corner = args.cut[2:]

    parse_charging_stations_csv(
        args.input_path,
        nw_corner=nw_corner,
        se_corner=se_corner
    ).to_json(args.dest_path, orient='records')