Monitor GTFS-realtime vehicle positions using Cronitor.io.

lambda_handler.py

from __future__ import print_function

import os
import json
import urllib2
from datetime import datetime

LAMBDA_API = 'https://vni1fkrx5b.execute-api.us-east-1.amazonaws.com/prod/gtfsrt-debug'
GTFSRT_FEED = 'https://data.texas.gov/download/eiei-9rpf/application/octet-stream'
CRONITOR = '<YOUR-CRONITOR-URL>'


def lambda_handler(event, context):
    url = '{}?url={}'.format(LAMBDA_API, GTFSRT_FEED)
    res = urllib2.urlopen(url)
    data = json.load(res)
    
    now = datetime.now().isoformat()
    
    print('*********************************')
    if len(data['vehicle_positions']) > 0:
        print('{} - SUCCESS - vehicle positions is NOT empty, url: {}'.format(now, url))
        urllib2.urlopen(CRONITOR)
    else:
        print('{} - FAILURE - vehicle positions is empty, url: {}'.format(now, url))
    print('*********************************')
    

from __future__ import print_function

import json
import urllib2
from datetime import datetime

LAMBDA_API = 'https://lnykjry6ze.execute-api.us-west-2.amazonaws.com/prod/gtfsrt-debug'
GTFSRT_FEED = 'https://data.texas.gov/download/eiei-9rpf/application/octet-stream'
CRONITOR = '...'


def lambda_handler(event, context):
    url = '{}?url={}'.format(LAMBDA_API, GTFSRT_FEED)
    res = urllib2.urlopen(url)
    data = json.load(res)

    date = datetime.now()
    now = date.isoformat()

    print('*********************************')
    if len(data['vehicle_positions']) > 0:
        print('{} - SUCCESS - vehicle positions is NOT empty, url: {}'.format(now, url))
        urllib2.urlopen(CRONITOR)
    elif date.hour > 6 and date.hour <= 11:
        print('{} - FAILURE - vehicle positions is empty, but IGNORED because hour is {}, url: {}'.format(now, date.hour, url))
        urllib2.urlopen(CRONITOR)
    else:
        print('{} - FAILURE - vehicle positions is empty, url: {}'.format(now, url))
    print('*********************************')