Updates to support the CAC data (#14)

* Cleaned up the ingester to work better for CAC data

* Added support for ingesting logging intervals

* Support for ingesting instrument and status

* Added flag ingest method

* Fixed bug in the flag ingest process

  Flags with null notes where not being matched

* Added the start of some testing files

  The test_flags script is not an automated test yet but I thought the
  data files and process would still be helpful to have committed

* Flagging updates and hourly data rollup fixes

* Some cleanup and bug fixes

  Added some methods that help in the testing/dev environments

Author: caparker

ingest/etl_process_measurements.sql

@@ -52,12 +52,16 @@ FROM staging_measurements;
 -- that duplicate sensors with the same ingest/source id are created
 	-- this is a short term fix
 	-- a long term fix would not allow duplicate source_id's
-WITH ranked_sensors AS (
+WITH staged_sensors AS (
+  -- this first part signficantly speeds it up on slow machines
+  SELECT DISTINCT ingest_id
+  FROM staging_measurements
+), ranked_sensors AS (
   SELECT s.sensors_id
 	, s.source_id
 	, RANK() OVER (PARTITION BY s.source_id ORDER BY added_on ASC) as rnk
 	FROM sensors s
-	JOIN staging_measurements m ON (s.source_id = m.ingest_id)
+	JOIN staged_sensors m ON (s.source_id = m.ingest_id)
 ), active_sensors AS (
 	SELECT source_id
 	, sensors_id
@@ -68,6 +72,7 @@ WITH ranked_sensors AS (
 	FROM active_sensors s
 	WHERE s.source_id=ingest_id;
 
+
 -- Now we have to fill in any missing information
 -- first add the nodes and systems that dont exist
 -- add just the bare minimum amount of data to the system
@@ -285,6 +290,7 @@ INSERT INTO sensors_rollup (
   , value_latest
   , value_count
   , value_avg
+  , value_sd
   , value_min
   , value_max
   , geom_latest
@@ -299,6 +305,7 @@ WITH numbered AS (
    , sum(1) OVER (PARTITION BY sensors_id) as value_count
    , min(datetime) OVER (PARTITION BY sensors_id) as datetime_min
    , avg(value) OVER (PARTITION BY sensors_id) as value_avg
+   , stddev(value) OVER (PARTITION BY sensors_id) as value_sd
    , row_number() OVER (PARTITION BY sensors_id ORDER BY datetime DESC) as rn
   FROM staging_inserted_measurements
 ), latest AS (
@@ -308,6 +315,7 @@ WITH numbered AS (
    , value
    , value_count
    , value_avg
+   , value_sd
    , datetime_min
    , lat
    , lon
@@ -320,6 +328,7 @@ SELECT l.sensors_id
 , l.value -- last value
 , l.value_count
 , l.value_avg
+, l.value_sd
 , l.value -- min
 , l.value -- max
 , public.pt3857(lon, lat)
@@ -348,12 +357,23 @@ SET datetime_last = GREATEST(sensors_rollup.datetime_last, EXCLUDED.datetime_las
 
 
 -- Update the table that will help to track hourly rollups
-INSERT INTO hourly_stats (datetime)
-  SELECT date_trunc('hour', datetime)
-  FROM staging_inserted_measurements
-  GROUP BY 1
-ON CONFLICT (datetime) DO UPDATE
-SET modified_on = now();
+-- this is a replacement to the hourly stats table
+  WITH inserted_hours AS (
+    -- first we group things, adding an hour to make it time-ending after truncating
+    SELECT datetime + '1h'::interval as datetime
+    , utc_offset(datetime + '1h'::interval, tz.tzid) as tz_offset
+    FROM staging_inserted_measurements m
+    JOIN sensors s ON (s.sensors_id = m.sensors_id)
+    JOIN sensor_systems sy ON (s.sensor_systems_id = sy.sensor_systems_id)
+    JOIN sensor_nodes sn ON (sy.sensor_nodes_id = sn.sensor_nodes_id)
+    JOIN timezones tz ON (sn.timezones_id = tz.timezones_id)
+    GROUP BY 1, 2
+   )
+    INSERT INTO hourly_data_queue (datetime, tz_offset)
+    SELECT as_utc_hour(datetime, tz_offset), tz_offset
+    FROM inserted_hours
+    ON CONFLICT (datetime, tz_offset) DO UPDATE
+    SET modified_on = now();
 
 
 --Update the export queue/logs to export these records

ingest/etl_process_nodes.sql

@@ -35,7 +35,7 @@ WHERE units IN ('µg/m��','��g/m³', 'ug/m3');
 
 
 
--- match the locations to the nodes using the source_name/id combo
+-- match the locations to existing nodes using the source_name/id combo
 UPDATE staging_sensornodes
 SET sensor_nodes_id = s.sensor_nodes_id
 , timezones_id = s.timezones_id
@@ -109,7 +109,9 @@ SELECT site_name
 , metadata
 , source_id
 , timezones_id
-, get_providers_id(source_name)
+-- default to the unknown provider
+-- just to make sure we have one set
+, COALESCE(get_providers_id(source_name), 1)
 , countries_id
 FROM staging_sensornodes
 WHERE sensor_nodes_id IS NULL
@@ -156,17 +158,20 @@ FROM r;
 -- Sensor Systems --
 --------------------
 
+
 -- make sure that we have a system entry for every ingest_id
 -- this is to deal with fetchers that do not add these data
 INSERT INTO staging_sensorsystems (sensor_nodes_id, ingest_id, fetchlogs_id, metadata)
 SELECT sensor_nodes_id
-, source_id -- the ingest_id has the source_name in it and we dont need/want that
+--, source_id -- the ingest_id has the source_name in it and we dont need/want that
+, ingest_id
 , fetchlogs_id
 , '{"note":"automatically added for sensor node"}'
 FROM staging_sensornodes
-WHERE is_new
+WHERE is_new AND ingest_id NOT IN (SELECT ingest_sensor_nodes_id FROM staging_sensorsystems)
 ON CONFLICT (ingest_id) DO UPDATE
-  SET sensor_nodes_id = EXCLUDED.sensor_nodes_id;
+  SET sensor_nodes_id = EXCLUDED.sensor_nodes_id
+  ;
 
 -- Now match the sensor nodes to the system
 UPDATE staging_sensorsystems
@@ -197,15 +202,18 @@ SELECT COUNT(1) INTO __rejected_systems
 FROM r;
 
 -- And finally we add/update the sensor systems
-INSERT INTO sensor_systems (sensor_nodes_id, source_id, metadata)
+INSERT INTO sensor_systems (sensor_nodes_id, source_id, instruments_id, metadata)
 SELECT sensor_nodes_id
-, ingest_id
+, s.ingest_id
+, i.instruments_id
 , metadata
-FROM staging_sensorsystems
+FROM staging_sensorsystems s
+LEFT JOIN instruments i ON (s.instrument_ingest_id = i.ingest_id)
 WHERE sensor_nodes_id IS NOT NULL
-GROUP BY sensor_nodes_id, ingest_id, metadata
+GROUP BY sensor_nodes_id, s.ingest_id, instruments_id, metadata
 ON CONFLICT (sensor_nodes_id, source_id) DO UPDATE SET
     metadata=COALESCE(sensor_systems.metadata, '{}') || COALESCE(EXCLUDED.metadata, '{}')
+    , instruments_id = EXCLUDED.instruments_id
     , modified_on = now();
 
 ----------------------------
@@ -266,10 +274,11 @@ AND sensors.source_id = staging_sensors.ingest_id;
 
 
 UPDATE staging_sensors
-SET measurands_id = measurands.measurands_id
-from measurands
-WHERE staging_sensors.measurand=measurands.measurand
-and staging_sensors.units=measurands.units;
+SET measurands_id = m.measurands_id
+FROM (SELECT measurand, MIN(measurands_id) AS measurands_id FROM measurands GROUP BY measurand) as m
+WHERE staging_sensors.measurand=m.measurand
+--AND staging_sensors.units=measurands.units
+;
 
 
 WITH r AS (
@@ -290,20 +299,34 @@ INSERT INTO sensors (
   source_id
 , sensor_systems_id
 , measurands_id
+, data_logging_period_seconds
+, data_averaging_period_seconds
+, sensor_statuses_id
 , metadata)
 SELECT ingest_id
 , sensor_systems_id
 , measurands_id
+, logging_interval_seconds
+, averaging_interval_seconds
+, COALESCE(ss.sensor_statuses_id, 1)
 , metadata
-FROM staging_sensors
+FROM staging_sensors s
+LEFT JOIN sensor_statuses ss ON (ss.short_code = s.status)
 WHERE measurands_id is not null
 AND sensor_systems_id is not null
 GROUP BY ingest_id
 , sensor_systems_id
 , measurands_id
+, logging_interval_seconds
+, averaging_interval_seconds
+, ss.sensor_statuses_id
 , metadata
 ON CONFLICT (sensor_systems_id, measurands_id, source_id) DO UPDATE
 SET metadata = COALESCE(sensors.metadata, '{}') || COALESCE(EXCLUDED.metadata, '{}')
+  , data_logging_period_seconds = EXCLUDED.data_logging_period_seconds
+  , data_averaging_period_seconds = EXCLUDED.data_averaging_period_seconds
+  , sensor_statuses_id = EXCLUDED.sensor_statuses_id
+  , modified_on = now()
 RETURNING 1)
 SELECT COUNT(1) INTO __inserted_sensors
 FROM inserts;
@@ -327,6 +350,61 @@ RETURNING 1)
 SELECT COUNT(1) INTO __rejected_sensors
 FROM r;
 
+
+-- update the period so that we dont have to keep doing it later
+-- we could do this on import as well if we feel this is slowing us down
+UPDATE staging_flags
+  SET period = tstzrange(COALESCE(datetime_from, '-infinity'::timestamptz),COALESCE(datetime_to, 'infinity'::timestamptz), '[]');
+
+-- Now we have to match things
+-- get the right node id and sensors id for the flags
+UPDATE staging_flags
+SET sensors_id = s.sensors_id
+  , sensor_nodes_id = sy.sensor_nodes_id
+FROM sensors s
+JOIN sensor_systems sy ON (s.sensor_systems_id = sy.sensor_systems_id)
+WHERE staging_flags.sensor_ingest_id = s.source_id;
+
+-- and then get the right flags_id
+UPDATE staging_flags
+SET flag_types_id = ft.flag_types_id
+FROM flag_types ft
+WHERE split_part(staging_flags.ingest_id, '::', 1) = ft.ingest_id;
+
+-- now we should look to see if we should be just extending a flag
+UPDATE staging_flags sf
+  SET flags_id = fm.flags_id
+  FROM flags fm
+  -- where the core information is the same (exactly)
+  WHERE sf.sensor_nodes_id = fm.sensor_nodes_id
+  AND sf.flag_types_id = fm.flag_types_id
+  AND ((sf.note = fm.note) OR (sf.note IS NULL AND fm.note IS NULL))
+  -- the periods touch or overlap
+  AND fm.period && sf.period
+  -- and the flagged record sensors contains the current sensors
+  AND fm.sensors_ids @> ARRAY[sf.sensors_id];
+
+-- and finally we will insert the new flags
+INSERT INTO flags (flag_types_id, sensor_nodes_id, sensors_ids, period, note)
+  SELECT flag_types_id
+  , sensor_nodes_id
+  , CASE WHEN sensors_id IS NOT NULL THEN ARRAY[sensors_id] ELSE NULL END
+  , period
+  , note
+  FROM staging_flags
+  WHERE flag_types_id IS NOT NULL
+  AND sensor_nodes_id IS NOT NULL
+  AND flags_id IS NULL;
+
+-- And then update any that need to be updated
+ UPDATE flags fm
+  SET period = sf.period + fm.period
+  , note = sf.note
+  , modified_on = now()
+  FROM staging_flags sf
+  WHERE sf.flags_id = fm.flags_id;
+
+
 ------------------
 -- Return stats --
 ------------------