<commit_before>"""Suck in MADIS data into the iemdb"""
import netCDF4
import datetime
import os
import sys
from pyiem.observation import Observation
from pyiem.datatypes import temperature, distance, speed
import subprocess
import pytz
import psycopg2.extras
pgconn = psycopg2.connect(database='iem', host='iemdb')
icursor = pgconn.cursor(cursor_factory=psycopg2.extras.DictCursor)

fn = None
for i in range(0, 4):
    ts = datetime.datetime.utcnow() - datetime.timedelta(hours=i)
    testfn = ts.strftime("/mesonet/data/madis/mesonet1/%Y%m%d_%H00.nc")
    if os.path.isfile(testfn):
        fn = testfn
        break

if fn is None:
    sys.exit()

try:
    nc = netCDF4.Dataset(fn)
except:
    # File may be in progress of being read, wait a little bit
    import time
    time.sleep(20)
    nc = netCDF4.Dataset(fn)


def sanityCheck(val, lower, upper, rt):
    if val > lower and val < upper:
        return float(val)
    return rt

stations = nc.variables["stationId"][:]
providers = nc.variables["dataProvider"][:]
names = nc.variables["stationName"][:]
tmpk = nc.variables["temperature"][:]
tmpk_dd = nc.variables["temperatureDD"][:]
obTime = nc.variables["observationTime"][:]
pressure = nc.variables["stationPressure"][:]
altimeter = nc.variables["altimeter"][:]
slp = nc.variables["seaLevelPressure"][:]
dwpk = nc.variables["dewpoint"][:]
drct = nc.variables["windDir"][:]
smps = nc.variables["windSpeed"][:]
gmps = nc.variables["windGust"][:]
gmps_drct = nc.variables["windDirMax"][:]
pcpn = nc.variables["precipAccum"][:]
rtk1 = nc.variables["roadTemperature1"][:]
rtk2 = nc.variables["roadTemperature2"][:]
rtk3 = nc.variables["roadTemperature3"][:]
rtk4 = nc.variables["roadTemperature4"][:]
subk1 = nc.variables["roadSubsurfaceTemp1"][:]

db = {}

MY_PROVIDERS = ["KYTC-RWIS",
                "KYMN",
                "NEDOR",
                "MesoWest"]


def provider2network(p):
    """ Convert a MADIS network ID to one that I use, here in IEM land"""
    if p in ['KYMN']:
        return p
    if p == 'MesoWest':
        return 'VTWAC'
    if len(p) == 5 or p in ['KYTC-RWIS', 'NEDOR']:
        if p[:2] == 'IA':
            return None
        return '%s_RWIS' % (p[:2],)
    print("Unsure how to convert %s into a network" % (p,))
    return None

for recnum in range(len(providers)):
    thisProvider = providers[recnum].tostring().replace('\x00', '')
    thisStation = stations[recnum].tostring().replace('\x00', '')
    if not thisProvider.endswith('DOT') and thisProvider not in MY_PROVIDERS:
        continue
    name = names[recnum].tostring().replace("'",
                                            "").replace('\x00',
                                                        '').replace('\xa0',
                                                                    ' '
                                                                    ).strip()
    if name == '':
        continue
    if thisProvider == 'MesoWest':
        # get the network from the last portion of the name
        network = name.split()[-1]
        if network != 'VTWAC':
            continue
    else:
        network = provider2network(thisProvider)
    if network is None:
        continue
    db[thisStation] = {}
    ticks = obTime[recnum]
    ts = datetime.datetime(1970, 1, 1) + datetime.timedelta(seconds=ticks)
    db[thisStation]['ts'] = ts.replace(tzinfo=pytz.timezone('UTC'))
    db[thisStation]['network'] = network
    db[thisStation]['pres'] = sanityCheck(pressure[recnum], 0, 1000000, -99)
    db[thisStation]['tmpk'] = sanityCheck(tmpk[recnum], 0, 500, -99)
    db[thisStation]['dwpk'] = sanityCheck(dwpk[recnum], 0, 500, -99)
    db[thisStation]['tmpk_dd'] = tmpk_dd[recnum]
    db[thisStation]['drct'] = sanityCheck(drct[recnum], -1, 361, -99)
    db[thisStation]['smps'] = sanityCheck(smps[recnum], -1, 200, -99)
    db[thisStation]['gmps'] = sanityCheck(gmps[recnum], -1, 200, -99)
    db[thisStation]['rtk1'] = sanityCheck(rtk1[recnum], 0, 500, -99)
    db[thisStation]['rtk2'] = sanityCheck(rtk2[recnum], 0, 500, -99)
    db[thisStation]['rtk3'] = sanityCheck(rtk3[recnum], 0, 500, -99)
    db[thisStation]['rtk4'] = sanityCheck(rtk4[recnum], 0, 500, -99)
    db[thisStation]['subk'] = sanityCheck(subk1[recnum], 0, 500, -99)
    db[thisStation]['pday'] = sanityCheck(pcpn[recnum], -1, 5000, -99)

for sid in db.keys():
    iem = Observation(sid, db[sid]['network'], db[sid]['ts'])
    # if not iem.load(icursor):
    #    print 'Missing fp: %s network: %s station: %s' % (fp,
    #                                                      db[sid]['network'],
    #                                                      sid)
    #    subprocess.call("python sync_stations.py %s" % (fp,), shell=True)
    #    os.chdir("../../dbutil")
    #    subprocess.call("sh SYNC_STATIONS.sh", shell=True)
    #    os.chdir("../ingestors/madis")
    iem.data['tmpf'] = temperature(db[sid]['tmpk'], 'K').value('F')
    iem.data['dwpf'] = temperature(db[sid]['dwpk'], 'K').value('F')
    if db[sid]['drct'] >= 0:
        iem.data['drct'] = db[sid]['drct']
    if db[sid]['smps'] >= 0:
        iem.data['sknt'] = speed(db[sid]['smps'], 'MPS').value('KT')
    if db[sid]['gmps'] >= 0:
        iem.data['gust'] = speed(db[sid]['gmps'], 'MPS').value('KT')
    if db[sid]['pres'] > 0:
        iem.data['pres'] = (float(db[sid]['pres']) / 100.00) * 0.02952
    if db[sid]['rtk1'] > 0:
        iem.data['tsf0'] = temperature(db[sid]['rtk1'], 'K').value('F')
    if db[sid]['rtk2'] > 0:
        iem.data['tsf1'] = temperature(db[sid]['rtk2'], 'K').value('F')
    if db[sid]['rtk3'] > 0:
        iem.data['tsf2'] = temperature(db[sid]['rtk3'], 'K').value('F')
    if db[sid]['rtk4'] > 0:
        iem.data['tsf3'] = temperature(db[sid]['rtk4'], 'K').value('F')
    if db[sid]['subk'] > 0:
        iem.data['rwis_subf'] = temperature(db[sid]['subk'], 'K').value('F')
    if db[sid]['pday'] >= 0:
        iem.data['pday'] = round(distance(db[sid]['pday'], 'MM').value("IN"),
                                 2)
    if not iem.save(icursor):
        print(("MADIS Extract: %s found new station: %s network: %s"
              "") % (fn.split("/")[-1], sid, db[sid]['network']))
        subprocess.call("python sync_stations.py %s" % (fn,), shell=True)
        os.chdir("../../dbutil")
        subprocess.call("sh SYNC_STATIONS.sh", shell=True)
        os.chdir("../ingestors/madis")
        print("...done with sync.")
    del(iem)
nc.close()
icursor.close()
pgconn.commit()
pgconn.close()<commit_msg>review MADIS ingest as per #107<commit_after>"""Suck in MADIS data into the iemdb"""
from __future__ import print_function
import time
import datetime
import os
import sys
import subprocess

import netCDF4
from pyiem.observation import Observation
from pyiem.datatypes import temperature, distance, speed
import pytz
import psycopg2.extras

MY_PROVIDERS = ["KYTC-RWIS",
                "KYMN",
                "NEDOR",
                "MesoWest"]


def find_file():
    """Find the most recent file"""
    fn = None
    for i in range(0, 4):
        ts = datetime.datetime.utcnow() - datetime.timedelta(hours=i)
        testfn = ts.strftime("/mesonet/data/madis/mesonet1/%Y%m%d_%H00.nc")
        if os.path.isfile(testfn):
            fn = testfn
            break

    if fn is None:
        sys.exit()
    return fn


def sanity_check(val, lower, upper, defaultval):
    """Simple bounds check"""
    if val > lower and val < upper:
        return float(val)
    return defaultval


def provider2network(provider):
    """ Convert a MADIS network ID to one that I use, here in IEM land"""
    if provider in ['KYMN']:
        return provider
    if provider == 'MesoWest':
        return 'VTWAC'
    if len(provider) == 5 or provider in ['KYTC-RWIS', 'NEDOR']:
        if provider[:2] == 'IA':
            return None
        return '%s_RWIS' % (provider[:2],)
    print("Unsure how to convert %s into a network" % (provider,))
    return None


def main():
    """Do Something"""
    pgconn = psycopg2.connect(database='iem', host='iemdb')
    icursor = pgconn.cursor(cursor_factory=psycopg2.extras.DictCursor)
    fn = find_file()
    try:
        nc = netCDF4.Dataset(fn)
    except Exception as _:
        # File may be in progress of being read, wait a little bit
        time.sleep(20)
        nc = netCDF4.Dataset(fn)

    stations = nc.variables["stationId"][:]
    providers = nc.variables["dataProvider"][:]
    names = nc.variables["stationName"][:]
    tmpk = nc.variables["temperature"][:]
    tmpk_dd = nc.variables["temperatureDD"][:]
    obtime = nc.variables["observationTime"][:]
    pressure = nc.variables["stationPressure"][:]
    # altimeter = nc.variables["altimeter"][:]
    # slp = nc.variables["seaLevelPressure"][:]
    dwpk = nc.variables["dewpoint"][:]
    drct = nc.variables["windDir"][:]
    smps = nc.variables["windSpeed"][:]
    gmps = nc.variables["windGust"][:]
    # gmps_drct = nc.variables["windDirMax"][:]
    pcpn = nc.variables["precipAccum"][:]
    rtk1 = nc.variables["roadTemperature1"][:]
    rtk2 = nc.variables["roadTemperature2"][:]
    rtk3 = nc.variables["roadTemperature3"][:]
    rtk4 = nc.variables["roadTemperature4"][:]
    subk1 = nc.variables["roadSubsurfaceTemp1"][:]

    db = {}

    for recnum, provider in enumerate(providers):
        this_provider = provider.tostring().replace('\x00', '')
        this_station = stations[recnum].tostring().replace('\x00', '')
        if (not this_provider.endswith('DOT') and
                this_provider not in MY_PROVIDERS):
            continue
        name = names[recnum].tostring(
            ).replace("'",  "").replace('\x00', '').replace('\xa0',
                                                            ' ').strip()
        if name == '':
            continue
        if this_provider == 'MesoWest':
            # get the network from the last portion of the name
            network = name.split()[-1]
            if network != 'VTWAC':
                continue
        else:
            network = provider2network(this_provider)
        if network is None:
            continue
        db[this_station] = {}
        ticks = obtime[recnum]
        ts = datetime.datetime(1970, 1, 1) + datetime.timedelta(seconds=ticks)
        db[this_station]['ts'] = ts.replace(tzinfo=pytz.timezone('UTC'))
        db[this_station]['network'] = network
        db[this_station]['pres'] = sanity_check(pressure[recnum], 0, 1000000,
                                                -99)
        db[this_station]['tmpk'] = sanity_check(tmpk[recnum], 0, 500, -99)
        db[this_station]['dwpk'] = sanity_check(dwpk[recnum], 0, 500, -99)
        db[this_station]['tmpk_dd'] = tmpk_dd[recnum]
        db[this_station]['drct'] = sanity_check(drct[recnum], -1, 361, -99)
        db[this_station]['smps'] = sanity_check(smps[recnum], -1, 200, -99)
        db[this_station]['gmps'] = sanity_check(gmps[recnum], -1, 200, -99)
        db[this_station]['rtk1'] = sanity_check(rtk1[recnum], 0, 500, -99)
        db[this_station]['rtk2'] = sanity_check(rtk2[recnum], 0, 500, -99)
        db[this_station]['rtk3'] = sanity_check(rtk3[recnum], 0, 500, -99)
        db[this_station]['rtk4'] = sanity_check(rtk4[recnum], 0, 500, -99)
        db[this_station]['subk'] = sanity_check(subk1[recnum], 0, 500, -99)
        db[this_station]['pday'] = sanity_check(pcpn[recnum], -1, 5000, -99)

    for sid in db:
        # print("Processing %s[%s]" % (sid, db[sid]['network']))
        iem = Observation(sid, db[sid]['network'], db[sid]['ts'])
        # if not iem.load(icursor):
        #    print 'Missing fp: %s network: %s station: %s' % (fp,
        #                                                      db[sid]['network'],
        #                                                      sid)
        #    subprocess.call("python sync_stations.py %s" % (fp,), shell=True)
        #    os.chdir("../../dbutil")
        #    subprocess.call("sh SYNC_STATIONS.sh", shell=True)
        #    os.chdir("../ingestors/madis")
        iem.data['tmpf'] = temperature(db[sid]['tmpk'], 'K').value('F')
        iem.data['dwpf'] = temperature(db[sid]['dwpk'], 'K').value('F')
        if db[sid]['drct'] >= 0:
            iem.data['drct'] = db[sid]['drct']
        if db[sid]['smps'] >= 0:
            iem.data['sknt'] = speed(db[sid]['smps'], 'MPS').value('KT')
        if db[sid]['gmps'] >= 0:
            iem.data['gust'] = speed(db[sid]['gmps'], 'MPS').value('KT')
        if db[sid]['pres'] > 0:
            iem.data['pres'] = (float(db[sid]['pres']) / 100.00) * 0.02952
        if db[sid]['rtk1'] > 0:
            iem.data['tsf0'] = temperature(db[sid]['rtk1'], 'K').value('F')
        if db[sid]['rtk2'] > 0:
            iem.data['tsf1'] = temperature(db[sid]['rtk2'], 'K').value('F')
        if db[sid]['rtk3'] > 0:
            iem.data['tsf2'] = temperature(db[sid]['rtk3'], 'K').value('F')
        if db[sid]['rtk4'] > 0:
            iem.data['tsf3'] = temperature(db[sid]['rtk4'], 'K').value('F')
        if db[sid]['subk'] > 0:
            iem.data['rwis_subf'] = temperature(db[sid]['subk'],
                                                'K').value('F')
        if db[sid]['pday'] >= 0:
            iem.data['pday'] = round(distance(db[sid]['pday'],
                                              'MM').value("IN"),
                                     2)
        if not iem.save(icursor):
            print(("MADIS Extract: %s found new station: %s network: %s"
                  "") % (fn.split("/")[-1], sid, db[sid]['network']))
            subprocess.call("python sync_stations.py %s" % (fn,), shell=True)
            os.chdir("../../dbutil")
            subprocess.call("sh SYNC_STATIONS.sh", shell=True)
            os.chdir("../ingestors/madis")
            print("...done with sync.")
        del iem
    nc.close()
    icursor.close()
    pgconn.commit()
    pgconn.close()


if __name__ == '__main__':
    main()
{'fraction_non_alphanumeric': 0.18097431100237463, 'fraction_numerical': 0.023674174282219185, 'mean_word_length': 2.5016376921138828, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 25, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '117394', 'n_tokens_mistral': 4984, 'n_tokens_neox': 4700, 'n_words': 943}