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))
                for colkey, cid in self.colids.items()
            ])

        # Otherwise, run `columns` (slow!) to generate output
        already_in_session = [obj for obj in session]
        output = OrderedDict()

        input_ = self.input()
        for col_key, col in self.columns().items():
            if not isinstance(col, OBSColumn):
                raise RuntimeError(
                    'Values in `.columns()` must be of type OBSColumn, but '
                    '"{col}" is type {type}'.format(col=col_key, type=type(col)))
            if not col.version:
                col.version = self.version()
            col.id = '.'.join([self.prefix(), col.id or col_key])
            tags = self.tags(input_, col_key, col)
            if isinstance(tags, TagTarget):
                col.tags.append(tags)
            else:
                col.tags.extend(tags)

            output[col_key] = ColumnTarget(col, self)
        now_in_session = [obj for obj in session]
        for obj in now_in_session:
            if obj not in already_in_session:
                if obj in session:
                    session.expunge(obj)
        return output

    @property
    def colids(self):
        '''
        Return colids for the output columns, this can be cached
        '''
        if not hasattr(self, '_colids'):
            self._colids = OrderedDict([
                (colkey, ct._id) for colkey, ct in self.output().items()
            ])
        return self._colids

    def complete(self):
        '''
        Custom complete method that attempts to check if output exists, as is
        default, but in case of failure allows attempt to run dependencies (a
        missing dependency could result in exception on `output`).
        '''
        deps = self.deps()
        if deps and not all([d.complete() for d in deps]):
            return False
        else:
            # bulk check that all columns exist at proper version
            cnt = current_session().execute(
                '''
                SELECT COUNT(*)
                FROM observatory.obs_column
                WHERE id IN ('{ids}') AND version = '{version}'
                '''.format(
                    ids="', '".join(list(self.colids.values())),
                    version=self.version()
                )).fetchone()[0]
            return cnt == len(list(self.colids.values()))

    def tags(self, input_, col_key, col):
        '''
        Replace with an iterable of :class:`OBSColumn <tasks.meta.OBSColumn>`
        that should be applied to each column

        :param input_: A saved version of this class's :meth:`input <luigi.Task.input>`
        :param col_key: The key of the column this will be applied to.
        :param column: The :class:`OBSColumn <tasks.meta.OBSColumn>` these tags
                       will be applied to.
        '''
        return [] class TableTask(Task):
    '''
    This task creates a single output table defined by its name, path, and
    defined columns.

    :meth:`~.TableTask.timespan`, :meth:`~.TableTask.columns`, and
    :meth:`~.TableTask.populate` must be overwritten in subclasses.

    When run, this task will automatically create the table with a schema
    corresponding to the defined columns, with a unique name.  It will also
    generate all relevant metadata for the table, and link it to the columns.
    '''

    def _requires(self):
        reqs = super(TableTask, self)._requires()
        if self._testmode:
            return [r for r in reqs if isinstance(r, (TagsTask, TableTask, ColumnsTask))]
        else:
            return reqs

    def version(self):
        '''
        Must return a version control number, which is useful for forcing a
        re-run/overwrite without having to track down and delete output
        artifacts.
        '''
        return 0

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

    def on_success(self):
        session_commit(self)
        super(TableTask, self).on_success()

    def columns(self):
        '''
        Must be overriden by subclasses.  Columns returned from this function
        determine the schema of the resulting :class:`~tasks.targets.TableTarget`.

        The return for this function should be constructed by selecting the
        desired columns from :class:`~tasks.base_tasks.ColumnTask`s, specified as
        inputs in :meth:`~.tasks.TableTask.requires()`

        Must return a :py:class:`~collections.OrderedDict` of
        (colname, :class:`~tasks.targets.ColumnTarget`) pairs.

        '''
        raise NotImplementedError('Must implement columns method that returns '
                                  'a dict of ColumnTargets')

    def populate(self):
        '''
        This method must populate (most often via ``INSERT``) the output table.

        For example:
        '''
        raise NotImplementedError('Must implement populate method that '
                                  'populates the table')

    def fake_populate(self, output):
        '''
        Put one empty row in the table
        '''
        session = current_session()
        session.execute('INSERT INTO {output} ({col}) VALUES (NULL)'.format(
            output=output.table,
            col=list(self._columns.keys())[0]
        ))

    def description(self):
        '''
        Optional description for the :class:`~tasks.meta.OBSTable`.  Not
        currently used anywhere.
        '''
        return None

    def targets(self):
        return {}

    def table_timespan(self):
        '''
        Must return an OBSTimespan instance that identifies the date range
        or point-in-time for this table.
        Must be implemented by subclass.
        '''
        raise NotImplementedError('Must return an OBSTimespan for table')

    def the_geom(self, output, colname):
        session = current_session()
        return session.execute(
            'SELECT ST_AsText( '
            '  ST_Intersection( '
            '    ST_MakeEnvelope(-179.999, -89.999, 179.999, 89.999, 4326), '
            '    ST_Multi( '
            '      ST_CollectionExtract( '
            '        ST_MakeValid( '
            '          ST_SnapToGrid( '
            '            ST_Buffer( '
            '              ST_Union( '
            '                ST_MakeValid( '
            '                  ST_Simplify( '
            '                    ST_SnapToGrid({geom_colname}, 0.3) '
            '                  , 0) '
            '                ) '
            '              ) '
            '            , 0.3, 2) '
            '          , 0.3) '
            '        ) '
            '      , 3) '
            '    ) '
            '  ) '
            ') the_geom '
            'FROM {output}'.format(
                geom_colname=colname,
                output=output.table
            )).fetchone()['the_geom']

    @property
    def _testmode(self):
        if os.environ.get('ENVIRONMENT') == 'test':
            return True
        return getattr(self, '_test', False)

    def run(self):
        LOGGER.info('getting output()')
        before = time.time()
        output = self.output()
        after = time.time()
        LOGGER.info('time: %s', after - before)

        LOGGER.info('update_or_create_table')
        before = time.time()
        output.update_or_create_table()
        after = time.time()
        LOGGER.info('time: %s', after - before)

        if self._testmode:
            LOGGER.info('fake_populate')
            before = time.time()
            self.fake_populate(output)
            after = time.time()
            LOGGER.info('time: %s', after - before)
        else:
            LOGGER.info('populate')
            before = time.time()
            self.populate()
            after = time.time()
            LOGGER.info('time: %s', after - before)

        before = time.time()
        LOGGER.info('update_or_create_metadata')
        output.update_or_create_metadata(_testmode=self._testmode)
        after = time.time()
        LOGGER.info('time: %s', after - before)

        LOGGER.info('analyzing the table %s', self.output().table)
        session = current_session()
        session.execute('ANALYZE {table}'.format(table=self.output().table))

        if not self._testmode:
            LOGGER.info('checking for null columns on %s', self.output().table)
            self.check_null_columns()

            LOGGER.info('checking for medians/averages without universe target on %s',
                        self.output().table)
            self.check_universe_in_aggregations()

            LOGGER.info('checking for table_to_table relations on %s',
                        self.output().table)
            self.check_table_to_table_relations()

            LOGGER.info('create_indexes')
            self.create_indexes(output)
            current_session().flush()

            LOGGER.info('create_geom_summaries')
            self.create_geom_summaries(output)

            LOGGER.info('checking for null columns on %s', self.output().table)
            self.check_null_columns()

            LOGGER.info('checking for medians/averages without universe target on %s',
                        self.output().table)
            self.check_universe_in_aggregations()

    def create_indexes(self, output):
        session = current_session()
        tablename = output.table
        for colname, coltarget in self._columns.items():
            col = coltarget._column
            index_type = col.index_type
            if index_type:
                index_name = '{}_{}_idx'.format(tablename.split('.')[-1], colname)
                session.execute('CREATE {unique} INDEX IF NOT EXISTS {index_name} ON {table} '
                                'USING {index_type} ({colname})'.format(
                                    unique='',
                                    index_type=index_type,
                                    index_name=index_name,
                                    table=tablename, colname=colname))

    def create_geom_summaries(self, output):
        geometry_columns = [
            (colname, coltarget._id) for colname, coltarget in
            self.columns().items() if coltarget._column.type.lower().startswith('geometry')
        ]

        if len(geometry_columns) == 0:
            return
        elif len(geometry_columns) > 1:
            raise Exception('Having more than one geometry column in one table '
                            'could lead to problematic behavior ')
        colname, colid = geometry_columns[0]
        # Use SQL directly instead of SQLAlchemy because we need the_geom set
        # on obs_table in this session
        current_session().execute("UPDATE observatory.obs_table "
                                  "SET the_geom = ST_GeomFromText('{the_geom}', 4326) "
                                  "WHERE id = '{id}'".format(
                                      the_geom=self.the_geom(output, colname),
                                      id=output._id
                                  ))
        generate_tile_summary(current_session(),
                              output._id, colid, output.table, colname)

    def check_null_columns(self):
        session = current_session()
        query = ('''
                 SELECT attname FROM pg_stats WHERE schemaname = 'observatory'
                 AND tablename = '{table}' AND null_frac = 1
                 '''.format(table=self.output()._tablename))
        result = session.execute(query).fetchall()

        if result:
            raise ValueError('The following columns of the table "{table}" contain only NULL values: {columns}'.format(
                table=self.output().table, columns=', '.join([x[0] for x in result])))

    def check_universe_in_aggregations(self):
        session = current_session()
        query = ('''
                 SELECT c.id, c.aggregate, STRING_AGG(COALESCE(cc.reltype,''), ',') reltype
                 FROM observatory.obs_table t
                 INNER JOIN observatory.obs_column_table ct ON t.id = ct.table_id
                 INNER JOIN observatory.obs_column c ON ct.column_id = c.id
                 FULL JOIN observatory.obs_column_to_column cc ON c.id = cc.source_id
                 WHERE t.tablename = '{table}'
                 AND c.aggregate IN ('average', 'median')
                 GROUP BY 1, 2
                 HAVING 'universe' <> ANY(ARRAY_AGG(COALESCE(LOWER(cc.reltype),'')))
                 '''.format(table=self.output()._tablename))
        result = session.execute(query).fetchall()

        if result:
            raise ValueError("The following columns of the table \"{table}\" are aggregated as 'median' or 'average' "
                             "but lack of 'universe' target: {columns}".format(
                                table=self.output().table, columns=', '.join([x[0] for x in result])))

    def check_table_to_table_relations(self):
        session = current_session()
        # Relations between tables defined as follows:
        # TABLE(SOURCE) -- COLUMN_TABLE(source) -- COLUMN_TO_COLUMN -- COLUMN_TABLE(target) -- TABLE(target)
        # that are not in OBS_TABLE_TO_TABLE
        check_missing_relations_query = ('''
                 SELECT DISTINCT ts.id || '.' || cs.id ||
                 ' / ' || tt.id || '.' || ct.id || ' (' || cc.reltype || ')' relation
                 FROM observatory.obs_table ts, observatory.obs_column_table cts, observatory.obs_column cs,
                 observatory.obs_column_to_column cc,
                 observatory.obs_table tt, observatory.obs_column_table ctt, observatory.obs_column ct
                 WHERE cc.reltype = 'geom_ref'  -- WARNING! only geom_ref reltype ATM
                 AND ts.tablename = '{table}'
                 AND ts.id = cts.table_id
                 AND cts.column_id = cs.id
                 AND cs.id = cc.source_id
                 AND cc.target_id = ct.id
                 AND ct.id = ctt.column_id
                 AND ctt.table_id = tt.id
                 AND NOT EXISTS (
                 SELECT 1
                 FROM observatory.obs_table_to_table ttt
                 WHERE ttt.source_id = ts.id
                 AND ttt.target_id = tt.id)
                 ''').format(table=self.output()._tablename)
        result = session.execute(check_missing_relations_query).fetchall()

        if result:
            raise ValueError('The table "{table}" lacks some table_to_table relations: {relations}'.format(
                table=self.output().table, relations=', '.join([x[0] for x in result])))

        # Relations in OBS_TABLE_TO_TABLE
        # that are not in the relations between tables defined as follows:
        # TABLE(SOURCE) -- COLUMN_TABLE(source) -- COLUMN_TO_COLUMN -- COLUMN_TABLE(target) -- TABLE(target)
        check_excess_relation_query = ('''
                 SELECT DISTINCT ttt.source_id || ' / '
                 || ttt.target_id || ' (' || ttt.reltype || ')' relation
                 FROM observatory.obs_table_to_table ttt, observatory.obs_table t
                 WHERE ttt.source_id = t.id
                 AND t.tablename = '{table}'
                 AND ttt.source_id <> ttt.target_id
                 AND NOT EXISTS (
                 SELECT 1
                 FROM observatory.obs_table ts, observatory.obs_column_table cts, observatory.obs_column cs,
                 observatory.obs_column_to_column cc,
                 observatory.obs_table tt, observatory.obs_column_table ctt, observatory.obs_column ct
                 WHERE cc.reltype = 'geom_ref'  -- WARNING! only geom_ref reltype ATM
                 AND ts.tablename = t.tablename
                 AND ts.id = cts.table_id
                 AND cts.column_id = cs.id
                 AND cs.id = cc.source_id
                 AND cc.target_id = ct.id
                 AND ct.id = ctt.column_id
                 AND ctt.table_id = tt.id
                 AND ts.id <> tt.id  -- avoid relationships between columns of the same table
                 AND NOT EXISTS (  -- avoid columns with the same id as columns in the target table
                 SELECT 1
                 FROM observatory.obs_column_table oct
                 WHERE oct.table_id = ts.id
                 AND oct.column_id = ct.id)
                 AND ttt.source_id = ts.id
                 AND ttt.reltype = cc.reltype
                 AND ttt.target_id = tt.id)
                 ''').format(table=self.output()._tablename)
        result = session.execute(check_excess_relation_query).fetchall()

        if result:
            raise ValueError('The following relations for table "{table}" need to be removed: {relations}'.format(
                table=self.output().table, relations=', '.join([x[0] for x in result])))

    def schema(self):
        return classpath(self)

    def name(self):
        return underscore_slugify(unqualified_task_id(self.task_id))

    def output(self):
        if not hasattr(self, '_columns'):
            self._columns = self.columns()

        tt = TableTarget(self.schema(),
                         self.name(),
                         OBSTable(description=self.description(),
                                  version=self.version(),
                                  table_timespan=self.table_timespan(),
                                  targets=self.targets()),
                         self._columns, self)
        return tt

    def complete(self):
        return TableTarget(self.schema(),
                           self.name(),
                           OBSTable(description=self.description(),
                                    version=self.version(),
                                    table_timespan=self.table_timespan()),
                           [], self).exists() class TagsTask(Task):
    '''
    This will update-or-create :class:`OBSTag <tasks.meta.OBSTag>` objects
    int the database when run.

    The :meth:`~.TagsTask.tags` method must be overwritten.

    :meth:`~TagsTask.version` is used to control updates to the database.
    '''

    def tags(self):
        '''
        This method must be overwritten in subclasses.

        The return value must be an iterable of instances of
        :class:`OBSTag <tasks.meta.OBSTag>`.
        '''
        raise NotImplementedError('Must return iterable of OBSTags')

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

    def on_success(self):
        session_commit(self)

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

    def version(self):
        return 0

    def output(self):
        # if self.deps() and not all([d.complete() for d in self.deps()]):
        #    raise Exception('Must run prerequisites first')
        output = {}
        for tag in self.tags():
            orig_id = tag.id
            tag.id = '.'.join([classpath(self), orig_id])
            if not tag.version:
                tag.version = self.version()
            output[orig_id] = TagTarget(tag, self)
        return output class GeoFile2TempTableTask(TempTableTask):
    '''
    A task that loads :meth:`~.tasks.GeoFile2TempTableTask.input_files()`
    into a temporary Postgres table. That method must be overriden.
    '''

    encoding = Parameter(default='latin1', significant=False)
    other_options = ''

    def input_files(self):
        '''
        This method must be implemented by subclasses.  Should return either
        a path to a single file, or an iterable of paths to files.
        In that case, the first in the list will determine the schema.
        '''
        raise NotImplementedError("Must specify `input_files` method")

    def run(self):
        if isinstance(self.input_files(), str):
            files = [self.input_files()]
        else:
            files = self.input_files()
        schema = self.output().schema
        tablename = self.output().tablename
        operation = '-overwrite -lco OVERWRITE=yes -lco SCHEMA={schema} -lco PRECISION=no'.format(
            schema=schema)
        for geofile in files:
            # First we  try to import using utf8 as encoding if not able we
            # fallback to the encoding passed as parameter
            cmd = self.build_ogr_command(encoding='utf-8',
                                         schema=schema,
                                         tablename=tablename,
                                         input=geofile,
                                         operation=operation,
                                         other_options=self.other_options)
            output = shell(cmd)
            if self.utf8_error(output):
                cmd = self.build_ogr_command(encoding=self.encoding,
                                             schema=schema,
                                             tablename=tablename,
                                             input=geofile,
                                             operation=operation,
                                             other_options=self.other_options)
                shell(cmd)
            # We don't add lco (layer creation options) because in the append mode
            # are ignored
            operation = '-append '.format(schema=schema)

    def utf8_error(self, cmd_output):
        regex = re.compile('invalid byte sequence for encoding \"UTF8\"', re.IGNORECASE)
        return re.search(regex, cmd_output) is not None

    def build_ogr_command(self, **args):
        return 'PG_USE_COPY=yes PGCLIENTENCODING=UTF8 ' \
               'ogr2ogr --config SHAPE_ENCODING {encoding} -f PostgreSQL PG:"dbname=$PGDATABASE ' \
               'active_schema={schema}" -t_srs "EPSG:4326" ' \
               '{other_options} ' \
               '-nlt MultiPolygon -nln {table} ' \
               '{operation} \'{input}\' '.format(encoding=args['encoding'],
                                                 schema=args['schema'],
                                                 table=args['tablename'],
                                                 input=args['input'],
                                                 operation=args['operation'],
                                                 other_options=args['other_options']) class SimplifiedTempTableTask(TempTableTask):
    def get_table_id(self):
        '''
        Subclasses may override this method if an ETL task
        needs a custom table_id.

        Returns schema.tablename_without_hash
        '''
        return '.'.join([self.input().schema, '_'.join(self.input().tablename.split('_')[:-1])])

    def get_suffix(self):
        '''
        Subclasses may override this method if an ETL task
        needs a custom suffix.
        '''
        return SIMPLIFIED_SUFFIX

    def run(self):
        yield Simplify(schema=self.input().schema,
                       table=self.input().tablename,
                       table_id=self.get_table_id(),
                       suffix=self.get_suffix())

    def output(self):
        return PostgresTarget(self.input().schema,
                              self.input().tablename + self.get_suffix()) class RepoFile(Task):
    resource_id = Parameter()
    url = Parameter()
    version = IntParameter()
    downloader = Parameter(default='base_downloader', significant=False)

    _repo_dir = 'repository'
    _new_file_name = str(uuid.uuid4())

    def requires(self):
        return CreateRepoTable()

    def run(self):
        self.output().makedirs()
        self._retrieve_remote_file()
        digest = digest_file(self.output().path)
        self._to_db(self.resource_id, self.version, digest, self.url, self.output().path)

    def _retrieve_remote_file(self):
        LOGGER.info('Downloading remote file')
        if os.path.isfile(self.output().path):
            os.remove(self.output().path)

        # As you can't pass functions through Luigi parameters, this
        # choose of behaviour is needed. A better design would involve
        # dependencies.
        downloader = cdrc_downloader if self.downloader == 'cdrc' else base_downloader
        downloader(self.url, self.output().path)

    def _to_db(self, resource_id, version, checksum, url, path):
        LOGGER.info('Storing entry in repository')
        query = '''
                DELETE FROM "{schema}".{table}
                WHERE id = '{resource_id}' AND version = {version};
                '''.format(schema=self.input().schema,
                           table=self.input().tablename,
                           resource_id=resource_id,
                           version=version,
                           url=url,
                           checksum=checksum,
                           path=path)
        current_session().execute(query)
        query = '''
                INSERT INTO "{schema}".{table}
                    (id, version, url, checksum, path, added)
                SELECT '{resource_id}', {version}, '{url}', '{checksum}', '{path}', now();
                '''.format(schema=self.input().schema,
                           table=self.input().tablename,
                           resource_id=resource_id,
                           version=version,
                           url=url,
                           checksum=checksum,
                           path=path)
        current_session().execute(query)
        current_session().commit()

    def output(self):
        return RepoTarget(self.input().schema, self.input().tablename,
                          self._repo_dir, self.resource_id, self.version, self._new_file_name) def get_timespan(timespan_id):
    ts_id, ts_alias, ts_name, ts_description, ts_timespan = parse_timespan(timespan_id)
    session = current_session()
    return OBSTimespan.as_unique(session, ts_id, id=ts_id, alias=ts_alias, name=ts_name,
                                 description=ts_description, timespan=ts_timespan) 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 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 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) class SectionTags(TagsTask):

    def version(self):
        return 5

    def tags(self):
        return [
            OBSTag(id='global',
                   name='Global',
                   type='section',
                   description=''),
            OBSTag(id='eu',
                   name='European Union',
                   type='section',
                   description=''),
            OBSTag(id='united_states',
                   name='United States',
                   type='section',
                   description=''),
            OBSTag(id='spain',
                   name='Spain',
                   type='section',
                   description=''),
            OBSTag(id='mx',
                   name='Mexico',
                   type='section',
                   description=''),
            OBSTag(id='uk',
                   name='United Kingdom',
                   type='section',
                   description=''),
            OBSTag(id='fr',
                   name='France',
                   type='section',
                   description=''),
            OBSTag(id='ca',
                   name='Canada',
                   type='section',
                   description=''),
            OBSTag(id='th',
                   name='Thailand',
                   type='section',
                   description=''),
            OBSTag(id='au',
                   name='Australia',
                   type='section',
                   description=''),
            OBSTag(id='br',
                   name='Brazil',
                   type='section',
                   description='')
        ] class SubsectionTags(TagsTask):

    def version(self):
        return 24

    def tags(self):
        return [
            OBSTag(id='housing',
                   name='Housing',
                   type='subsection',
                   description='What type of housing exists and how do people live in it?'),
            OBSTag(id='income',
                   name='Income',
                   type='subsection',
                   description='''How much people earn.'''),
            OBSTag(id='education',
                   name='Education',
                   type='subsection',
                   description='Educational attainment and enrollment.'),
            OBSTag(id='employment',
                   name='Employment',
                   type='subsection',
                   description='How people are employed.'),
            OBSTag(id='families',
                   name="Families",
                   type='subsection',
                   description='Familial arrangements of people and households.'),
            OBSTag(id='language',
                   name="Language",
                   type='subsection',
                   description='What languages people speak.'),
            OBSTag(id='age_gender',
                   name='Age and Gender',
                   type='subsection',
                   description='Population breakdowns by age and gender.'),
            OBSTag(id='nationality',
                   name='Nationality',
                   type='subsection',
                   description='''Population breakdowns by nationality and place of birth.''',),
            OBSTag(id='race_ethnicity',
                   name='Race and Ethnicity',
                   type='subsection',
                   description='''Population breakdowns by race and ethnicity. '''),
            OBSTag(id='transportation',
                   name='Transportation',
                   type='subsection',
                   description='How do people move from place to place?'),
            OBSTag(id='boundary',
                   name='Boundaries',
                   type='subsection',
                   description='Political, administrative, and census-based boundaries.'),
            OBSTag(id='religion',
                   name='Religion',
                   type='subsection',
                   description='Breakdowns of the population by religion.'),
            OBSTag(id='migration',
                   name='Migration',
                   type='subsection',
                   description='Patterns of migration.'),
            OBSTag(id='health',
                   name='Health',
                   type='subsection',
                   description='Breakdowns of the population by health.'),
            OBSTag(id='commerce_economy',
                   name='Commerce and Economy',
                   type='subsection',
                   description='Broader measures of economic and commercial activity.'),
            OBSTag(id='business_demography',
                   name='Business Demography',
                   type='subsection',
                   description='Measures of enterprise growth and decline.'),
            OBSTag(id='segments',
                   name='Population segments',
                   type='subsection',
                   description='Segmentations of the population.'),
            OBSTag(id='society',
                   name='Society',
                   type='subsection',
                   description='Measures of society quality of life'),
            OBSTag(id='energy',
                   name='Energy',
                   type='subsection',
                   description='Measures related to energy production and consumption.'),
            OBSTag(id='environmental',
                   name='Environmental',
                   type='subsection',
                   description='Attributes of the natural environment'),
            OBSTag(id='roads',
                   name='Roads',
                   type='subsection',
                   description=''),
            OBSTag(id='names',
                   name='Proper names',
                   type='subsection',
                   description='Proper names for boundaries and other features'),
            OBSTag(id='poi',
                   name='Points of Interest',
                   type='subsection',
                   description=''),
        ] class BoundaryTags(TagsTask):
    '''
    Boundary tags ids:
    - `cartographic_boundary`: shoreline clipped (and looks good on a map, hence "cartographic").
    - `interpolation_boundary`: clipped boundaries that are full coverage of the country/region.
    '''
    def version(self):
        return 1

    def tags(self):
        return [
            OBSTag(id='interpolation_boundary',
                   name='Interpolation Boundary',
                   type='boundary_type',
                   description='Boundary appropriate for interpolation calculation'),
            OBSTag(id='cartographic_boundary',
                   name='Cartographic Boundary',
                   type='boundary_type',
                   description='Boundary appropriate for cartographic use'),
            ] class OBSTable(Base):
    '''
    Describes a physical table in our database.

    These should *never* be instantiated manually.  They are automatically
    created by :meth:`~.tasks.TableTask.output`.  The unique key is
    :attr:~.meta.OBSTable.id:.

    .. py:attribute:: id

       The unique identifier for this table.  Is always qualified by
       the module name of the class that created it.

    .. py:attribute:: columns

       An iterable of all the :class:`~.meta.OBSColumnTable` instances
       contained in this table.

    .. py:attribute:: tablename

       The automatically generated name of this table, which can be used
       directly in select statements.  Of the format ``obs_<hash>``.

    .. py:attribute:: timespan

       An OBSTimespan instance containing information about the timespan
       this table applies to.
       Obtained from :meth:`~.tasks.TableTask.timespan`.

    .. py:attribute:: the_geom

       A simple geometry approximating the boundaries of the data in this
       table.

    .. py:attribute:: description

       A description of the table.  Not used.

    .. py:attribute:: version

       A version control number, used to determine whether the table and its
       metadata should be updated.

    '''
    __tablename__ = 'obs_table'

    id = Column(Text, primary_key=True)  # fully-qualified id like 'us.census.acs.extract_2013_5yr_state_18357fba'

    columns = relationship("OBSColumnTable", back_populates="table",
                           cascade="all,delete")

    tablename = Column(Text, nullable=False)
    timespan = Column(Text, ForeignKey('obs_timespan.id'))
    table_timespan = relationship("OBSTimespan")

    the_geom = Column(Geometry)
    description = Column(Text)

    targets = association_proxy('tgts', 'reltype', creator=tabletotabletargets_creator)

    version = Column(Numeric, default=0, nullable=False)

    def geom_column(self):
        '''
        Return the column geometry column for this table, if it has one.

        Returns None if there is none.
        '''
        for col in self.columns:
            if lower(col.type) == 'geometry':
                return col

    def geomref_column(self):
        '''
        Return the geomref column for this table, if it has one.

        Returns None if there is none.
        '''
        for col in self.columns:
            col = col.column
            if col.is_geomref():
                return col class OBSColumn(Base):
    '''
    Describes the characteristics of data in a column, which can exist in
    multiple physical tables.

    These should *only* be instantiated to generate the return value of
    :meth:`.ColumnsTask.columns`.  Any other usage could have unexpected
    consequences.

    .. py:attribute:: id

       The unique identifier for this column.  Should be qualified by
       the module name of the class that created it.  Is automatically
       generated by :class:`~.tasks.ColumnsTask` if left unspecified,
       and automatically qualified by module name either way.

    .. py:attribute:: type

       The type of this column -- for example, ``Text``, ``Numeric``,
       ``Geometry``, etc.  This is used to generate a schema for any
       table using this column.

    .. py:attribute:: name

       The human-readable name of this column.  This is used in the
       catalog and API.

    .. py:attribute:: description

       The human-readable description of this column.  THis is
       used in the catalog.

    .. py:attribute:: weight

       A numeric weight for this column.  Higher weights are favored
       in certain rankings done by the API. Defaults to zero, which hides
       the column from catalog.

    .. py:attribute:: aggregate

       How this column can be aggregated.  For example,
       populations can be summed, so a population column should
       be ``sum``. Should be left blank if it is not to
       aggregate.

    .. py:attribute:: tables

       Iterable of all linked :class:`~.meta.OBSColumnTable`s, which could be
       traversed to find all tables with this column.

    .. py:attribute:: tags

       Iterable of all linked :class:`~.meta.OBSColumnTag`s, which could be
       traversed to find all tags applying to this column.

    .. py:attribute:: targets

       Dict of all related columns.  Format is ``<OBSColumn>: <reltype>``.

    .. py:attribute:: version

       A version control number, used to determine whether the column and its
       metadata should be updated.

    .. py:attribute:: extra

       Arbitrary additional information about this column stored as JSON.

    '''
    __tablename__ = 'obs_column'

    id = Column(Text, primary_key=True) # fully-qualified id like '"us.census.acs".b01001001'

    type = Column(Text, nullable=False) # postgres type, IE numeric, string, geometry, etc.
    name = Column(Text) # human-readable name to provide in bigmetadata

    description = Column(Text) # human-readable description to provide in
                                 # bigmetadata

    weight = Column(Numeric, default=1)
    aggregate = Column(Text) # what aggregate operation to use when adding
                               # these together across geoms: AVG, SUM etc.

    tables = relationship("OBSColumnTable", back_populates="column", cascade="all,delete")
    session = current_session()
    # Passing the same session object to the creator saves thousands
    # of `select 1` queries, because each use of the session was pinging
    tag_creator_with_session = functools.partial(tag_creator, session=session)
    tags = association_proxy('column_column_tags', 'tag', creator=tag_creator_with_session)

    targets = association_proxy('tgts', 'reltype', creator=coltocoltargets_creator)

    version = Column(Numeric, default=0, nullable=False)
    extra = Column(JSON)

    @property
    def sources(self):
        sources = {}
        session = current_session()
        for c2c in session.query(OBSColumnToColumn).filter_by(target_id=self.id):
            sources[c2c.source] = c2c.reltype
        return sources

    @property
    def index_type(self):
        if hasattr(self, '_index_type'):
            pass
        elif 'geom_ref' in list(self.targets.values()):
            self._index_type = 'btree'
        elif self.type.lower() == 'geometry':
            self._index_type = 'gist'
        else:
            self._index_type = None

        return self._index_type

    def children(self):
        children = [col for col, reltype in self.sources.items() if reltype == DENOMINATOR]
        children.sort(key=lambda x: natural_sort_key(x.name))
        return children

    def is_cartographic(self):
        '''
        Returns True if this column is a geometry that can be used for cartography.
        '''
        for tag in self.tags:
            if 'cartographic_boundary' in tag.id:
                return True
        return False

    def is_interpolation(self):
        '''
        Returns True if this column is a geometry that can be used for interpolation.
        '''
        for tag in self.tags:
            if 'interpolation_boundary' in tag.id:
                return True
        return False

    def is_geomref(self):
        '''
        Returns True if the column is a geomref, else Null
        '''
        return GEOM_REF in list(self.targets.values())

    def has_children(self):
        '''
        Returns True if this column has children, False otherwise.
        '''
        return len(self.children()) > 0

    def has_denominators(self):
        '''
        Returns True if this column has no denominator, False otherwise.
        '''
        return DENOMINATOR in list(self.targets.values())

    def has_catalog_image(self):
        '''
        Returns True if this column has a pre-generated image for the catalog.
        '''
        return os.path.exists(os.path.join('catalog', 'img', self.id + '.png'))

    def denominators(self):
        '''
        Return the :class:`~.meta.OBSColumn` denominator of this column.
        '''
        if not self.has_denominators():
            return []
        return [k for k, v in self.targets.items() if v == DENOMINATOR]

    def unit(self):
        '''
        Return the :class:`~.meta.OBSTag` unit of this column.
        '''
        units = [tag for tag in self.tags if tag.type == 'unit']
        if units:
            return units[0]

    def summable(self):
        '''
        Returns True if we can sum this column and calculate for arbitrary
        areas.
        '''
        return self.aggregate == 'sum'

    def catalog_lonlat(self):
        '''
        Return tuple (longitude, latitude) for the catalog for this measurement.
        '''
        return catalog_lonlat(self.id)

    def geom_timespans(self):
        '''
        Return a dict of geom columns and timespans that this measure is
        available for.
        '''
        geom_timespans = {}
        for table in self.tables:
            table = table.table
            geomref_column = table.geomref_column()
            geom_column = [t for t, rt in geomref_column.targets.items()
                           if rt == GEOM_REF][0]
            if geom_column not in geom_timespans:
                geom_timespans[geom_column] = []
            geom_timespans[geom_column].append(table.timespan)
        return geom_timespans

    def source_tags(self):
        '''
        Return source tags.
        '''
        return [tag for tag in self.tags if tag.type.lower() == 'source']

    def license_tags(self):
        '''
        Return license tags.
        '''
        return [tag for tag in self.tags if tag.type.lower() == 'license'] GEOM_REF = 'geom_ref' GEOM_NAME = 'geom_name' class OBSTag(Base):
    '''
    Tags permit arbitrary groupings of columns.

    They should only be created as part of a :meth:`~.tasks.TagsTask.tags`
    implementation.

    .. py:attribute:: id

       The unique identifier for this table.  Is always qualified by
       the module name of the :class:`~.tasks.TagsTask` that created it, and
       should never be specified manually.

    .. py:attribute:: name

       The name of this tag.  This is exposed in the API and user interfaces.

    .. py:attribute:: type

       The type of this tag.  This is used to provide more information about
       what the tag means.  Examples are ``section``, ``subsection``,
       ``license``, ``unit``, although any arbitrary type can be defined.

    .. py:attribute:: description

       Description of this tag.  This may be exposed in the API and catalog.

    .. py:attribute:: columns

       An iterable of all the :class:`~.meta.OBSColumn`s tagged with this tag.

    .. py:attribute:: version

       A version control number, used to determine whether the tag and its
       metadata should be updated.

    '''
    __tablename__ = 'obs_tag'

    id = Column(Text, primary_key=True)

    name = Column(Text, nullable=False)
    type = Column(Text, nullable=False)
    description = Column(Text)

    columns = association_proxy('tag_column_tags', 'column')

    version = Column(Numeric, default=0, nullable=False) def current_session():
    '''
    Returns the session relevant to the currently operating :class:`Task`, if
    any.  Outside the context of a :class:`Task`, this can still be used for
    manual session management.
    '''
    return _current_session.get()
from luigi import Task, Parameter, LocalTarget, WrapperTask
from tasks.base_tasks import (ColumnsTask, TableTask, TagsTask, GeoFile2TempTableTask, SimplifiedTempTableTask,
                              RepoFile)
from lib.timespan import get_timespan
from tasks.util import shell, classpath, copyfile
from tasks.tags import SectionTags, SubsectionTags, BoundaryTags
from tasks.meta import OBSTable, OBSColumn, GEOM_REF, GEOM_NAME, OBSTag, current_session
from collections import OrderedDict
import os
# -*- coding: utf-8 -*-






class DownloadGeometry(Task):

    seq = Parameter()
    # request url: http://centrodedescargas.cnig.es/CentroDescargas/descargaDir
    # arguments:
    # secDescDirLA:114023
    # pagActual:1
    # numTotReg:5
    # codSerieSel:CAANE
    URL = 'http://centrodedescargas.cnig.es/CentroDescargas/descargaDir?secDescDirLA={seq}&pagActual=1&numTotReg=5&codSerieSel=CAANE'

    def requires(self):
        return RepoFile(resource_id=self.task_id, version=self.version(), url=self.URL.format(seq=self.seq))

    def version(self):
        return 1

    def run(self):
        copyfile(self.input().path, '{output}.zip'.format(output=self.output().path))
        shell('unzip -d {output} {output}.zip'.format(output=self.output().path))

    def output(self):
        return LocalTarget(os.path.join('tmp', classpath(self), self.seq).lower())


class ImportGeometry(GeoFile2TempTableTask):

    resolution = Parameter()
    timestamp = Parameter()
    id_aux = Parameter()  # X for Peninsula and Balearic Islands, Y for Canary Islands

    def requires(self):
        return DownloadGeometry(seq='114023')

    def input_files(self):
        '''
        We don't know precise name of file inside zip archive beforehand, so
        use find to track it down.
        '''

        return shell("find '{dirpath}' -iname *_{resolution}_*_{aux}.shp | grep {timestamp}".format(