nexrad-archive/lib/nexrad/station.py

import csv
import re

from nexrad.coord import Coord

RE_PARSE = re.compile(r'^\s*(\d+)([NS]*)\s+/\s+(\d+)([EW]*)\s*$')

def parse_int(text: str):
    size = len(text)

    degree = int(text[0:size-4])
    minute = int(text[size-4:size-2])
    second = int(text[size-2:])

    return degree + (minute  / 60) + (second / 3600)

def parse(text: str):
    match = RE_PARSE.match(text)

    if match is None:
        raise Exception('Invalid coordinates \'%s\'' % text)

    sign_lat = -1 if match[2] == 'S' else  1
    sign_lon =  1 if match[4] == 'E' else -1

    lat = parse_int(match[1])
    lon = parse_int(match[3])

    return Coord(sign_lat * lat, sign_lon * lon)

class Station():
    __slots__ = 'wban', 'call', 'name', 'coord', 'site_elevation', 'tower_height',

    wban:           int
    call:           str
    name:           str
    coord:          Coord
    site_elevation: float
    tower_height:   float

    @staticmethod
    def from_tsv_row(row: list):
        station = Station()
        station.wban           = int(row[0]) if row[0] != 'PENDING' else None
        station.call           = row[1]
        station.name           = row[2]
        station.coord          = parse(row[3])
        station.site_elevation = 0.3048 * float(row[4])
        station.tower_height   = float(row[5])

        return station

    @staticmethod
    def each_from_tsv(file: str):
        with open(file) as fh:
            reader = csv.reader(fh, delimiter='\t')

            for row in reader:
                for i in range(0, len(row)):
                    row[i] = row[i].rstrip()

                yield Station.from_tsv_row(row)

    def add_to_db(self, db):
        sql = """
            insert into nexrad_station (
                wban, call, name, site_elevation, tower_height, coord
            ) values (
                ?, ?, ?, ?, ?, MakePoint(?, ?, 4326)
            )
        """

        db.execute(sql, (
            self.wban, self.call, self.name,
            self.site_elevation, self.tower_height,
            self.coord.lon, self.coord.lat
        ))