import express from 'express' ; import { Query } from 'express-serve-static-core' ; import { isUUID , isPct , isTimestamp , isFloat , isInsideBoundingBox } from '@mds-core/mds-utils' ; import stream from '@mds-core/mds-stream' ; import { UUID , Device , VehicleEvent , Telemetry , ErrorObject , isEnum , VEHICLE_EVENTS , VEHICLE_TYPES , VEHICLE_STATUSES , VEHICLE_REASONS , PROPULSION_TYPES , EVENT_STATUS_MAP , BoundingBox , VEHICLE_STATUS , VEHICLE_EVENT } from '@mds-core/mds-types' ; import db from '@mds-core/mds-db' ; import logger from '@mds-core/mds-logger' ; import cache from '@mds-core/mds-agency-cache' ; import { isArray } from 'util' ; import { VehiclePayload , TelemetryResult , CompositeVehicle , PaginatedVehiclesList } from './types' ; export function badDevice ( device ) { if ( ! device . device_id ) { return { error : 'missing_param' , error_description : 'missing device_id' } ; } if ( ! isUUID ( device . device_id ) ) { return { error : 'bad_param' , error_description : `device_id ${ device . device_id } is not a UUID` } ; } if ( ! Array . isArray ( device . propulsion ) ) { return { error : 'missing_param' , error_description : 'missing propulsion types' } ; } for ( const prop of device . propulsion ) { if ( ! isEnum ( PROPULSION_TYPES , prop ) ) { return { error : 'bad_param' , error_description : `invalid propulsion type ${ prop }` } ; } } if ( device . year !== null && device . year !== undefined ) { if ( ! Number . isInteger ( device . year ) ) { return { error : 'bad_param' , error_description : `invalid device year ${ device . year } is not an integer` } ; } if ( device . year < 1980 || device . year > 2020 ) { return { error : 'bad_param' , error_description : `invalid device year ${ device . year } is out of range` } ; } } if ( device . type === undefined ) { return { error : 'missing_param' , error_description : 'missing enum field "type"' } ; } if ( ! isEnum ( VEHICLE_TYPES , device . type ) ) { return { error : 'bad_param' , error_description : `invalid device type ${ device . type }` } ; } return null ; } export async function getVehicles ( skip , take , url , reqQuery , provider_id ? , bbox ? ) { function fmt ( query ) { const flat = { ... reqQuery , ... query } ; let s = `${ url }?` ; s += Object . keys ( flat ) . map ( key => `${ key }=${ flat [ key ] }` ) . join ( '&' ) ; return s ; } const rows = await db . readDeviceIds ( provider_id ) ; const total = rows . length ; logger . info ( `read ${ total } deviceIds in /vehicles` ) ; const events = await cache . readEvents ( rows . map ( record => record . device_id ) ) ; const eventMap = { } ; events . map ( event => { if ( event ) { eventMap [ event . device_id ] = event ; } } ) ; const deviceIdSuperset = bbox ? rows . filter ( record => { return eventMap [ record . device_id ] ? isInsideBoundingBox ( eventMap [ record . device_id ] . telemetry , bbox ) : true ; } ) : rows ; const deviceIdSubset = deviceIdSuperset . slice ( skip , skip + take ) . map ( record => record . device_id ) ; const devices = ( await db . readDeviceList ( deviceIdSubset ) ) . reduce ( ( acc , device ) => { if ( ! device ) { throw new Error ( 'device in DB but not in cache' ) ; } const event = eventMap [ device . device_id ] ; const status = event ? EVENT_STATUS_MAP [ event . event_type ] : VEHICLE_STATUSES . inactive ; const telemetry = event ? event . telemetry : null ; const updated = event ? event . timestamp : null ; return [ ... acc , { ... device , status , telemetry , updated } ] ; } , [ ] ) ; const noNext = skip + take >= deviceIdSuperset . length ; const noPrev = skip === 0 || skip > deviceIdSuperset . length ; const lastSkip = take * Math . floor ( deviceIdSuperset . length / take ) ; return { total , links : { first : fmt ( { skip : 0 , take } ) , last : fmt ( { skip : lastSkip , take } ) , prev : noPrev ? null : fmt ( { skip : skip - take , take } ) , next : noNext ? null : fmt ( { skip : skip + take , take } ) } , vehicles : devices } ; } const usBounds = { latMax : 49.45 , latMin : 24.74 , lonMax : - 66.94 , lonMin : - 124.79 } ; export function badTelemetry ( telemetry ) { if ( ! telemetry ) { return { error : 'missing_param' , error_description : 'invalid missing telemetry' } ; } const { device_id , timestamp , gps , charge } = telemetry ; if ( typeof gps !== 'object' ) { return { error : 'missing_param' , error_description : 'invalid missing gps' } ; } const { altitude , accuracy , speed , satellites } = gps ; const { lat , lng } = gps ; if ( ! isUUID ( device_id ) ) { return { error : 'missing_param' , error_description : 'no device_id included in telemetry' } ; } if ( typeof lat !== 'number' || Number . isNaN ( lat ) || lat < usBounds . latMin || lat > usBounds . latMax ) { return { error : 'bad_param' , error_description : `invalid lat ${ lat }` } ; } if ( typeof lng !== 'number' || Number . isNaN ( lng ) || lng < usBounds . lonMin || lng > usBounds . lonMax ) { return { error : 'bad_param' , error_description : `invalid lng ${ lng }` } ; } if ( altitude !== undefined && ! isFloat ( altitude ) ) { return { error : 'bad_param' , error_description : `invalid altitude ${ altitude }` } ; } if ( accuracy !== undefined && ! isFloat ( accuracy ) ) { return { error : 'bad_param' , error_description : `invalid accuracy ${ accuracy }` } ; } if ( speed !== undefined && ! isFloat ( speed ) ) { return { error : 'bad_param' , error_description : `invalid speed ${ speed }` } ; } if ( satellites !== undefined && satellites !== null && ! Number . isInteger ( satellites ) ) { return { error : 'bad_param' , error_description : `invalid satellites ${ satellites }` } ; } if ( charge !== undefined && ! isPct ( charge ) ) { return { error : 'bad_param' , error_description : `invalid charge ${ charge }` } ; } if ( ! isTimestamp ( timestamp ) ) { return { error : 'bad_param' , error_description : `invalid timestamp ${ timestamp } (note: should be in milliseconds)` } ; } return null ; } export async function badEvent ( event ) { if ( event . timestamp === undefined ) { return { error : 'missing_param' , error_description : 'missing enum field "timestamp"' } ; } if ( ! isTimestamp ( event . timestamp ) ) { return { error : 'bad_param' , error_description : `invalid timestamp ${ event . timestamp }` } ; } if ( event . event_type === undefined ) { return { error : 'missing_param' , error_description : 'missing enum field "event_type"' } ; } if ( ! isEnum ( VEHICLE_EVENTS , event . event_type ) ) { return { error : 'bad_param' , error_description : `invalid event_type ${ event . event_type }` } ; } if ( event . event_type_reason && ! isEnum ( VEHICLE_REASONS , event . event_type_reason ) ) { return { error : 'bad_param' , error_description : `invalid event_type_reason ${ event . event_type_reason }` } ; } if ( event . trip_id === '' ) { event . trip_id = null ; } const { trip_id } = event ; if ( trip_id !== null && trip_id !== undefined && ! isUUID ( event . trip_id ) ) { return { error : 'bad_param' , error_description : `invalid trip_id ${ event . trip_id } is not a UUID` } ; } function missingTripId ( ) { if ( ! trip_id ) { return { error : 'missing_param' , error_description : 'missing trip_id' } ; } return null ; } switch ( event . event_type ) { case VEHICLE_EVENTS . trip_start : return badTelemetry ( event . telemetry ) || missingTripId ( ) ; case VEHICLE_EVENTS . trip_end : return badTelemetry ( event . telemetry ) || missingTripId ( ) ; case VEHICLE_EVENTS . trip_enter : return badTelemetry ( event . telemetry ) || missingTripId ( ) ; case VEHICLE_EVENTS . trip_leave : return badTelemetry ( event . telemetry ) || missingTripId ( ) ; case VEHICLE_EVENTS . service_start : case VEHICLE_EVENTS . service_end : case VEHICLE_EVENTS . provider_pick_up : case VEHICLE_EVENTS . provider_drop_off : return badTelemetry ( event . telemetry ) ; case VEHICLE_EVENTS . register : case VEHICLE_EVENTS . deregister : case VEHICLE_EVENTS . reserve : case VEHICLE_EVENTS . cancel_reservation : return null ; default : logger . warn ( `unsure how to validate mystery event_type ${ event . event_type }` ) ; break ; } return null ; } export function lower ( s ) { if ( typeof s === 'string' ) { return s . toLowerCase ( ) ; } return s ; } export async function writeTelemetry ( telemetry ) { const recorded_telemetry = await db . writeTelemetry ( Array . isArray ( telemetry ) ? telemetry : [ telemetry ] ) ; try { await Promise . all ( [ cache . writeTelemetry ( recorded_telemetry ) , stream . writeTelemetry ( recorded_telemetry ) ] ) ; } catch ( err ) { logger . warn ( `Failed to write telemetry to cache/stream, ${ err }` ) ; } return recorded_telemetry ; } export async function refresh ( device_id , provider_id ) { const device = await db . readDevice ( device_id , provider_id ) ; await cache . writeDevice ( device ) ; try { const event = await db . readEvent ( device_id ) ; await cache . writeEvent ( event ) ; } catch ( err ) { logger . info ( 'no events for' , device_id , err ) ; } try { await db . readTelemetry ( device_id ) ; } catch ( err ) { logger . info ( 'no telemetry for' , device_id , err ) ; } return 'done' ; } /**
 * for some functions we will want to validate the :device_id param
 */ export async function validateDeviceId ( req , res , next ) { const { device_id } = req . params ; if ( ! device_id ) { logger . warn ( 'agency: missing device_id' , req . originalUrl ) ; res . status ( 400 ) . send ( { error : 'missing_param' , error_description : 'missing device_id' } ) ; return ; } if ( device_id && ! isUUID ( device_id ) ) { logger . warn ( 'agency: bogus device_id' , device_id , req . originalUrl ) ; res . status ( 400 ) . send ( { error : 'bad_param' , error_description : `invalid device_id ${ device_id } is not a UUID` } ) ; return ; } next ( ) ; } export async function writeRegisterEvent ( device , recorded ) { const event = { device_id : device . device_id , provider_id : device . provider_id , event_type : VEHICLE_EVENTS . register , event_type_reason : null , telemetry : null , timestamp : recorded , trip_id : null , recorded , telemetry_timestamp : undefined , service_area_id : null } ; try { const recorded_event = await db . writeEvent ( event ) ; try { await Promise . all ( [ cache . writeEvent ( recorded_event ) , stream . writeEvent ( recorded_event ) ] ) ; } catch ( err ) { logger . warn ( '/event exception cache/stream' , err ) ; } } catch ( err ) { logger . error ( 'writeRegisterEvent failure' , err ) ; throw new Error ( 'writeEvent exception db' ) ; } } export function computeCompositeVehicleData ( payload ) { const { device , event , telemetry } = payload ; const composite = { ... device } ; if ( event ) { composite . prev_event = event . event_type ; composite . updated = event . timestamp ; composite . status = ( EVENT_STATUS_MAP [ event . event_type as VEHICLE_EVENT ] || 'unknown' ) as VEHICLE_STATUS ; } else { composite . status = VEHICLE_STATUSES . inactive ; composite . prev_event = VEHICLE_EVENTS . deregister ; } if ( telemetry ) { if ( telemetry . gps ) { composite . gps = telemetry . gps ; } } return composite ; } const normalizeTelemetry = ( telemetry ) => { if ( isArray ( telemetry ) ) { return telemetry [ 0 ] ; } return telemetry ; } ; export async function readPayload ( device_id ) { const payload = { } ; try { payload . device = await db . readDevice ( device_id ) ; } catch ( err ) { logger . error ( err ) ; } try { payload . event = await cache . readEvent ( device_id ) ; if ( payload . event ) { if ( payload . event . telemetry ) { payload . telemetry = normalizeTelemetry ( payload . event . telemetry ) ; } } } catch ( err ) { logger . error ( err ) ; } return payload ; }