xmet/lib/xmet/sounding.py

import datetime
import shapely

from xmet.db    import Database, DatabaseTable, DatabaseOrder
from xmet.coord import COORD_SYSTEM

LAPSE_RATE_DRY   = 9.8 # degrees C per 1000m
LAPSE_RATE_MOIST = 4.0

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'
    )

    def __init__(self):
        super().__init__()
        self.id: int = None
        self.sounding_id: int   = None
        self.elapsed:     int   = None
        self.pressure:    float = None
        self.pressure_qa: str   = None
        self.height:      float = None
        self.height_qa:   str   = None
        self.temp:        float = None
        self.temp_qa:     str   = None
        self.humidity:    float = None
        self.dewpoint:    float = None
        self.wind_dir:    float = None
        self.wind_speed:  float = None

    def is_saturated(self) -> bool:
        return self.humidity >= 100.0

class Sounding(DatabaseTable):
    __slots__ = (
        'id', 'station', 'timestamp_observed', 'timestamp_released',
        'data_source_pressure', 'data_source_other', 'samples', 'location'
    )

    __table__ = 'xmet_sounding'
    __key__   = 'id'

    __columns__ = (
        'id', 'station', 'timestamp_observed', 'timestamp_released',
        'data_source_pressure', 'data_source_other', 'location'
    )

    __columns_read__ = {
        'location': 'ST_AsText(location) as location'
    }

    __values_read__ = {
        'location': shapely.from_wkt
    }

    __columns_write__ = {
        'location': 'ST_GeomFromText(:location, {crs})'.format(crs=COORD_SYSTEM)
    }

    __values_write__ = {
        'location': lambda v: {'location': shapely.to_wkt(v)}
    }

    id:                   int
    station:              str
    timestamp_observed:   datetime.datetime
    timestamp_released:   datetime.datetime
    data_source_pressure: str
    data_source_other:    str
    location:             shapely.Point
    samples:              list[SoundingSample]

    def __init__(self):
        super().__init__()
        self.id = None

    @staticmethod
    def valid_by_station(db: Database,
                         station: str,
                         timestamp: datetime.datetime=None):
        sql = """
            select
                id, station, timestamp_observed, timestamp_released,
                data_source_pressure, data_source_other,
                ST_AsText(location) as location
            from
                xmet_sounding
            where
                station = :station
                and timestamp_observed <= :timestamp
            order by
                timestamp_observed desc
            limit 1
        """

        if timestamp is None:
            timestamp = datetime.datetime.now(datetime.UTC)

        st = db.query_sql(Sounding, sql, {
            'station':   station,
            'timestamp': timestamp
        })

        sounding = st.fetchone()
        sounding.samples = list(db.query(SoundingSample,
            clauses = [
                'sounding_id = :sounding_id'
            ],
            values = {
                'sounding_id': sounding.id
            },
            order_by = [[
                'pressure', DatabaseOrder.DESC
            ]]).fetchall())

        return sounding

    @staticmethod
    def valid_by_location(db: Database,
                          location: shapely.Point,
                          timestamp: datetime.datetime):
        sql = """
            select
                id, station, timestamp_observed, timestamp_released,
                data_source_pressure, data_source_other,
                ST_AsText(location) as location,
                ST_Distance(location, MakePoint(:lon, :lat, {crs})) as distance
            from
                xmet_sounding
            where
                timestamp_observed <= :timestamp
            order by
                distance asc,
                timestamp_observed desc
            limit 1
        """.format(crs=COORD_SYSTEM)

        if timestamp is None:
            timestamp = datetime.datetime.now(datetime.UTC)

        st = db.query_sql(Sounding, sql, {
            'lon':       location.x,
            'lat':       location.y,
            'timestamp': timestamp
        })

        sounding = st.fetchone()
        sounding.samples = list(db.query(SoundingSample,
            clauses = [
                'sounding_id = :sounding_id'
            ],
            values = {
                'sounding_id': sounding.id
            },
            order_by = [[
                'pressure', DatabaseOrder.DESC
            ]]).fetchall())

        return sounding