def digest_file(file):
    hash_md5 = hashlib.md5()
    with open(file, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b''):
            hash_md5.update(chunk)
    return hash_md5.hexdigest() def get_logger(name):
    '''
    Obtain a logger outputing to stderr with specified name. Defaults to INFO
    log level.
    '''
    logger = logging.getLogger(name)
    if LUIGI_LOG_ARG in sys.argv:
        index = sys.argv.index(LUIGI_LOG_ARG) + 1
        if len(sys.argv) > index:
            logger.setLevel(sys.argv[index])

    return logger LOGGER = get_logger(__name__)
DENOMINATOR = 'denominator'
UNIVERSE = 'universe'
GEOM_REF = 'geom_ref'
GEOM_NAME = 'geom_name'
class CurrentSession(object):
class UniqueMixin(object):
class Raster(UserDefinedType):
class Geometry(UserDefinedType):
class Linestring(UserDefinedType):
class Point(UserDefinedType):
class OBSColumnTable(Base):
class OBSColumnToColumn(Base):
class OBSTableToTable(Base):
class OBSColumn(Base):
class OBSTable(Base):
class OBSTag(Base):
class OBSColumnTag(Base):
class OBSDumpVersion(Base):
class OBSColumnTableTile(Base):
class OBSColumnTableTileSimple(Base):
class OBSTimespan(UniqueMixin, Base):
class UpdatedMetaTarget(Target):
    def __init__(self, async_conn=False):
    def begin(self):
    def get(self):
    def commit(self):
    def rollback(self):
def current_session():
def natural_sort_key(s, _nsre=re.compile('([0-9]+)')):
async def async_pool(user=None, password=None, host=None, port=None,
                     db=None, max_connections=150, timeout=60):
def get_engine(user=None, password=None, host=None, port=None,
               db=None, readonly=False):
    def receive_begin(conn):
    def connect(dbapi_connection, connection_record):
    def checkout(dbapi_connection, connection_record, connection_proxy):
def catalog_latlng(column_id):
def _unique(session, cls, hashfunc, queryfunc, constructor, arg, kw):
    def unique_hash(cls, *arg, **kw):
    def unique_filter(cls, query, *arg, **kw):
    def as_unique(cls, session, *arg, **kw):
    def get_col_spec(self):
    def bind_expression(self, bindvalue):
    def column_expression(self, col):
    def get_col_spec(self):
    def bind_expression(self, bindvalue):
    def column_expression(self, col):
    def get_col_spec(self):
    def bind_expression(self, bindvalue):
    def column_expression(self, col):
    def get_col_spec(self):
    def bind_expression(self, bindvalue):
    def column_expression(self, col):
def tag_creator(tagtarget, session):
def coltocoltargets_creator(coltarget_or_col, reltype):
def tabletotabletargets_creator(target, reltype):
    def sources(self):
    def index_type(self):
    def children(self):
    def is_cartographic(self):
    def is_interpolation(self):
    def is_geomref(self):
    def has_children(self):
    def has_denominators(self):
    def has_catalog_image(self):
    def denominators(self):
    def unit(self):
    def summable(self):
    def catalog_lonlat(self):
    def geom_timespans(self):
    def source_tags(self):
    def license_tags(self):
    def geom_column(self):
    def geomref_column(self):
    def unique_hash(cls, id):
    def unique_filter(cls, query, id):
def session_commit(task, session=_current_session):
def session_rollback(task, exception):
def fromkeys(d, l):
    def exists(self): class ColumnTarget(Target):
    '''
    '''

    def __init__(self, column, task):
        self._id = column.id
        self._task = task
        self._column = column

    def get(self, session):
        '''
        Return a copy of the underlying OBSColumn in the specified session.
        '''
        with session.no_autoflush:
            return session.query(OBSColumn).get(self._id)

    def update_or_create(self):
        self._column = current_session().merge(self._column)

    def exists(self):
        existing = self.get(current_session())
        new_version = float(self._column.version or 0.0)
        if existing:
            existing_version = float(existing.version or 0.0)
            current_session().expunge(existing)
        else:
            existing_version = 0.0
        if existing and existing_version == new_version:
            return True
        elif existing and existing_version > new_version:
            raise Exception('Metadata version mismatch: cannot run task {task} '
                            '(id "{id}") '
                            'with ETL version ({etl}) older than what is in '
                            'DB ({db})'.format(task=self._task.task_id,
                                               id=self._id,
                                               etl=new_version,
                                               db=existing_version))
        return False class TagTarget(Target):
    '''
    '''

    def __init__(self, tag, task):
        self._id = tag.id
        self._tag = tag
        self._task = task

    _tag_cache = {}

    def get(self, session):
        '''
        Return a copy of the underlying OBSTag in the specified session.
        '''
        if not self._tag_cache.get(self._id, None):
            with session.no_autoflush:
                self._tag_cache[self._id] = session.query(OBSTag).get(self._id)
        return self._tag_cache[self._id]

    def update_or_create(self):
        with current_session().no_autoflush:
            self._tag = current_session().merge(self._tag)

    def exists(self):
        session = current_session()
        existing = self.get(session)
        new_version = self._tag.version or 0.0
        if existing:
            if existing in session:
                session.expunge(existing)
            existing_version = existing.version or 0.0
            if float(existing_version) == float(new_version):
                return True
            if existing_version > new_version:
                raise Exception('Metadata version mismatch: cannot run task {task} '
                                '(id "{id}") '
                                'with ETL version ({etl}) older than what is in '
                                'DB ({db})'.format(task=self._task.task_id,
                                                   id=self._id,
                                                   etl=new_version,
                                                   db=existing_version))
        return False class CartoDBTarget(Target):
    '''
    Target which is a CartoDB table
    '''

    def __init__(self, tablename, carto_url=None, api_key=None):
        self.tablename = tablename
        self.carto_url = carto_url
        self.api_key = api_key

    def __str__(self):
        return self.tablename

    def exists(self):
        resp = query_cartodb(
            'SELECT row_number() over () FROM "{tablename}" LIMIT 1'.format(
                tablename=self.tablename),
            api_key=self.api_key,
            carto_url=self.carto_url)
        if resp.status_code != 200:
            return False
        return resp.json()['total_rows'] > 0

    def remove(self, carto_url=None, api_key=None):
        api_key = api_key or os.environ['CARTODB_API_KEY']

        try:
            while True:
                resp = requests.get('{url}/api/v1/tables/{tablename}?api_key={api_key}'.format(
                    url=carto_url,
                    tablename=self.tablename,
                    api_key=api_key
                ))
                viz_id = resp.json()['id']
                # delete dataset by id DELETE
                # https://observatory.cartodb.com/api/v1/viz/ed483a0b-7842-4610-9f6c-8591273b8e5c
                try:
                    requests.delete('{url}/api/v1/viz/{viz_id}?api_key={api_key}'.format(
                        url=carto_url,
                        viz_id=viz_id,
                        api_key=api_key
                    ), timeout=1)
                except requests.Timeout:
                    pass
        except ValueError:
            pass
        query_cartodb('DROP TABLE IF EXISTS {tablename}'.format(tablename=self.tablename))
        assert not self.exists() class PostgresTarget(Target):
    '''
    PostgresTarget which by default uses command-line specified login.
    '''

    def __init__(self, schema, tablename, non_empty=True, where="1 = 1"):
        self._schema = schema
        self._tablename = tablename
        self._non_empty = non_empty
        self._where = where

    @property
    def table(self):
        return '"{schema}".{tablename}'.format(schema=self._schema,
                                               tablename=self._tablename)

    @property
    def tablename(self):
        return self._tablename

    @property
    def schema(self):
        return self._schema

    @property
    def qualified_tablename(self):
        return '"{}".{}'.format(self.schema, self.tablename)

    def _existenceness(self):
        '''
        Returns 0 if the table does not exist, 1 if it exists but has no
        rows (is empty), and 2 if it exists and has one or more rows.
        '''
        session = current_session()
        sql = '''
                SELECT COUNT(*) FROM information_schema.tables
                WHERE table_schema ILIKE '{schema}'
                  AND table_name ILIKE '{tablename}'
              '''.format(
            schema=self._schema,
            tablename=self._tablename)
        resp = session.execute(sql)
        if int(resp.fetchone()[0]) == 0:
            return 0
        resp = session.execute(
            'SELECT row_number() over () FROM "{schema}".{tablename} WHERE {where} LIMIT 1'.format(
                schema=self._schema, tablename=self._tablename,
                where=self._where))
        if resp.fetchone() is None:
            return 1
        else:
            return 2

    def empty(self):
        '''
        Returns True if the table exists but has no rows in it.
        '''
        return self._existenceness() == 1

    def exists(self):
        '''
        Returns True if the table exists and has at least one row in it.
        '''
        if self._non_empty:
            return self._existenceness() == 2
        else:
            return self._existenceness() >= 1

    def exists_or_empty(self):
        '''
        Returns True if the table exists, even if it is empty.
        '''
        return self._existenceness() >= 1 class TableTarget(Target):

    def __init__(self, schema, name, obs_table, columns, task):
        '''
        columns: should be an ordereddict if you want to specify columns' order
        in the table
        '''
        self._id = '.'.join([schema, name])
        obs_table.id = self._id
        obs_table.tablename = '{prefix}{name}'.format(prefix=OBSERVATORY_PREFIX, name=sha1(
            underscore_slugify(self._id).encode('utf-8')).hexdigest())
        self.table = '{schema}.{table}'.format(schema=OBSERVATORY_SCHEMA, table=obs_table.tablename)
        self.qualified_tablename = '"{schema}".{table}'.format(schema=OBSERVATORY_SCHEMA, table=obs_table.tablename)
        self.obs_table = obs_table
        self._tablename = obs_table.tablename
        self._schema = schema
        self._name = name
        self._obs_dict = obs_table.__dict__.copy()
        self._columns = columns
        self._task = task
        if obs_table.tablename in metadata.tables:
            self._table = metadata.tables[obs_table.tablename]
        else:
            self._table = None

    @property
    def tablename(self):
        return self._tablename

    @property
    def schema(self):
        return 'observatory'

    def sync(self):
        '''
        Whether this data should be synced to carto. Defaults to True.
        '''
        return True

    def exists(self):
        '''
        We always want to run this at least once, because we can always
        regenerate tabular data from scratch.
        '''
        session = current_session()
        existing = self.get(session)
        new_version = float(self.obs_table.version or 0.0)
        if existing:
            existing_version = float(existing.version or 0.0)
            if existing in session:
                session.expunge(existing)
        else:
            existing_version = 0.0
        if existing and existing_version == new_version:
            resp = session.execute(
                'SELECT COUNT(*) FROM information_schema.tables '
                "WHERE table_schema = '{schema}'  "
                "  AND table_name = '{tablename}' ".format(
                    schema='observatory',
                    tablename=existing.tablename))
            if int(resp.fetchone()[0]) == 0:
                return False
            resp = session.execute(
                'SELECT row_number() over () '
                'FROM "{schema}".{tablename} LIMIT 1 '.format(
                    schema='observatory',
                    tablename=existing.tablename))
            return resp.fetchone() is not None
        elif existing and existing_version > new_version:
            raise Exception('Metadata version mismatch: cannot run task {task} '
                            '(id "{id}") '
                            'with ETL version ({etl}) older than what is in '
                            'DB ({db})'.format(task=self._task.task_id,
                                               id=self._id,
                                               etl=new_version,
                                               db=existing_version))
        return False

    def get(self, session):
        '''
        Return a copy of the underlying OBSTable in the specified session.
        '''
        with session.no_autoflush:
            return session.query(OBSTable).get(self._id)

    def update_or_create_table(self):
        session = current_session()

        # create new local data table
        columns = []
        for colname, coltarget in list(self._columns.items()):
            colname = colname.lower()
            col = coltarget.get(session)

            # Column info for sqlalchemy's internal metadata
            if col.type.lower() == 'geometry':
                coltype = Geometry
            elif col.type.lower().startswith('geometry(point'):
                coltype = Point
            elif col.type.lower().startswith('geometry(linestring'):
                coltype = Linestring

            # For enum type, pull keys from extra["categories"]
            elif col.type.lower().startswith('enum'):
                cats = list(col.extra['categories'].keys())
                coltype = types.Enum(*cats, name=col.id + '_enum')
            else:
                coltype = getattr(types, col.type.capitalize())
            columns.append(Column(colname, coltype))

        obs_table = self.get(session) or self.obs_table
        # replace local data table
        if obs_table.id in metadata.tables:
            metadata.tables[obs_table.id].drop()
        self._table = Table(obs_table.tablename, metadata, *columns,
                            extend_existing=True, schema='observatory')
        session.commit()
        self._table.drop(checkfirst=True)
        self._table.create()

    def update_or_create_metadata(self, _testmode=False):
        session = current_session()

        # replace metadata table
        self.obs_table = session.merge(self.obs_table)
        obs_table = self.obs_table

        for i, colname_coltarget in enumerate(self._columns.items()):
            colname, coltarget = colname_coltarget
            colname = colname.lower()
            col = coltarget.get(session)

            if _testmode:
                coltable = OBSColumnTable(colname=colname, table=obs_table,
                                          column=col)
            else:
                # Column info for obs metadata
                coltable = session.query(OBSColumnTable).filter_by(
                    column_id=col.id, table_id=obs_table.id).first()
                if coltable:
                    coltable.colname = colname
                else:
                    # catch the case where a column id has changed
                    coltable = session.query(OBSColumnTable).filter_by(
                        table_id=obs_table.id, colname=colname).first()
                    if coltable:
                        coltable.column = col
                    else:
                        coltable = OBSColumnTable(colname=colname, table=obs_table,
                                                  column=col)

            session.add(coltable) class RepoTarget(LocalTarget):
    def __init__(self, schema, tablename, repo_dir, resource_id, version, filename):
        self.format = None
        self.is_tmp = False
        self.schema = schema
        self.tablename = tablename
        self.repo_dir = repo_dir
        self.resource_id = resource_id
        self.version = version
        self.filename = filename

    @property
    def path(self):
        path = self._get_path()
        if path and os.path.isfile(path):
            return path
        else:
            return self._build_path()

    def _build_path(self):
        return os.path.join(self.repo_dir, self.resource_id, str(self.version), self.filename)

    def _get_path(self):
        path = None
        query = '''
                SELECT path FROM "{schema}".{table}
                WHERE id = '{resource_id}'
                  AND version = {version}
                '''.format(schema=self.schema,
                           table=self.tablename,
                           resource_id=self.resource_id,
                           version=self.version)
        try:
            result = current_session().execute(query).fetchone()
            if result:
                path = result[0]
        except:
            path = None

        return path

    def exists(self):
        path = self._get_path()
        return path and os.path.isfile(path) class URLTarget(Target):
    '''
    Accepts both local paths and urls
    '''
    def __init__(self, url):
        self.path = url
        scheme = urlparse(url).scheme
        if scheme == '':
            self.url = 'file://{}'.format(url)
        else:
            self.url = url

    def exists(self):
        try:
            urlopen(self.url)
            return True
        except URLError:
            return False def classpath(obj):
    '''
    Returns the path to this object's class, relevant to current working dir.
    Excludes the first element of the path.  If there is only one element,
    returns ``tmp``.

    >>> classpath(object)
    'tmp'
    >>> from tasks.base_tasks import ColumnsTask
    >>> classpath(ColumnsTask())
    'util'
    >>> from tasks.es.ine import FiveYearPopulation
    >>> classpath(FiveYearPopulation())
    'es.ine'
    '''
    classpath_ = '.'.join(obj.__module__.split('.')[1:])
    return classpath_ if classpath_ else 'tmp' def query_cartodb(query, carto_url=None, api_key=None):
    '''
    Convenience function to query CARTO's SQL API with an arbitrary SQL string.
    The account connected via ``.env`` is queried.

    Returns the raw ``Response`` object.  Will not raise an exception in case
    of non-200 response.

    :param query: The query to execute on CARTO.
    '''
    carto_url = (carto_url or os.environ['CARTODB_URL']) + '/api/v2/sql'
    resp = requests.post(carto_url, data={
        'api_key': api_key or os.environ['CARTODB_API_KEY'],
        'q': query
    })
    return resp def sql_to_cartodb_table(outname, localname, json_column_names=None,
                         schema='observatory'):
    '''
    Move a table to CARTO using the
    `import API <https://carto.com/docs/carto-engine/import-api/importing-geospatial-data/>`_

    :param outname: The destination name of the table.
    :param localname: The local name of the table, exclusive of schema.
    :param json_column_names:  Optional iterable of column names that will
                               be converted to ``JSON`` type after the fact.
                               Otherwise those columns would be ``Text``.
    :param schema: Optional schema for the local table.  Defaults to
                   ``observatory``.
    '''
    private_outname = outname + '_private'
    upload_via_ogr2ogr(private_outname, localname, schema)

    # populate the_geom_webmercator
    # if you try to use the import API to copy a table with the_geom populated
    # but not the_geom_webmercator, we get an error for unpopulated column
    resp = query_cartodb(
        'UPDATE {tablename} '
        'SET the_geom_webmercator = CDB_TransformToWebmercator(the_geom) '.format(
            tablename=private_outname
        )
    )
    assert resp.status_code == 200

    print('copying via import api')
    import_api({
        'table_name': outname,
        'table_copy': private_outname,
        'create_vis': False,
        'type_guessing': False,
        'privacy': 'public'
    }, json_column_names=json_column_names)
    resp = query_cartodb('DROP TABLE "{}" CASCADE'.format(private_outname))
    assert resp.status_code == 200 def underscore_slugify(txt):
    '''
    Given a string, converts from camelcase to underscore style, and then
    slugifies.

    >>> underscore_slugify('FooBarBaz')
    'foo_bar_baz'

    >>> underscore_slugify('Foo, bar, baz')
    'foo_bar_baz'

    >>> underscore_slugify('Foo_Bar_Baz')
    'foo_bar_baz'

    >>> underscore_slugify('Foo:barBaz')
    'foo_bar_baz'
    '''
    return slugify(camel_to_underscore(re.sub(
        r'[^a-zA-Z0-9]+', '_', txt))).replace('-', '_') def shell(cmd, encoding='utf-8'):
    '''
    Run a shell command, uses :py:func:`subprocess.check_output(cmd,
    shell=True)` under the hood.

    Returns the ``STDOUT`` output, and raises an error if there is a
    none-zero exit code.
    '''
    try:
        return subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT).decode(encoding)
    except subprocess.CalledProcessError as err:
        LOGGER.error(err.output)
        raise def create_temp_schema(task):
    shell("psql -c 'CREATE SCHEMA IF NOT EXISTS \"{schema}\"'".format(schema=classpath(task))) def unqualified_task_id(task_id):
    '''
    Returns the name of the task from the task_id.
    '''
    return task_id.split('.')[-1] def generate_tile_summary(session, table_id, column_id, tablename, colname):
    '''
    Add entries to obs_column_table_tile for the given table and column.
    '''
    tablename_ns = tablename.split('.')[-1]

    query = '''
        DELETE FROM observatory.obs_column_table_tile_simple
        WHERE table_id = '{table_id}'
          AND column_id = '{column_id}';'''.format(
              table_id=table_id, column_id=column_id)
    resp = session.execute(query)

    query = '''
        DELETE FROM observatory.obs_column_table_tile
        WHERE table_id = '{table_id}'
          AND column_id = '{column_id}';'''.format(
              table_id=table_id, column_id=column_id)
    resp = session.execute(query)

    query = '''
        DROP TABLE IF EXISTS raster_empty_{tablename_ns};
        CREATE TEMPORARY TABLE raster_empty_{tablename_ns} AS
        SELECT ROW_NUMBER() OVER () AS id, rast FROM (
          WITH tilesize AS (SELECT
            CASE WHEN SUM(ST_Area({colname})) > 5000 THEN 2.5
                 ELSE 0.5 END AS tilesize,
            ST_SetSRID(ST_Extent({colname}), 4326) extent
            FROM {tablename}
          ), summaries AS (
            SELECT ST_XMin(extent) xmin, ST_XMax(extent) xmax,
                   ST_YMin(extent) ymin, ST_YMax(extent) ymax
            FROM tilesize
          ) SELECT
              ST_Tile(ST_SetSRID(
                ST_AddBand(
                  ST_MakeEmptyRaster(
                    ((xmax - xmin) / tilesize)::Integer + 1,
                    ((ymax - ymin) / tilesize)::Integer + 1,
                    ((xmin / tilesize)::Integer)::Numeric * tilesize,
                    ((ymax / tilesize)::Integer)::Numeric * tilesize,
                    tilesize
                  ), ARRAY[
                    (1, '32BF', -1, 0)::addbandarg,
                    (2, '32BF', -1, 0)::addbandarg,
                    (3, '32BF', -1, 0)::addbandarg
                  ])
              , 4326)
          , ARRAY[1, 2, 3], 25, 25) rast
          FROM summaries, tilesize
          ) foo;
        '''.format(colname=colname,
                   tablename=tablename,
                   tablename_ns=tablename_ns)
    resp = session.execute(query)
    assert resp.rowcount > 0

    query = '''
      DROP TABLE IF EXISTS raster_pap_{tablename_ns};
      CREATE TEMPORARY TABLE raster_pap_{tablename_ns} As
      SELECT id, (ST_PixelAsPolygons(FIRST(rast), 1, False)).*
           FROM raster_empty_{tablename_ns} rast,
                {tablename} vector
           WHERE rast.rast && vector.{colname}
           GROUP BY id;
    '''.format(tablename_ns=tablename_ns, tablename=tablename, colname=colname)
    resp = session.execute(query)
    assert resp.rowcount > 0

    query = '''
      CREATE UNIQUE INDEX ON raster_pap_{tablename_ns} (id, x, y);
      CREATE INDEX ON raster_pap_{tablename_ns} using gist (geom);
    '''.format(tablename_ns=tablename_ns, colname=colname)
    resp = session.execute(query)

    # Travis doesn't support ST_ClipByBox2D because of old GEOS version, but
    # our Docker container supports this optimization
    if os.environ.get('TRAVIS'):
        st_clip = 'ST_Intersection'
    else:
        st_clip = 'ST_ClipByBox2D'

    query = '''
      DROP TABLE IF EXISTS raster_vals_{tablename_ns};
      CREATE TEMPORARY TABLE raster_vals_{tablename_ns} AS
      WITH vector AS (SELECT CASE
                        WHEN ST_GeometryType({colname}) IN ('ST_Polygon', 'ST_MultiPolygon') THEN
                             ST_CollectionExtract(ST_MakeValid(
                             ST_SimplifyVW({colname}, 0.00001)), 3)
                          ELSE {colname}
                        END the_geom
                      FROM {tablename} vector)
      SELECT id
             , (null::geometry, null::numeric)::geomval median
             , (FIRST(geom),
               -- determine number of geoms, including fractions
               Nullif(SUM(CASE ST_GeometryType(vector.the_geom)
                 WHEN 'ST_Point' THEN 1
                 WHEN 'ST_LineString' THEN
                       ST_Length({st_clip}(vector.the_geom, ST_Envelope(geom))) /
                       ST_Length(vector.the_geom)
                 ELSE
                   CASE WHEN ST_Within(geom, vector.the_geom) THEN
                             ST_Area(geom) / ST_Area(vector.the_geom)
                        WHEN ST_Within(vector.the_geom, geom) THEN 1
                        ELSE ST_Area({st_clip}(vector.the_geom, ST_Envelope(geom))) /
                             ST_Area(vector.the_geom)
                   END
                END), 0)
              )::geomval cnt
             , (FIRST(geom),
               -- determine % pixel area filled with geoms
               SUM(CASE WHEN ST_Within(geom, vector.the_geom) THEN 1
                    WHEN ST_Within(vector.the_geom, geom) THEN
                         ST_Area(vector.the_geom) / ST_Area(geom)
                    ELSE ST_Area({st_clip}(vector.the_geom, ST_Envelope(geom))) /
                         ST_Area(geom)
               END)
              )::geomval percent_fill
         FROM raster_pap_{tablename_ns}, vector
         WHERE geom && vector.the_geom
         GROUP BY id, x, y;
    '''.format(tablename_ns=tablename_ns, tablename=tablename, colname=colname,
               st_clip=st_clip)
    resp = session.execute(query)
    assert resp.rowcount > 0

    query = '''
        INSERT INTO observatory.obs_column_table_tile
        WITH pixelspertile AS (SELECT id,
          ARRAY_AGG(median) medians,
          ARRAY_AGG(cnt) counts,
          ARRAY_AGG(percent_fill) percents
        FROM raster_vals_{tablename_ns}
        GROUP BY id)
        SELECT '{table_id}', '{column_id}', er.id,
                 ST_SetValues(ST_SetValues(ST_SetValues(er.rast,
                 1, medians),
                 2, counts),
                 3, percents) geom
          FROM raster_empty_{tablename_ns} er, pixelspertile ppt
          WHERE er.id = ppt.id
       '''.format(tablename_ns=tablename_ns, tablename=tablename,
                  table_id=table_id, column_id=column_id)
    resp = session.execute(query)
    assert resp.rowcount > 0

    resp = session.execute('''
        UPDATE observatory.obs_column_table_tile
        SET tile = st_setvalues(st_setvalues(st_setvalues(tile,
                    1, geomvals, false),
                    2, geomvals, false),
                    3, geomvals, false)
        FROM (
            SELECT table_id, column_id, tile_id, array_agg(((geomval).geom, 0)::geomval) geomvals FROM (
                SELECT table_id, column_id, tile_id, st_dumpaspolygons(tile, 2, false) geomval
                FROM observatory.obs_column_table_tile
                WHERE column_id = '{column_id}'
                  AND table_id = '{table_id}'
            ) bar
            WHERE (geomval).val = -1
            GROUP BY table_id, column_id, tile_id
        ) foo
          WHERE obs_column_table_tile.table_id = foo.table_id
            AND obs_column_table_tile.column_id = foo.column_id
            AND obs_column_table_tile.tile_id = foo.tile_id
       ; '''.format(table_id=table_id, column_id=column_id,
                    colname=colname, tablename=tablename))
    resp = session.execute('''
       INSERT INTO observatory.obs_column_table_tile_simple
       SELECT table_id, column_id, tile_id, ST_Reclass(
         ST_Band(tile, ARRAY[2, 3]),
         -- regenerate band 2 setting 1 as the minimum value to include very small geometries
         -- that would be considered spurious due to rounding
         ROW(2, '[0-0]::0, (0-1]::1-255, (1-100]::255-255', '8BUI', 0)::reclassarg
       ) AS tile
       FROM observatory.obs_column_table_tile
       WHERE table_id='{table_id}'
         AND column_id='{column_id}';
       '''.format(table_id=table_id, column_id=column_id,
                  colname=colname, tablename=tablename))

    real_num_geoms = session.execute('''
        SELECT COUNT({colname}) FROM {tablename}
    '''.format(colname=colname,
               tablename=tablename)).fetchone()[0]

    est_num_geoms = session.execute('''
        SELECT (ST_SummaryStatsAgg(tile, 1, false)).sum
        FROM observatory.obs_column_table_tile_simple
        WHERE column_id = '{column_id}'
          AND table_id = '{table_id}'
    '''.format(column_id=column_id,
               table_id=table_id)).fetchone()[0]

    assert abs(real_num_geoms - est_num_geoms) / real_num_geoms < 0.05, \
        "Estimate of {} total geoms more than 5% off real {} num geoms " \
        "for column '{}' in table '{}' (tablename '{}')".format(
            est_num_geoms, real_num_geoms, column_id, table_id, tablename) def uncompress_file(zip_file):
    LOGGER.debug("Uncompressing {zip_file}".format(zip_file=zip_file))
    try:
        with zipfile.ZipFile('{zip_file}.zip'.format(zip_file=zip_file), 'r') as z:
            z.extractall(path='{zip_file}'.format(zip_file=zip_file))
    except NotImplementedError as err:
        s_err = str(err)
        LOGGER.warn("%s.zip error: %s. Fallback to command line...", zip_file, s_err)
        if s_err == 'compression type 9 (deflate64)':
            # Support for unsupported file types, such as PKWare deflate64 format
            subprocess.check_call(['unzip', '{zip_file}.zip'.format(zip_file=zip_file), '-d', zip_file])
        else:
            raise def copyfile(src, dst):
    # From http://luigi.readthedocs.io/en/stable/_modules/luigi/local_target.html#LocalTarget.makedirs
    normpath = os.path.normpath(dst)
    parentfolder = os.path.dirname(normpath)
    if parentfolder:
        try:
            os.makedirs(parentfolder, exist_ok=True)
        except OSError as e:
            LOGGER.error('Error making dirs for copyfile: %s', e)
            pass

    shutil.copyfile(src, dst) SIMPLIFIED_SUFFIX = '_simpl' class Simplify(WrapperTask):
    schema = Parameter()
    table = Parameter()
    table_id = Parameter(default='')
    suffix = Parameter(default=SIMPLIFIED_SUFFIX)

    def __init__(self, *args, **kwargs):
        super(Simplify, self).__init__(*args, **kwargs)

        self.table_key = self.table
        if self.table_id:
            self.table_key = self.table_id

    def requires(self):
        params = get_simplification_params(self.table_key.lower())
        if not params:
            LOGGER.error("Simplification not found. Edit 'simplifications.json' and add an entry for '{}'".format(self.table_key.lower()))

        simplification = params.get(SIMPLIFICATION, DEFAULT_SIMPLIFICATION)
        LOGGER.info("Simplifying %s.%s using %s", self.schema, self.table, simplification)
        table_output = '{tablename}{suffix}'.format(tablename=self.table, suffix=self.suffix)
        if simplification == SIMPLIFICATION_MAPSHAPER:
            return SimplifyGeometriesMapshaper(schema=self.schema,
                                               table_input=self.table,
                                               table_output=table_output,
                                               geomfield=params[GEOM_FIELD],
                                               retainfactor=params[FACTOR],
                                               skipfailures=params.get(SKIP_FAILURES, DEFAULT_SKIPFAILURES),
                                               maxmemory=params.get(MAX_MEMORY, DEFAULT_MAXMEMORY))
        elif simplification == SIMPLIFICATION_POSTGIS:
            return SimplifyGeometriesPostGIS(schema=self.schema,
                                             table_input=self.table,
                                             table_output=table_output,
                                             geomfield=params[GEOM_FIELD],
                                             retainfactor=params[FACTOR])
        else:
            raise ValueError('Invalid simplification "{simplification}" for {table}'.format(
                simplification=simplification, table=self.table))
import os
import re
import json
import time
import requests
import subprocess
import importlib
import inspect
import zipfile
import gzip
import csv
import uuid
import urllib.request
from urllib.parse import quote_plus
from collections import OrderedDict
from datetime import date
from luigi import (Task, Parameter, LocalTarget, BoolParameter, IntParameter,
                   ListParameter, DateParameter, WrapperTask, Event, ExternalTask)
from luigi.contrib.s3 import S3Target
from sqlalchemy.dialects.postgresql import JSON
from google.cloud import storage
from lib.util import digest_file
from lib.logger import get_logger
from tasks.meta import (OBSColumn, OBSTable, metadata, current_session,
                        session_commit, session_rollback, GEOM_REF)
from tasks.targets import (ColumnTarget, TagTarget, CartoDBTarget, PostgresTarget, TableTarget, RepoTarget, URLTarget)
from tasks.util import (classpath, query_cartodb, sql_to_cartodb_table, underscore_slugify, shell,
                        create_temp_schema, unqualified_task_id, generate_tile_summary, uncompress_file,
                        copyfile)
from tasks.simplification import SIMPLIFIED_SUFFIX
from tasks.simplify import Simplify







LOGGER = get_logger(__name__)

MAX_PG_IDENTIFIER_LENGTH = 63

class ColumnsTask(Task):
    '''
    The ColumnsTask provides a structure for generating metadata. The only
    required method is :meth:`~.ColumnsTask.columns`.
    The keys may be used as human-readable column names in tables based off
    this metadata, although that is not always the case. If the id of the
    :class:`OBSColumn <tasks.meta.OBSColumn>` is left blank, the dict's key will be used to
    generate it (qualified by the module).

    Also, conventionally there will be a requires method that brings in our
    standard tags: :class:`SectionTags <tasks.tags.SectionTags>`,
    :class:`SubsectionTags <tasks.tags.SubsectionTags>`, and
    :class:`UnitTags <tasks.tags.UnitTags>`.
    This is an example of defining several tasks as prerequisites: the outputs
    of those tasks will be accessible via ``self.input()[<key>]`` in other
    methods.

    This will update-or-create columns defined in it when run.  Whether a
    column is updated or not depends on whether it exists in the database with
    the same :meth:`~.ColumnsTask.version` number.
    '''

    def columns(self):
        '''
        This method must be overriden in subclasses.  It must return a
        :py:class:`collections.OrderedDict` whose values are all instances of
        :class:`OBSColumn <tasks.meta.OBSColumn>` and whose keys are all strings.
        '''
        raise NotImplementedError('Must return iterable of OBSColumns')

    def on_failure(self, ex):
        session_rollback(self, ex)
        super(ColumnsTask, self).on_failure(ex)

    def on_success(self):
        session_commit(self)

    def run(self):
        for _, coltarget in self.output().items():
            coltarget.update_or_create()

    def version(self):
        '''
        Returns a number that will be linked to the all
        :py:attr:`tasks.meta.OBSColumn.version` objects resulting from this
        task.  If this is identical to what's already in the database, the
        :class:`OBSColumn <tasks.meta.OBSColumn>` will not be replaced.
        '''
        return 0

    def prefix(self):
        if self.task_namespace and self.task_namespace != '__not_user_specified':
            splitted_namespace = self.task_namespace.split('.')
            # Backward compatibility to avoid rerun all the tasks
            # to me removed for the next rebuild
            if splitted_namespace[0] == 'tasks':
                splitted_namespace.pop(0)
                return '.'.join(splitted_namespace)
            else:
                return self.task_namespace
        else:
            return classpath(self)

    def output(self):
        session = current_session()

        # Return columns from database if the task is already finished
        if hasattr(self, '_colids') and self.complete():
            return OrderedDict([

                (colkey, ColumnTarget(session.query(OBSColumn).get(cid), self))