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

    id:          int
    sounding_id: int
    elapsed:     int
    pressure:    float
    pressure_qa: str
    height:      float
    height_qa:   str
    temp:        float
    temp_qa:     str
    humidity:    float
    dewpoint:    float
    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)
        )

    def saturated_vapor_pressure(self) -> float:
        return 6.11 * 10 * (
            (7.5 * self.temp) / (237.3 * self.temp)
        )

    def mixing_ratio(self) -> float:
        e = self.vapor_pressure()

        return 621.97 * (e / (self.pressure - e))

    def saturated_mixing_ratio(self) -> float:
        es = self.saturated_vapor_pressure()

        return 621.97 * (es / (self.pressure - es))

    def lcl(self) -> float: # meters
        return (self.temp - self.dewpoint) / 0.008

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

    def lapse(self, h: float, rate=None) -> float:
        if rate is None:
            rate = LAPSE_RATE_MOIST if self.is_saturated() else LAPSE_RATE_DRY

        hd = h - self.height

        return self.temp - (rate * (hd / 1000))

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_released <= :timestamp
            order by
                timestamp_released desc
            limit 1
        """

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

        pass

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

        sounding = st.fetchone()
        sounding.samples = list(db.query(SoundingSample, {
            'sounding_id': sounding.id
        }, [['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_released <= :timestamp
            order by
                distance asc,
                timestamp_released 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, {
            'sounding_id': sounding.id
        }, [['pressure', DatabaseOrder.DESC]]).fetchall())

        return sounding