text: with map_subset as (SELECT * FROM smart_columbus_cve__captcha where os_partition = '2020_12' and sourcedevice = 'morse_morsecrossing_cvcp' and messagetype = 'MAP' and timestamp > '2020-12-09T07:00:00.000Z' and timestamp < '2020-12-09T07:05:00.000Z' limit 1),
  extracted_values as (
      select json_extract(messageBody, '$.intersections[0].laneSet[0]') as cursor, 
             json_extract(messageBody, '$.intersections[0].laneWidth') as laneWidth, 
             cast(json_extract_scalar(messageBody, '$.intersections[0].refPoint.lat') as double) as refLat, 
             cast(json_extract_scalar(messageBody, '$.intersections[0].refPoint.long') as double) as refLon, 
  
             transform(
                 cast(json_extract(messageBody, '$.intersections[0].laneSet') AS ARRAY<JSON>),
                   x -> element_at(transform(cast(JSON_EXTRACT(x, '$.nodeList[1]') as ARRAY<JSON>),
                     y -> cast(ROW(
                                   JSON_EXTRACT_scalar(x, '$.laneID'), 
                                   JSON_EXTRACT_scalar(x, '$.ingressApproach'),
                                   cast(transform(cast(JSON_EXTRACT(x, '$.connectsTo') as ARRAY<JSON>), connection -> JSON_EXTRACT_SCALAR(connection, '$.signalGroup')) as ARRAY<VARCHAR>), 
                                   JSON_EXTRACT_scalar(y, '$.delta[1].x'), 
                                   JSON_EXTRACT_scalar(y, '$.delta[1].y')
                                  ) 
                          as ROW(laneID VARCHAR, ingressApproach VARCHAR, signals ARRAY<VARCHAR>, x DOUBLE, y DOUBLE))
                   ), 1)
              ) as deltas,
      * from map_subset),
      
  extracted_coordinates as (select transform(deltas, x -> cast(ROW(x.laneID, x.ingressApproach, x.signals, refLat + x.y, refLon + x.x) as ROW(laneID VARCHAR, ingressApproach VARCHAR, signals ARRAY<VARCHAR>, lat DOUBLE, lon DOUBLE))) as coordinates, * from extracted_values),
  
  stop_points as (SELECT ingressApproach, laneID, signals, lat / 10000000 as lat, lon / 10000000 as lon, cursor FROM extracted_coordinates CROSS JOIN UNNEST(coordinates) AS t ( laneID, ingressApproach, signals, lat, lon) where ingressApproach is not null),
  
  lane_signal_groups as (select laneID, lat, lon, signalGroup from stop_points CROSS JOIN UNNEST(signals) as t (signalGroup)),
  
  
  spat_subset as (SELECT * FROM smart_columbus_cve__data_subset_smart_columbus_connected_vehicle_environment_project_20201213 where os_partition = '2020_12' and sourcedevice = 'morse_morsecrossing_cvcp' and messagetype = 'SPAT'),
  
  --  and timestamp > '2020-12-13T10:00:00.000Z' and timestamp < '2020-12-13T14:00:00.000Z'
  
  signal_states as (select sourcedevice, timestamp, transform(cast(json_extract(messagebody, '$.intersections[0].states') as ARRAY<JSON>), x -> cast(ROW(json_extract(x, '$.signalGroup'), json_extract(x, '$["state-time-speed"][0].eventState')) as ROW(signalGroup VARCHAR, state VARCHAR))) as signalStates from spat_subset),
  
  extracted_signal_states as (select sourceDevice, timestamp, signalGroup, state from signal_states CROSS JOIN UNNEST(signalStates) AS t ( signalGroup, state)),
  
  spat_states as (select state, timestamp, laneID from lane_signal_groups join extracted_signal_states using (signalGroup)),
  
  spat_truthy_states as (select state = 'stop-And-Remain' as signal_group_red, timestamp, laneID from spat_states),
  
  spat_lane_states as (select every(signal_group_red) as lane_red_light, laneID, timestamp from spat_truthy_states group by laneID, timestamp),
  
  --
  
  -- bsms as (SELECT * FROM smart_columbus_cve__data_subset_smart_columbus_connected_vehicle_environment_project_20201213 where messagetype = 'BSM'),
  
  bsm_subset as (select * from smart_columbus_cve__data_subset_smart_columbus_connected_vehicle_environment_project_20201213 where os_partition = '2020_12' and messagetype = 'BSM' and sourcedevice = 'morse_morsecrossing_cvcp'), 
  
  identified_locations as (select cast(json_extract_scalar(messagebody, '$.coreData.lat') as double) / 10000000 as lat, cast(json_extract_scalar(messagebody, '$.coreData.long') as double) / 10000000 as long, json_extract_scalar(messagebody, '$.coreData.id') as id, timestamp, cast(json_extract_scalar(messagebody, '$.coreData.speed') as double) as speed from bsm_subset),
  
  bsms_in_range as (select il.lat, il.long, id, il.timestamp, speed from identified_locations il join stop_points sp on ST_Distance(ST_Point(il.lat, il.long), ST_Point(sp.lat, sp.lon)) < 0.00002),
  
  distance_to_lane as (select bsm.lat, bsm.long, laneID, timestamp, speed, ST_Distance(ST_Point(bsm.lat, bsm.long), ST_Point(sp.lat, sp.lon)) as distance from bsms_in_range bsm join stop_points sp on 1=1),
  
  bsms as (select min_by(laneID, distance) as laneID, lat, long, timestamp, speed from distance_to_lane where speed > 1.0 group by lat, long, timestamp, speed),
  
  --
  
  bsms_unix as (select timestamp, to_unixtime(from_iso8601_timestamp(timestamp)) as unixtime, laneID from bsms),
  spats_unix as (select timestamp, to_unixtime(from_iso8601_timestamp(timestamp)) as unixtime, laneID from spat_lane_states),
  
  unix_join as (select bsms.timestamp as bsm_timestamp, spats.timestamp as spat_timestamp, spats.laneID from bsms_unix as bsms join spats_unix as spats on abs(bsms.unixtime - spats.unixtime) < 0.01 and bsms.laneID = spats.laneID),
  
  deduped_unix_join as (select min(spat_timestamp) as spat_timestamp, bsm_timestamp, laneID from unix_join group by bsm_timestamp, laneID)
  
  select bsms.lat, bsms.long, bsms.timestamp, lane_red_light from bsms join deduped_unix_join dd on bsms.timestamp = dd.bsm_timestamp and bsms.laneID = dd.laneID
                     join spat_lane_states on spat_lane_states.timestamp = dd.spat_timestamp and spat_lane_states.laneID = dd.laneID;