def run(self):
     self.output().makedirs()
     try:
         self.download();
     except subprocess.CalledProcessError:
         shell('rm -f {target}'.format(target=self.output().path))
         raise

 def __init__(self, **kwargs):
     if kwargs.get('force'):
         try:
             shell('aws s3 rm s3://data-observatory/observatory.pdf')
         except:
             pass
     super(PDFCatalogToS3, self).__init__()

 def input_shp(self):
     cmd = 'ls {input}/scince/shps/'.format(
         input=self.input().path
     )
     # handle differeing file naming conventions between the geographies
     # and the census data
     if self.resolution == 'municipio':
         resolution = 'municipal'
     elif self.resolution == 'entidad':
         resolution = 'estatal'
     elif self.resolution == 'localidad_urbana_y_rural_amanzanada':
         resolution = 'loc_urb'
     else:
         resolution = self.resolution
     for ent in shell(cmd).strip().split('\n'):
         if ent.lower() == 'national':
             continue
         if self.table.lower().startswith('pob'):
             path = 'ls {input}/scince/shps/{ent}/{ent}_{resolution}*.dbf'
         else:
             path = 'ls {{input}}/scince/shps/{{ent}}/tablas/' \
                 '{{ent}}_cpv2010_{{resolution}}*_{table}.dbf'.format(
                     table=DEMOGRAPHIC_TABLES[self.table])
         cmd = path.format(
             input=self.input().path,
             ent=ent,
             resolution=resolution,
         )
         for shp in shell(cmd).strip().split('\n'):
             yield shp

 def populate(self):
     for infile in self.input()['data']:
         # gunzip each CSV into the table
         cmd = r"gunzip -c '{input}' | psql -c '\copy {tablename} FROM STDIN " \
               r"WITH CSV HEADER'".format(input=infile.path,
                                          tablename=self.output().table)
         print cmd
         shell(cmd)

def test_download_unzip_task():
    '''
    Download unzip task should download remote assets and unzip them locally.
    '''
    task = TestDownloadUnzipTask()
    if task.output().exists():
        shell('rm -r {}'.format(task.output().path))
    assert_false(task.output().exists())
    runtask(task)
    assert_true(task.output().exists())

 def rename_files(self):
     rename = {
         'ustractclustersnew-41530.bin' : 'US_tract_clusters_new.dbf',
         'ustractclustersnew-41538.txt' : 'US_tract_clusters_new.prj',
         'ustractclustersnew-41563.bin' : 'US_tract_clusters_new.shp',
         'ustractclustersnew-41555.bin' : 'US_tract_clusters_new.shx'
     }
     for old_name, new_name in rename.iteritems():
         shell('mv {folder}/{old_name} {folder}/{new_name}'.format(
             old_name=old_name,
             new_name=new_name,
             folder=self.decompressed_folder()))

 def run(self):
     session = current_session()
     try:
         self.output().makedirs()
         session.execute(
             'INSERT INTO observatory.obs_dump_version (dump_id) '
             "VALUES ('{task_id}')".format(task_id=self.task_id))
         session.commit()
         shell('pg_dump -Fc -Z0 -x -n observatory -f {output}'.format(
             output=self.output().path))
     except Exception as err:
         session.rollback()
         raise err

 def output(self):
     path = self.input().path.replace('tmp/carto/Dump_', 'do-release-')
     path = path.replace('.dump', '/obs.dump')
     path = 's3://cartodb-observatory-data/{path}'.format(
         path=path
     )
     print path
     target = S3Target(path)
     if self.force:
         shell('aws s3 rm {output}'.format(
             output=path
         ))
         self.force = False
     return target

 def run(self):
     infiles = shell("ls {input}/LC*DATA.CSV".format(input=self.input().path))
     fhandle = self.output().open("w")
     for infile in infiles.strip().split("\n"):
         table = os.path.split(infile)[-1].split("DATA.CSV")[0]
         data = yield ImportEnglandWalesLocal(table=table)
         fhandle.write("{table}\n".format(table=data.table))
     fhandle.close()

 def run(self):
     infiles = shell('ls {input}/{survey_code}-{geo_code}*.[cC][sS][vV]'.format(
         input=self.input().path,
         survey_code=SURVEY_CODES[self.survey],
         geo_code=GEOGRAPHY_CODES[self.resolution]
     ))
     in_csv_files = infiles.strip().split('\n')
     os.makedirs(self.output().path)
     StatCanParser().parse_csv_to_files(in_csv_files, self.output().path)

    def run(self):
        # make the table
        session = current_session()
        session.execute('''
DROP TABLE IF EXISTS {tablename};
CREATE TABLE {tablename} (
    {columns}
);
                       '''.format(tablename=self.output().table,
                                  columns=self.columns()))
        session.commit()

        #cursor.connection.commit()

        for infile in self.input():
            print infile.path
            # gunzip each CSV into the table
            cmd = r"gunzip -c '{input}' | psql -c '\copy {tablename} FROM STDIN " \
                  r"WITH CSV HEADER'".format(input=infile.path, tablename=self.output().table)
            shell(cmd)

    def populate(self):
        table_name = self.output().table
        shell(r"psql -c '\copy {table} FROM {file_path} WITH CSV HEADER'".format(
            table=table_name,
            file_path=self.input()['data_file'].path
        ))
        for name, segment_id in SpielmanSingletonColumns.x10_mapping.iteritems():
            current_session().execute("update {table} set X10 = '{name}' "
                                      "where X10 ='{segment_id}'; ".format(
                                          table=table_name,
                                          name=name,
                                          segment_id=segment_id
                                      ))

        for name, segment_id in SpielmanSingletonColumns.x55_mapping.iteritems():
            current_session().execute("update {table} set X55 = '{name}' "
                                      "where X55 ='{segment_id}'; ".format(
                                          table=table_name,
                                          name=name,
                                          segment_id=segment_id
                                      ))

    def run(self):
        shapefiles = shell('ls {dir}/*.shp'.format(
            dir=os.path.join('tmp', classpath(self), str(self.year), self.geography)
        )).strip().split('\n')

        cmd = 'PG_USE_COPY=yes PGCLIENTENCODING=latin1 ' \
                'ogr2ogr -f PostgreSQL "PG:dbname=$PGDATABASE active_schema={schema}" ' \
                '-t_srs "EPSG:4326" -nlt MultiPolygon -nln {tablename} ' \
                '-lco OVERWRITE=yes ' \
                '-lco SCHEMA={schema} {shpfile_path} '.format(
                    tablename=self.output().tablename,
                    schema=self.output().schema,
                    shpfile_path=shapefiles.pop())
        shell(cmd)

        # chunk into 500 shapefiles at a time.
        for i, shape_group in enumerate(grouper(shapefiles, 500)):
            shell(
                'export PG_USE_COPY=yes PGCLIENTENCODING=latin1; '
                'echo \'{shapefiles}\' | xargs -P 16 -I shpfile_path '
                'ogr2ogr -f PostgreSQL "PG:dbname=$PGDATABASE '
                'active_schema={schema}" -append '
                '-t_srs "EPSG:4326" -nlt MultiPolygon -nln {tablename} '
                'shpfile_path '.format(
                    shapefiles='\n'.join([shp for shp in shape_group if shp]),
                    tablename=self.output().tablename,
                    schema=self.output().schema))
            print 'imported {} shapefiles'.format((i + 1) * 500)

        session = current_session()
        # Spatial index
        session.execute('ALTER TABLE {qualified_table} RENAME COLUMN '
                        'wkb_geometry TO geom'.format(
                            qualified_table=self.output().table))
        session.execute('CREATE INDEX ON {qualified_table} USING GIST (geom)'.format(
            qualified_table=self.output().table))

    def run(self):
        resp = requests.get(self.URL.format(resolution=self.resolution))
        encoded = resp.text.encode(resp.headers['Content-Type'].split('charset=')[1])
        reader = DictReader(encoded.split('\r\n'))

        for i, line in enumerate(reader):
            # TODO would be much, much faster in parallel...
            url = 'https://whosonfirst.mapzen.com/data/{path}'.format(path=line['path'])
            lfs_url = 'https://github.com/whosonfirst/whosonfirst-data/raw/master/data/{path}'.format(
                path=line['path'])
            cmd = 'wget \'{url}\' -O - | ogr2ogr -{operation} ' \
                    '-nlt MULTIPOLYGON -nln \'{table}\' ' \
                    '-f PostgreSQL PG:"dbname=$PGDATABASE ' \
                    'active_schema={schema}" /vsistdin/'.format(
                        url=url,
                        schema=self.output().schema,
                        table=self.output().tablename,
                        operation='append' if i > 0 else 'overwrite -lco OVERWRITE=yes'
                    )
            try:
                shell(cmd)
            except subprocess.CalledProcessError:
                cmd = cmd.replace(url, lfs_url)
                shell(cmd)

 def complete(self):
     try:
         exists = shell('ls {}'.format(os.path.join(self.directory, '*.shp')))
         return exists != ''
     except subprocess.CalledProcessError:
         return False

 def output(self):
     shps = shell('ls {}'.format(os.path.join(self.directory, '*.shp')))
     for path in shps:
         yield LocalTarget(path)

 def download(self):
     shell('wget --header=\'Cookie: auth_tkt="96a4778a0e3366127d4a47cf19a9c7d65751e5a9talos!userid_type:unicode"; auth_tkt="96a4778a0e3366127d4a47cf19a9c7d65751e5a9talos!userid_type:unicode";\' -O {output}.zip {url}'.format(
         output=self.output().path,
         url=self.URL))

 def run(self):
     self.output().makedirs()
     shell('wget \'{url}\' -O {output}'.format(url=self.url(),
                                               output=self.output().path))

'''
Test ACS columns
'''

from tasks.util import shell

# TODO clean this up in a more general init script

try:
    shell('createdb test')
except:
    pass

from nose.tools import assert_equals, with_setup, assert_false, assert_true

from tasks.meta import (OBSColumnTable, OBSColumn, OBSColumnToColumn, OBSTable,
                        OBSTag, OBSColumnTag, Base)

from tasks.us.census.lodes import WorkplaceAreaCharacteristicsColumns

from tests.util import runtask, setup, teardown


@with_setup(setup, teardown)
def test_wac_columns_run():
    runtask(WorkplaceAreaCharacteristicsColumns())