Initial implementation of IGRA ingest

Dockerfile

@@ -9,7 +9,11 @@ RUN mkdir -p /var/opt/xmet/lib/xmet
 RUN mkdir -p /var/opt/xmet/bin
 RUN mkdir -p /var/lib/xmet
 
-COPY db/xmet.sql doc/radars.tsv doc/wfo.tsv /tmp
+COPY db/xmet.sql \
+      doc/radars.tsv \
+      doc/wfo.tsv \
+    doc/igra2-station-list.txt /tmp
+
 COPY lib/xmet/*.py /var/opt/xmet/lib/xmet
 COPY bin/xmet-nexrad-archive bin/xmet-db-init /var/opt/xmet/bin
 
@@ -20,6 +24,7 @@ RUN sqlite3 -init /tmp/xmet.sql /var/lib/xmet/xmet.db .quit
 RUN /var/opt/xmet/bin/xmet-db-init \
  /var/lib/xmet/xmet.db \
  /tmp/radars.tsv \
- /tmp/wfo.tsv
+ /tmp/wfo.tsv \
+ /tmp/igra2-station-list.txt
 
 ENTRYPOINT ["/var/opt/xmet/bin/xmet-nexrad-archive", "/var/lib/xmet/xmet.db"]

bin/xmet-db-init

@@ -5,14 +5,16 @@ import argparse
 from xmet.db    import Database
 from xmet.radar import Radar
 from xmet.wfo   import WFO
+from xmet.igra  import IGRAStation
 
 parser = argparse.ArgumentParser(
     description = 'Initialize NEXRAD radar site database table'
 )
 
-parser.add_argument('db',         help='Path to SQLite3 database')
-parser.add_argument('radars-tsv', help='Path to NEXRAD radar station TSV file')
-parser.add_argument('wfo-tsv',    help='Path to forecast office TSV file')
+parser.add_argument('db',            help='Path to SQLite3 database')
+parser.add_argument('radars-tsv',    help='Path to NEXRAD radar station TSV file')
+parser.add_argument('wfo-tsv',       help='Path to forecast office TSV file')
+parser.add_argument('igra-stations', help='Path to IGRA station list')
 
 args = parser.parse_args()
 
@@ -26,4 +28,7 @@ for radar in Radar.each_from_tsv(getattr(args, 'radars-tsv')):
 for wfo in WFO.each_from_tsv(getattr(args, 'wfo-tsv')):
     db.add(wfo)
 
+for station in IGRAStation.each_from_file(getattr(args, 'igra-stations')):
+    db.add(station)
+
 db.commit()

bin/xmet-igra-ingest

@@ -0,0 +1,20 @@
+#! /usr/bin/env python3
+
+import sys
+
+from xmet.db   import Database
+from xmet.igra import IGRAReader
+
+db = Database.connect(sys.argv[1])
+db.execute('begin transaction')
+
+for path in sys.argv[2:]:
+    with open(path, 'r') as fh:
+        for sounding in IGRAReader.each_from_file(path):
+            db.add(sounding)
+
+            for sample in sounding.samples:
+                sample.sounding_id = sounding.id
+                db.add(sample)
+
+db.commit()

db/xmet.sql

@@ -119,7 +119,7 @@ select AddGeometryColumn('xmet_sounding', 'coord', 4326, 'POINT'),
 create table xmet_sounding_sample (
     id          INTEGER PRIMARY KEY NOT NULL,
     sounding_id INTEGER NOT NULL,
-    elapsed     INTEGER NOT NULL,
+    elapsed     INTEGER,
     pressure    FLOAT,
     pressure_qa TEXT NOT NULL,
     height      FLOAT,

lib/xmet/db.py

@@ -127,23 +127,20 @@ class Database():
     def value_placeholders(self, table, obj) -> list:
         ci = getattr(table, '__columns_write__', None)
 
-        if ci is None:
-            return [f':{c}' for c in table.__columns__]
-        else:
-            ret = list()
+        ret = list()
 
-            for c in table.__columns__:
-                v = getattr(obj, c, None)
+        for c in table.__columns__:
+            v = getattr(obj, c, None)
 
-                if v is None:
-                    continue
+            if v is None:
+                continue
 
-                if c in ci:
-                    ret.append(ci[c])
-                else:
-                    ret.append(f':{c}')
+            if ci is not None and c in ci:
+                ret.append(ci[c])
+            else:
+                ret.append(f':{c}')
 
-            return ret
+        return ret
 
     def row_values(self, table, obj) -> dict:
         ret = dict()

lib/xmet/igra.py

@@ -150,6 +150,19 @@ class IGRAReader():
 
         return sounding
 
+    @staticmethod
+    def each_from_file(path: str):
+        with open(path, 'r') as fh:
+            reader = IGRAReader(fh)
+
+            while True:
+                sounding = reader.read()
+
+                if sounding is None:
+                    break
+
+                yield sounding
+
 def cols(text: str, start: int, end: int):
     a = start - 1
     b = end
@@ -214,3 +227,14 @@ class IGRAStation(DatabaseTable):
         station.location   = shapely.Point(lon, lat)
 
         return station
+
+    @staticmethod
+    def each_from_file(path: str):
+        with open(path, 'r') as fh:
+            while True:
+                line = fh.readline()
+
+                if line == '' or line is None:
+                    break
+
+                yield IGRAStation.parse_station(line)

lib/xmet/sounding.py

@@ -1,16 +1,28 @@
 import datetime
 import shapely
 
+from xmet.db    import DatabaseTable
+from xmet.coord import COORD_SYSTEM
+
 LAPSE_RATE_DRY   = 9.8 # degrees C per 1000m
 LAPSE_RATE_MOIST = 4.0
 
-class SoundingSample():
+class SoundingSample(DatabaseTable):
     __slots__ = (
         'id', 'sounding_id', 'elapsed', 'pressure', 'pressure_qa',
         'height', 'height_qa', 'temp', 'temp_qa', 'humidity',
         'dewpoint', 'wind_dir', 'wind_speed'
     )
 
+    __table__ = 'xmet_sounding_sample'
+    __key__   = 'id'
+
+    __columns__ = (
+        'id', 'sounding_id', 'elapsed', 'pressure', 'pressure_qa',
+        'height', 'height_qa', 'temp', 'temp_qa', 'humidity',
+        'dewpoint', 'wind_dir', 'wind_speed'
+    )
+
     id:          int
     sounding_id: int
     elapsed:     int
@@ -25,6 +37,10 @@ class SoundingSample():
     wind_dir:    float
     wind_speed:  float
 
+    def __init__(self):
+        super().__init__()
+        self.id = None
+
     def vapor_pressure(self) -> float:
         return 6.11 * 10 * (
             (7.5 * self.dewpoint) / (237.3 * self.dewpoint)
@@ -65,6 +81,30 @@ class Sounding():
         'data_source_pressure', 'data_source_other', 'samples', 'coord'
     )
 
+    __table__ = 'xmet_sounding'
+    __key__   = 'id'
+
+    __columns__ = (
+        'id', 'station', 'timestamp_observed', 'timestamp_released',
+        'data_source_pressure', 'data_source_other', 'coord'
+    )
+
+    __columns_read__ = {
+        'coord': 'ST_AsText(coord) as coord'
+    }
+
+    __values_read__ = {
+        'coord': shapely.from_wkt
+    }
+
+    __columns_write__ = {
+        'coord': 'ST_GeomFromText(:coord, {crs})'.format(crs=COORD_SYSTEM)
+    }
+
+    __values_write__ = {
+        'coord': lambda v: {'coord': shapely.to_wkt(v)}
+    }
+
     id:                   int
     station:              str
     timestamp_observed:   datetime.datetime
@@ -73,3 +113,7 @@ class Sounding():
     data_source_other:    str
     coord:                shapely.Point
     samples:              list[SoundingSample]
+
+    def __init__(self):
+        super().__init__()
+        self.id = None