voc

#!/usr/bin/env python3
# -*- mode: python; coding: utf-8 -*-
"""
Retrieve information from VOC

Usage:
  voc (-h | --help)
  voc --version
  voc [-v|-vv] [options] list
  voc [-v|-vv] [options] status
  voc [-v|-vv] [options] trips [(--pretty|--json|--csv)]
  voc [-v|-vv] [options] owntracks
  voc [-v|-vv] [options] print [<attribute>]
  voc [-v|-vv] [options] (lock | unlock)
  voc [-v|-vv] [options] heater (start | stop)
  voc [-v|-vv] [options] engine (start | stop)
  voc [-v|-vv] [options] honk_and_blink
  voc [-v|-vv] [options] call <method>
  voc [-v|-vv] [options] mqtt
  voc [-v|-vv] [options] dashboard

Options:
  -u <username>         VOC username
  -p <password>         VOC password
  -r <region>           VOC region (na, cn, etc.)
  -s <url>              VOC service URL
  -i <vin>              Vehicle VIN or registration number
  -g                    Geolocate position
  --owntracks_key=<key> Owntracks encryption password
  -I <interval>         Polling interval (seconds) [default: 300]
  -h --help             Show this message
  --immutable           Read only mode
  -v,-vv                Increase verbosity
  -d                    More debugging
  --scandinavian_miles  Report using Scandinavian miles instead of km ISO unit
  --utc                 Print timestamps in UTC (+00:00) instead of local time
  --version             Show version
"""

import docopt
import logging
import asyncio
from time import time
from json import dumps as to_json
from sys import stderr
from collections import OrderedDict
from datetime import timezone
from volvooncall import (__version__, read_credentials, Connection, json_serialize, owntracks_encrypt)

_LOGGER = logging.getLogger(__name__)

LOGFMT = "%(asctime)s %(levelname)5s (%(threadName)s) [%(name)s] %(message)s"
DATEFMT = "%y-%m-%d %H:%M.%S"


def lookup_position(lat, lon):
    try:
        import ssl
        import certifi
        import geopy.geocoders
        from geopy.geocoders import Nominatim
        geolocator = Nominatim(user_agent='volvooncall/%s' % __version__,
                               timeout=10,
                               ssl_context=ssl.create_default_context(cafile=certifi.where()))
        return geolocator.reverse((lat, lon))
    except ImportError:
        _LOGGER.warning('geopy or certifi not installed. position lookup not available')


def print_vehicle(vehicle, geolocate=False):


    def yes_or_no(boolean):
        if boolean is None:
            return 'unknown'
        return ('no', 'yes')[boolean]

    def on_or_off(boolean):
        if boolean is None:
            return 'unknown'
        return ('off', 'on')[boolean]

    s = '%s %dkm' % (
        vehicle,
        vehicle.odometer / 1000)

    if vehicle.fuel_amount_level:
        s += ' (fuel %s%% %skm)' % (
            vehicle.fuel_amount_level,
            vehicle.distance_to_empty)

    print(s)
    try:
        lat, lon = (vehicle.position['latitude'],
                    vehicle.position['longitude'])
        pos = lookup_position(lat, lon) if geolocate else None
        if pos:
            print('    position: %.14f,%.14f (%s)' % (lat, lon, pos.address))
        else:
            print('    position: %.14f,%.14f' % (lat, lon))
    except AttributeError:
        pass

    print('    engine: %s' % on_or_off(vehicle.is_engine_running))
    print('    locked: %s' % yes_or_no(vehicle.is_locked))
    print('    heater: %s' % on_or_off(vehicle.is_heater_on))


async def main(args):
    """Command line interface."""

    credentials = read_credentials()

    if args['-u'] and args['-p']:
        credentials.update(username=args['-u'],
                           password=args['-p'])
    if args['-r']:
        credentials.update(region = args['-r'])
    if args['-s']:
        credentials.update(service_url = args['-s'])
    if args['-I']:
        credentials.update(interval = args['-I'])
    if args['--owntracks_key']:
        credentials.update(
            owntracks_key = args['--owntracks_key'])
    if args['--scandinavian_miles'] or '--scandinavian_miles' not in credentials:
        credentials.update(
            scandinavian_miles = args['--scandinavian_miles'])
    if args['--immutable'] or '--immutable' not in credentials:
        credentials.update(
            immutable = args['--immutable'])

    async with Connection(**credentials) as connection:

        if args['mqtt']:
            import mqtt
            return await mqtt.run(connection, credentials)

        journal = args['trips'] or args['dashboard'] or args['mqtt']
        res = await connection.update(journal=journal)

        if not res:
            exit('Could not connect to the server.')

        if args['list']:
            for vehicle in connection.vehicles:
                print(vehicle)
                exit()

        if args['-i']:
            vehicle = connection.vehicle(args['-i'])
        else:
            vehicle = next(connection.vehicles, None)

        if not vehicle:
            exit('Vehicle not found')

        if args['status']:
            print_vehicle(vehicle, args['-g'])
        elif args['trips']:
            if args['--json']:
                print(to_json(vehicle.trips, indent=4, default=json_serialize))
            else:

                def fix_timezone(dt):
                    return dt.astimezone(timezone.utc if args['--utc'] else None)

                for trip in vehicle.trips:
                    trip = trip['tripDetails'][0]
                    print('%.29s %25s %-10.10s %25s %10.10s %-10.10s' % (
                        fix_timezone(trip['startTime']),
                        trip['startPosition']['streetAddress'],
                        trip['startPosition']['city'],
                        fix_timezone(trip['endTime']),
                        trip['endPosition']['streetAddress'],
                        trip['endPosition']['city']))

        elif args['print']:
            attr = args['<attribute>']
            if attr:
                if not vehicle.has_attr(attr):
                    exit('unknown attribute')
                print(vehicle.get_attr(attr))
            else:
                print(vehicle.json)
        elif args['owntracks']:
            msg = to_json(
                dict(_type='location',
                     tid='volvo',
                     t='p',
                     lat=vehicle.position['latitude'],
                     lon=vehicle.position['longitude'],
                     acc=1,
                     tst=int(time())),
                indent=4, default=json_serialize)
            key = credentials.get('owntracks_key')
            if key:
                msg = to_json(
                    dict(_type='encrypted',
                         data=owntracks_encrypt(msg, key)))
            print(msg)
        elif args['heater']:
            if args['start']:
                vehicle.start_heater()
            else:
                vehicle.stop_heater()
        elif args['engine']:
            if args['start']:
                vehicle.start_engine()
            else:
                vehicle.stop_engine()
        elif args['lock']:
            vehicle.lock()
        elif args['unlock']:
            vehicle.unlock()
        elif args['honk_and_blink']:
            vehicle.honk_and_blink()
        elif args['dashboard']:
            from dashboard import Dashboard
            dashboard = Dashboard(vehicle, **credentials)
            for instrument in dashboard.instruments:
                print(f'{instrument}: {instrument.str_state}')
        elif args['call']:
            vehicle.call(args['<method>'])


if __name__ == '__main__':
    args = docopt.docopt(__doc__,
                         version=__version__)

    debug = args['-d']

    if debug:
        log_level = logging.DEBUG
    else:
        log_level = [logging.ERROR, logging.INFO, logging.DEBUG][args['-v']]

    try:
        import coloredlogs
        coloredlogs.install(level=log_level,
                            stream=stderr,
                            datefmt=DATEFMT,
                            fmt=LOGFMT)
    except ImportError:
        _LOGGER.debug("no colored logs. pip install coloredlogs?")
        logging.basicConfig(level=log_level,
                            stream=stderr,
                            datefmt=DATEFMT,
                            format=LOGFMT)

    logging.captureWarnings(debug)

    if debug:
        _LOGGER.info('Debug is on')

    try:
        asyncio.run(main(args), debug=debug)
    except KeyboardInterrupt:
        exit()