<commit_before>import requests
import os
import sys
import tarfile
import zipfile
from cStringIO import StringIO
from csvkit.unicsv import UnicodeCSVReader, UnicodeCSVWriter, \
    UnicodeCSVDictReader
from dateutil import parser
from datetime import datetime, date, timedelta
import calendar
from plenario.database import task_session as session, task_engine as engine, \
    Base
from sqlalchemy import Table, Column, String, Date, DateTime, Integer, Float, \
    VARCHAR, BigInteger, UniqueConstraint, and_, select
from sqlalchemy.dialects.postgresql import ARRAY
from geoalchemy2 import Geometry
from uuid import uuid4
DATA_DIR = os.environ['WOPR_DATA_DIR']

class WeatherError(Exception):
    def __init__(self, message):
        Exception.__init__(self, message)
        self.message = message

class WeatherETL(object):
    """ 
    Download, transform and insert weather data into plenario
    """
    weather_type_dict = {'+FC': 'TORNADO/WATERSPOUT',
                     'FC': 'FUNNEL CLOUD',
                     'TS': 'THUNDERSTORM',
                     'GR': 'HAIL',
                     'RA': 'RAIN',
                     'DZ': 'DRIZZLE',
                     'SN': 'SNOW',
                     'SG': 'SNOW GRAINS',
                     'GS': 'SMALL HAIL &/OR SNOW PELLETS',
                     'PL': 'ICE PELLETS',
                     'IC': 'ICE CRYSTALS',
                     'FG': 'FOG', # 'FG+': 'HEAVY FOG (FG & LE.25 MILES VISIBILITY)',
                     'BR': 'MIST',
                     'UP': 'UNKNOWN PRECIPITATION',
                     'HZ': 'HAZE',
                     'FU': 'SMOKE',
                     'VA': 'VOLCANIC ASH',
                     'DU': 'WIDESPREAD DUST',
                     'DS': 'DUSTSTORM',
                     'PO': 'SAND/DUST WHIRLS',
                     'SA': 'SAND',
                     'SS': 'SANDSTORM',
                     'PY': 'SPRAY',
                     'SQ': 'SQUALL',
                     'DR': 'LOW DRIFTING',
                     'SH': 'SHOWER',
                     'FZ': 'FREEZING',
                     'MI': 'SHALLOW',
                     'PR': 'PARTIAL',
                     'BC': 'PATCHES',
                     'BL': 'BLOWING',
                     'VC': 'VICINITY'
                     # Prefixes:
                     # - LIGHT
                     # + HEAVY
                     # "NO SIGN" MODERATE
                 }

    def __init__(self, data_dir=DATA_DIR, debug=False):
        self.base_url = 'http://cdo.ncdc.noaa.gov/qclcd_ascii'
        self.data_dir = data_dir
        self.debug_outfile = sys.stdout
        self.debug = debug
        if (self.debug == True):
            self.debug_outfile = open(os.path.join(self.data_dir, 'weather_etl_debug_out.txt'), 'w')

    # WeatherETL.initialize_last(): for debugging purposes, only initialize the most recent month of weather data.
    def initialize_last(self, start_line=0, end_line=None):
        self.make_tables()
        fname = self._extract_last_fname()
        raw_hourly, raw_daily, file_type = self._extract(fname)
        t_daily = self._transform_daily(raw_daily, file_type, start_line=start_line, end_line=end_line)
        self._load_daily(t_daily)
        t_hourly = self._transform_hourly(raw_hourly, file_type, start_line=start_line, end_line=end_line)
        self._load_hourly(t_hourly)

    def initialize(self): 
        self.make_tables()
        fnames = self._extract_fnames()
        for fname in fnames:
            if (self.debug==True):
                print "INITIALIZE: doing fname", fname
            self._do_etl(fname)

    def initialize_month(self, year, month, no_daily=False, no_hourly=False):
        self.make_tables()
        fname = self._extract_fname(year,month)
        self._do_etl(fname, no_daily, no_hourly)
        
    def _do_etl(self, fname, no_daily=False, no_hourly=False):
        raw_hourly, raw_daily, file_type = self._extract(fname)
        if (not no_daily):
            t_daily = self._transform_daily(raw_daily, file_type)
        if (not no_hourly):
            t_hourly = self._transform_hourly(raw_hourly, file_type)             # this returns a StringIO with all the transformed data
        if (not no_daily):
            self._load_daily(t_daily)                          # this actually imports the transformed StringIO csv
            self._update_daily()
        if (not no_hourly):
            self._load_hourly(t_hourly)    # this actually imports the transformed StringIO csv

    def _update_hourly(self):
        self.new_hourly_table = self._get_hourly_table(name='new')
        self.new_hourly_table.drop(engine, checkfirst=True)
        self.new_hourly_table.create(engine)
    
    def _update_daily(self):
        self.new_daily_table = Table('new_weather_observations_daily', Base.metadata,
                                    Column('wban_code', String(5)),
                                    Column('date', Date), extend_existing=True)
        self.new_daily_table.drop(engine, checkfirst=True)
        self.new_daily_table.create(engine)
        ins = self.new_daily_table.insert()\
                .from_select(['wban_code', 'date'], 
                    select([self.src_daily_table.c.wban_code, self.src_daily_table.c.date])\
                        .select_from(self.src_daily_table.join(self.daily_table,
                            and_(self.src_daily_table.c.wban_code == self.daily_table.c.wban_code,
                                 self.src_daily_table.c.date == self.daily_table.c.date),
                            isouter=True)
                    ).where(self.daily_table.c.id == None)
                )
        conn = engine.contextual_connect()
        try:
            conn.execute(ins)
            new = True
        except TypeError:
            new = False
        if new:
            ins = self.daily_table.insert()\
                    .from_select([c for c in self.daily_table.columns if c.name != 'id'], 
                        select([c for c in self.src_daily_table.columns])\
                            .select_from(self.src_daily_table.join(self.new_daily_table,
                                and_(self.src_daily_table.c.wban_code == self.new_daily_table.c.wban_code,
                                     self.src_daily_table.c.date == self.new_daily_table.c.date)
                            ))
                    )
            conn.execute(ins)

    def _transform_hourly(self, raw, file_type, start_line=0, end_line=None):
        t = getattr(self, '_transform_%s_hourly' % file_type)(raw, start_line, end_line)
        return t

    def _transform_daily(self, raw, file_type, start_line=0, end_line=None):<commit_msg>Update process seems to be working<commit_after>import requests
import os
import sys
import tarfile
import zipfile
from cStringIO import StringIO
from csvkit.unicsv import UnicodeCSVReader, UnicodeCSVWriter, \
    UnicodeCSVDictReader
from dateutil import parser
from datetime import datetime, date, timedelta
import calendar
from plenario.database import task_session as session, task_engine as engine, \
    Base
from sqlalchemy import Table, Column, String, Date, DateTime, Integer, Float, \
    VARCHAR, BigInteger, UniqueConstraint, and_, select
from sqlalchemy.dialects.postgresql import ARRAY
from geoalchemy2 import Geometry
from uuid import uuid4
DATA_DIR = os.environ['WOPR_DATA_DIR']

class WeatherError(Exception):
    def __init__(self, message):
        Exception.__init__(self, message)
        self.message = message

class WeatherETL(object):
    """ 
    Download, transform and insert weather data into plenario
    """
    weather_type_dict = {'+FC': 'TORNADO/WATERSPOUT',
                     'FC': 'FUNNEL CLOUD',
                     'TS': 'THUNDERSTORM',
                     'GR': 'HAIL',
                     'RA': 'RAIN',
                     'DZ': 'DRIZZLE',
                     'SN': 'SNOW',
                     'SG': 'SNOW GRAINS',
                     'GS': 'SMALL HAIL &/OR SNOW PELLETS',
                     'PL': 'ICE PELLETS',
                     'IC': 'ICE CRYSTALS',
                     'FG': 'FOG', # 'FG+': 'HEAVY FOG (FG & LE.25 MILES VISIBILITY)',
                     'BR': 'MIST',
                     'UP': 'UNKNOWN PRECIPITATION',
                     'HZ': 'HAZE',
                     'FU': 'SMOKE',
                     'VA': 'VOLCANIC ASH',
                     'DU': 'WIDESPREAD DUST',
                     'DS': 'DUSTSTORM',
                     'PO': 'SAND/DUST WHIRLS',
                     'SA': 'SAND',
                     'SS': 'SANDSTORM',
                     'PY': 'SPRAY',
                     'SQ': 'SQUALL',
                     'DR': 'LOW DRIFTING',
                     'SH': 'SHOWER',
                     'FZ': 'FREEZING',
                     'MI': 'SHALLOW',
                     'PR': 'PARTIAL',
                     'BC': 'PATCHES',
                     'BL': 'BLOWING',
                     'VC': 'VICINITY'
                     # Prefixes:
                     # - LIGHT
                     # + HEAVY
                     # "NO SIGN" MODERATE
                 }

    def __init__(self, data_dir=DATA_DIR, debug=False):
        self.base_url = 'http://cdo.ncdc.noaa.gov/qclcd_ascii'
        self.data_dir = data_dir
        self.debug_outfile = sys.stdout
        self.debug = debug
        if (self.debug == True):
            self.debug_outfile = open(os.path.join(self.data_dir, 'weather_etl_debug_out.txt'), 'w')

    # WeatherETL.initialize_last(): for debugging purposes, only initialize the most recent month of weather data.
    def initialize_last(self, start_line=0, end_line=None):
        self.make_tables()
        fname = self._extract_last_fname()
        raw_hourly, raw_daily, file_type = self._extract(fname)
        t_daily = self._transform_daily(raw_daily, file_type, start_line=start_line, end_line=end_line)
        self._load_daily(t_daily)
        t_hourly = self._transform_hourly(raw_hourly, file_type, start_line=start_line, end_line=end_line)
        self._load_hourly(t_hourly)

    def initialize(self): 
        self.make_tables()
        fnames = self._extract_fnames()
        for fname in fnames:
            if (self.debug==True):
                print "INITIALIZE: doing fname", fname
            self._do_etl(fname)

    def initialize_month(self, year, month, no_daily=False, no_hourly=False):
        self.make_tables()
        fname = self._extract_fname(year,month)
        self._do_etl(fname, no_daily, no_hourly)
        
    def _do_etl(self, fname, no_daily=False, no_hourly=False):
        raw_hourly, raw_daily, file_type = self._extract(fname)
        if (not no_daily):
            t_daily = self._transform_daily(raw_daily, file_type)
        if (not no_hourly):
            t_hourly = self._transform_hourly(raw_hourly, file_type)             # this returns a StringIO with all the transformed data
        if (not no_daily):
            self._load_daily(t_daily)                          # this actually imports the transformed StringIO csv
            self._update(span='daily')
        if (not no_hourly):
            self._load_hourly(t_hourly)    # this actually imports the transformed StringIO csv
            self._update(span='hourly')
        self._cleanup_temp_tables()

    def _cleanup_temp_tables(self):
        for span in ['daily', 'hourly']:
            for tname in ['src', 'new']:
                try:
                    table = getattr(self, '%s_%s_table' % (tname, span))
                    table.drop(engine, checkfirst=True)
                except AttributeError:
                    continue

    def _update(self, span=None):
        new_table = Table('new_weather_observations_%s' % span, Base.metadata,
                          Column('wban_code', String(5)), extend_existing=True)
        dat_table = getattr(self, '%s_table' % span)
        src_table = getattr(self, 'src_%s_table' % span)
        from_sel_cols = ['wban_code']
        if span == 'daily':
            from_sel_cols.append('date')
            src_date_col = src_table.c.date
            dat_date_col = dat_table.c.date
            new_table.append_column(Column('date', Date))
            new_date_col = new_table.c.date
        elif span == 'hourly':
            from_sel_cols.append('datetime')
            src_date_col = src_table.c.datetime
            dat_date_col = dat_table.c.datetime
            new_table.append_column(Column('datetime', DateTime))
            new_date_col = new_table.c.datetime
        new_table.drop(engine, checkfirst=True)
        new_table.create(engine)
        ins = new_table.insert()\
                .from_select(from_sel_cols, 
                    select([src_table.c.wban_code, src_date_col])\
                        .select_from(src_table.join(dat_table,
                            and_(src_table.c.wban_code == dat_table.c.wban_code,
                                 src_date_col == dat_date_col),
                            isouter=True)
                    ).where(dat_table.c.id == None)
                )
        conn = engine.contextual_connect()
        try:
            conn.execute(ins)
            new = True
        except TypeError:
            new = False
        if new:
            ins = dat_table.insert()\
                    .from_select([c for c in dat_table.columns if c.name != 'id'], 
                        select([c for c in src_table.columns])\
                            .select_from(src_table.join(new_table,
                                and_(src_table.c.wban_code == new_table.c.wban_code,
                                     src_date_col == new_date_col)
                            ))
                    )
            conn.execute(ins)

    def _transform_hourly(self, raw, file_type, start_line=0, end_line=None):
        t = getattr(self, '_transform_%s_hourly' % file_type)(raw, start_line, end_line)
        return t

    def _transform_daily(self, raw, file_type, start_line=0, end_line=None):
{'fraction_non_alphanumeric': 0.10340884728078324, 'fraction_numerical': 0.0011434288572857857, 'mean_word_length': 1.7628825271470878, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6927685', 'n_tokens_mistral': 4368, 'n_tokens_neox': 4113, 'n_words': 865}