GEOGRAPHIES = (
    GEO_CT,
    GEO_PR,
    GEO_CD,
    GEO_CSD,
    GEO_CMA,
    GEO_DA,
    GEO_FSA,
    GEO_DB,
) class GeographyColumns(ColumnsTask):

    resolution = Parameter(default=GEO_PR)
    year = IntParameter()

    weights = {
        GEO_PR: 1,
        GEO_CD: 2,
        GEO_CMA: 3,
        GEO_CSD: 4,
        GEO_CT: 5,
        GEO_DA: 6,
        GEO_FSA: 7,
        GEO_DB: 8,
    }

    def version(self):
        return 14

    def requires(self):
        return {
            'sections': SectionTags(),
            'subsections': SubsectionTags(),
            'boundary': BoundaryTags(),
            'license': LicenseTags(),
            'source': SourceTags(),
        }

    def columns(self):
        input_ = self.input()
        sections = input_['sections']
        subsections = input_['subsections']
        boundary_type = input_['boundary']
        data_license = input_['license']['statcan-license']
        data_source = input_['source']['statcan-census-2011']
        geom = OBSColumn(
            id='{resolution}_{year}'.format(resolution=self.resolution, year=self.year),
            type='Geometry',
            name=GEOGRAPHY_NAMES[self.resolution],
            description=GEOGRAPHY_DESCS[self.resolution],
            weight=self.weights[self.resolution],
            tags=[sections['ca'], subsections['boundary'], data_license, data_source],
        )
        geom_id = OBSColumn(
            id='{resolution}_id_{year}'.format(resolution=self.resolution, year=self.year),
            type='Text',
            weight=0,
            targets={geom: GEOM_REF},
        )
        geom_name = OBSColumn(
            id='{resolution}_name_{year}'.format(resolution=self.resolution, year=self.year),
            type='Text',
            weight=1,
            name='Name of ' + GEOGRAPHY_NAMES[self.resolution],
            targets={geom: GEOM_NAME},
            tags=[sections['ca'], subsections['names'], data_license, data_source]
        )
        geom.tags.extend(boundary_type[i] for i in GEOGRAPHY_TAGS[self.resolution])

        return OrderedDict([
            ('geom_name', geom_name),
            ('geom_id', geom_id),   # cvegeo
            ('the_geom', geom),     # the_geom
        ]) class Geography(TableTask):

    resolution = Parameter(default=GEO_PR)
    year = IntParameter()

    def version(self):
        return 10

    def requires(self):
        return {
            'data': SimplifiedImportGeography(resolution=self.resolution, year=self.year),
            'columns': GeographyColumns(resolution=self.resolution, year=self.year)
        }

    def table_timespan(self):
        return get_timespan(self.year)

    # TODO: https://github.com/CartoDB/bigmetadata/issues/435
    def targets(self):
        return {
            OBSTable(id='.'.join([self.schema(), self.name()])): GEOM_REF,
        }

    def columns(self):
        return self.input()['columns']

    def populate(self):
        session = current_session()
        session.execute('INSERT INTO {output} '
                        'SELECT {name} as geom_name, '
                        '       {code} as geom_id, '
                        '       wkb_geometry as geom '
                        'FROM {input} '.format(
                            name=GEOGRAPHY_PROPERNAMES[self.resolution],
                            output=self.output().table,
                            code=GEOGRAPHY_UID[self.resolution],
                            input=self.input()['data'].table)) GEO_CT = 'ct_' GEO_PR = 'pr_' GEO_CD = 'cd_' GEO_CSD = 'csd' GEO_CMA = 'cma' GEO_DA = 'da_' GEO_FSA = 'fsa' GEO_DB = 'db_' class CensusColumns(ColumnsTask):
    resolution = Parameter()
    topic = Parameter()

    def requires(self):
        return {
            'sections': SectionTags(),
            'subsections': SubsectionTags(),
            'units': UnitTags(),
            'license': LicenseTags(),
            'source': SourceTags(),
            'public': PublicTags()
        }

    def version(self):
        return 1

    def columns(self):
        input_ = self.input()

        license_ca = input_['license']['statcan-license']
        source_ca = input_['source']['statcan-census-2016']
        section = input_['sections']['ca']
        subsections = input_['subsections']
        public = input_['public']['public']

        units = input_['units']

        cols = OrderedDict()

        for key, column in COLUMNS_DEFINITION.items():
            tags = [source_ca, section, public, license_ca, subsections[column['subsection']]]
            tags += [units[column.get('units')]] if column.get('units') is not None else []

            aggregate = column.get('aggregate', DEFAULT_AGGREGATE)
            target = UNIVERSE if aggregate in ('median', 'average') else DENOMINATOR

            cols[key + '_t'] = OBSColumn(
                id=column['id'] + '_t',
                name=column['name'],
                description=column.get('description', ''),
                type='Numeric',
                weight=column.get('weight', DEFAULT_WEIGHT),
                aggregate=aggregate,
                targets={cols[denom + '_t']: target for denom in column['denominators']},
                tags=tags
            )
            if column.get('gender_split', 'no').lower() == 'yes':
                for suffix, gender in GENDERS.items():
                    denominators = ['{}_{}'.format(denom, SEGMENT_TOTAL) for denom in column['denominators']] + \
                                   ['{}_{}'.format(denom, suffix) for denom in column['denominators']]

                    cols['{}_{}'.format(key, suffix)] = OBSColumn(
                        id='{}_{}'.format(column['id'], suffix),
                        name='{} ({})'.format(column['name'], gender),
                        description=column.get('description', ''),
                        type='Numeric',
                        weight=column.get('weight', DEFAULT_WEIGHT),
                        aggregate=aggregate,
                        targets={cols[denom]: target for denom in denominators},
                        tags=tags
                    )

        columnsFilter = ColumnsDeclarations(os.path.join(os.path.dirname(__file__), 'census_columns_filter.json'))
        parameters = '{{"resolution":"{resolution}", "topic":"{topic}"}}'.format(
                        resolution=self.resolution, topic=self.topic)
        cols = columnsFilter.filter_columns(cols, parameters)

        return cols COLUMNS_DEFINITION = load_definition() TOPICS = {
    't001': ['segments', 'age_gender', 'families', 'housing'],
    't002': ['education'],
    't003': ['language'],
    't004': ['race_ethnicity'],
    't005': ['nationality', 'migration'],
    't006': ['employment', 'income'],
} SEGMENT_ALL = 'all' SEGMENT_TOTAL = 't' SEGMENT_FEMALE = 'f' SEGMENT_MALE = 'm' class RepoFileUnzipTask(RepoFileUncompressTask):
    '''
    Download a zip file to location {output}.zip and unzip it to the folder
    {output}.
    '''
    compressed_extension = 'zip'

    def uncompress(self):
        uncompress_file(self.output().path) class TempTableTask(Task):
    '''
    A Task that generates a table that will not be referred to in metadata.

    This is useful for intermediate processing steps that can benefit from the
    session guarantees of the ETL, as well as automatic table naming.

    :param force: Optional Boolean, ``False`` by default.  If ``True``, will
                  overwrite output table even if it exists already.
    '''

    force = BoolParameter(default=False, significant=False)

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

    def on_success(self):
        session_commit(self)

    def target_tablename(self):
        '''
        Can be overriden to expose a different table name
        :return:
        '''
        return unqualified_task_id(self.task_id)[:MAX_PG_IDENTIFIER_LENGTH]

    def run(self):
        '''
        Must be overriden by subclass.  Should create and populate a table
        named from ``self.output().table``

        If this completes without exceptions, the :func:`~.util.current_session
        will be committed; if there is an exception, it will be rolled back.
        '''
        raise Exception('Must override `run`')

    def output(self):
        '''
        By default, returns a :class:`~.targets.TableTarget` whose associated
        table lives in a special-purpose schema in Postgres derived using
        :func:`~.util.classpath`.
        '''
        return PostgresTarget(classpath(self), self.target_tablename()) 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 ReverseCoupledInterpolationTask(BaseInterpolationTask):
    '''
    This task interpolates the data for one geography level from the data/geometries from
    another geography level when both layers are coupled and the source geometries are bigger
    than the target ones.
    Calculating the measurements for the target layer is a matter of weighting the measurements
    by the relation between the area of the target geometries and the area of the source ones.
    '''

    def populate(self):
        input_ = self.input()
        interpolation_params = self.get_interpolation_parameters()

        colnames = [x for x in list(self.columns().keys()) if x.lower() != interpolation_params['target_geom_geoid']]

        stmt = '''
                INSERT INTO {output} ({target_data_geoid}, {out_colnames})
                SELECT target_geom_table.{target_geom_geoid}, {ratio_colnames}
                  FROM {source_data_table} source_data_table,
                       {source_geom_table} source_geom_table,
                       {target_geom_table} target_geom_table
                 WHERE source_geom_table.{source_geom_geoid} = source_data_table.{source_data_geoid}
                   AND ST_Intersects(source_geom_table.{source_geom_geomfield},
                                     ST_PointOnSurface(target_geom_table.{target_geom_geomfield}))
               '''.format(
                    output=self.output().table,
                    out_colnames=', '.join(colnames),
                    ratio_colnames=', '.join(['''
                                              ROUND(({x} * (ST_Area(target_geom_table.{geo_t}) /
                                                            ST_Area(source_geom_table.{geo_s})))::numeric, 2) {x}
                                              '''.format(x=x,
                                                         geo_s=interpolation_params['source_geom_geomfield'],
                                                         geo_t=interpolation_params['target_geom_geomfield'])
                                              for x in colnames]),
                    source_data_table=input_['source_data'].table,
                    source_geom_table=input_['source_geom'].table,
                    target_geom_table=input_['target_geom'].table,
                    target_data_geoid=interpolation_params['target_data_geoid'],
                    target_geom_geoid=interpolation_params['target_geom_geoid'],
                    source_data_geoid=interpolation_params['source_data_geoid'],
                    source_geom_geoid=interpolation_params['source_geom_geoid'],
                    source_geom_geomfield=interpolation_params['source_geom_geomfield'],
                    target_geom_geomfield=interpolation_params['target_geom_geomfield'],
                )

        current_session().execute(stmt) 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() GEOM_REF = 'geom_ref' 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 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)
import os
import glob
from collections import OrderedDict
from luigi import Parameter, WrapperTask
from tasks.ca.statcan.geo import (GEOGRAPHIES, GeographyColumns, Geography,
                                  GEO_CT, GEO_PR, GEO_CD, GEO_CSD, GEO_CMA, GEO_DA, GEO_FSA, GEO_DB)
from tasks.ca.statcan.census2016.cols_census import (CensusColumns, COLUMNS_DEFINITION, TOPICS,
                                                     SEGMENT_ALL, SEGMENT_TOTAL, SEGMENT_FEMALE, SEGMENT_MALE)
from tasks.base_tasks import RepoFileUnzipTask, TempTableTask, TableTask, ReverseCoupledInterpolationTask
from tasks.meta import current_session, GEOM_REF
from lib.logger import get_logger
from lib.timespan import get_timespan


LOGGER = get_logger(__name__)

GEOGRAPHY_CODES = {
    GEO_PR: '044',
    GEO_CD: '044',
    GEO_CSD: '044',
    GEO_DA: '044',
    GEO_CMA: '043',
    GEO_CT: '043',
    GEO_FSA: '046',
}

GEOLEVEL_FROM_GEOGRAPHY = {
    GEO_PR: '1',
    GEO_CD: '2',
    GEO_CSD: '3',
    GEO_DA: '4',
    GEO_CMA: '1',
    GEO_CT: '2',
    GEO_FSA: '2',
}

# https://www12.statcan.gc.ca/census-recensement/2016/dp-pd/prof/details/download-telecharger/comp/page_dl-tc.cfm?Lang=E
URL = 'https://www12.statcan.gc.ca/census-recensement/2016/dp-pd/prof/details/download-telecharger/comp/GetFile.cfm?Lang=E&FILETYPE=CSV&GEONO={geo_code}'
NUM_MEASUREMENTS = 2247


def safe_float_cast(val):
    try:
        return float(val)
    except (ValueError, TypeError):
        return 'x'


class DownloadData(RepoFileUnzipTask):
    geocode = Parameter()

    def get_url(self):
        return URL.format(geo_code=self.geocode)


class CopyData(TempTableTask):
    geocode = Parameter()

    def requires(self):
        return DownloadData(geocode=self.geocode)

    def _create_table(self, session):
        query = '''
                CREATE TABLE {output} (
                    year TEXT,
                    geocode TEXT,
                    geolevel TEXT,
                    geoname TEXT,
                    gnr TEXT,
                    gnrlf TEXT,
                    quality TEXT,
                    {csdtypename}
                    altgeocode TEXT,
                    name TEXT,
                    profileid NUMERIC,
                    profiledesc TEXT,
                    total TEXT,
                    male TEXT,
                    female TEXT,
                    PRIMARY KEY(geocode, geolevel, profileid)
                )
                '''.format(
                    output=self.output().table,
                    csdtypename='csdtypename TEXT,' if self.geocode == '044' else ''
                )

        LOGGER.debug(query)
        session.execute(query)
        session.commit()

    def run(self):
        session = current_session()

        self._create_table(session)

        path = glob.glob(os.path.join(self.input().path, '*data.csv'))[0]
        with open(path, 'rb') as f:
            cursor = session.connection().connection.cursor()
            sql_copy = """
                        COPY {table} FROM STDIN WITH CSV HEADER DELIMITER AS ','
                       """.format(table=self.output().table)
            cursor.copy_expert(sql=sql_copy, file=f)
            session.commit()
            cursor.close()


class ImportData(TempTableTask):
    resolution = Parameter()
    topic = Parameter()
    segment = Parameter()

    def requires(self):
        return {
            'data': CopyData(geocode=GEOGRAPHY_CODES[self.resolution]),
        }

    def _create_table(self):
        cols = []
        for key, column in COLUMNS_DEFINITION.items():
            if column['subsection'] in TOPICS[self.topic]:

                if self.segment in [SEGMENT_ALL, SEGMENT_TOTAL]: