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_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) 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 InterpolationTask(BaseInterpolationTask):
    '''
    This task interpolates the data for one geography level from the data/geometries from
    another geography level by computing the intersections and area relations between the
    geometries.
    '''

    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_geoid}, {sum_colnames}
                  FROM (
                    SELECT CASE WHEN ST_Within(target_geom_table.{target_geom_geomfield},
                                               source_geom_table.{source_geom_geomfield})
                                     THEN ST_Area(target_geom_table.{target_geom_geomfield}) /
                                          Nullif(ST_Area(source_geom_table.{source_geom_geomfield}), 0)
                                WHEN ST_Within(source_geom_table.{source_geom_geomfield},
                                               target_geom_table.{target_geom_geomfield})
                                     THEN 1
                                ELSE ST_Area(ST_Intersection(source_geom_table.{source_geom_geomfield},
                                                             target_geom_table.{target_geom_geomfield})) /
                                             Nullif(ST_Area(source_geom_table.{source_geom_geomfield}), 0)
                           END area_ratio,
                           target_geom_table.{target_geom_geoid}, {in_colnames}
                      FROM {source_data_table} source_data_table,
                           {source_geom_table} source_geom_table,
                           {target_geom_table} target_geom_table
                     WHERE source_data_table.{source_data_geoid} = source_geom_table.{source_geom_geoid}
                       AND ST_Intersects(source_geom_table.{source_geom_geomfield},
                                         target_geom_table.{target_geom_geomfield}) = True
                    ) q GROUP BY {target_geom_geoid}
               '''.format(
                    output=self.output().table,
                    sum_colnames=', '.join(['round(sum({x} * Nullif(area_ratio, 0))) {x}'.format(x=x) for x in colnames]),
                    out_colnames=', '.join(colnames),
                    in_colnames=', '.join(colnames),
                    source_data_table=input_['source_data'].table,
                    source_geom_table=input_['source_geom'].table,
                    target_geom_table=input_['target_geom'].table,
                    source_data_geoid=interpolation_params['source_data_geoid'],
                    source_geom_geoid=interpolation_params['source_geom_geoid'],
                    target_data_geoid=interpolation_params['target_data_geoid'],
                    target_geom_geoid=interpolation_params['target_geom_geoid'],
                    source_geom_geomfield=interpolation_params['source_geom_geomfield'],
                    target_geom_geomfield=interpolation_params['target_geom_geomfield'],
                )

        current_session().execute(stmt) class CoupledInterpolationTask(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 target geometries are bigger
    than the source ones.
    Calculating the measurements for the target layer is a matter of adding up the values
    from the source layer.
    '''

    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}, {sum_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(target_geom_table.{target_geom_geomfield},
                                     ST_PointOnSurface(source_geom_table.{source_geom_geomfield}))
                 GROUP BY target_geom_table.{target_geom_geoid}
               '''.format(
                    output=self.output().table,
                    out_colnames=', '.join(colnames),
                    sum_colnames=', '.join(['sum({x}) {x}'.format(x=x) 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) class OutputAreas(TableTask):

    def requires(self):
        return {
            'geom_columns': OutputAreaColumns(),
            'data': SimplifiedImportOutputAreas(),
        }

    def version(self):
        return 9

    def table_timespan(self):
        return get_timespan('2011')

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

    def columns(self):
        input_ = self.input()
        cols = OrderedDict()
        cols.update(input_['geom_columns'])
        return cols

    def populate(self):
        session = current_session()
        session.execute('INSERT INTO {output} '
                        'SELECT ST_MakeValid(wkb_geometry), {geoname_column} '
                        'FROM {input}'.format(
                            output=self.output().table,
                            input=self.input()['data'].table,
                            geoname_column=self.geoname_column()
                        ))

    @staticmethod
    def geoid_column():
        return OutputAreaColumns.geoid_column()

    @staticmethod
    def geoname_column():
        return OutputAreaColumns.geoname_column() class OutputAreaColumns(ColumnsTask):

    def version(self):
        return 4

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

    def columns(self):
        input_ = self.input()
        license = input_['license']['uk_ogl']
        source = input_['source']['cdrc-source']
        boundary_type = input_['boundary']
        geom = OBSColumn(
            type='Geometry',
            name='Census Output Areas',
            description='The smallest unit for which census data are published '
                        'in the UK.  They contain at least 40 households and '
                        '100 persons, the target size being 125 households. '
                        'They were built up from postcode blocks after the '
                        'census data were available, with the intention of '
                        'standardising population sizes, geographical shape '
                        'and social homogeneity (in terms of dwelling types '
                        'and housing tenure). The OAs generated in 2001 were '
                        'retained as far as possible for the publication of '
                        'outputs from the 2011 Census (less than 3% were '
                        'changed). -`Wikipedia <https://en.wikipedia.org/'
                        'wiki/ONS_coding_system#Geography_of_the_UK_Census>`_',
            weight=8,
            tags=[input_['subsections']['boundary'], input_['sections']['uk'], source, license,
                  boundary_type['cartographic_boundary'], boundary_type['interpolation_boundary']]
        )
        geomref = OBSColumn(
            type='Text',
            name='Census Output Area ID',
            weight=0,
            targets={geom: GEOM_REF}
        )

        return OrderedDict([
            ('the_geom', geom),
            (self.geoname_column(), geomref)
        ])

    @staticmethod
    def geoname_column():
        return 'oa_sa'

    @staticmethod
    def geoid_column():
        return 'oa_sa' class CensusColumns(ColumnsTask):
    def requires(self):
        return {
            'units': UnitTags(),
            'sections': SectionTags(),
            'subsections': SubsectionTags(),
            'source': SourceTags(),
            'license': LicenseTags(),
        }

    def version(self):
        return 5

    def prefix(self):
        return 'uk.ons'

    def columns(self):
        input_ = self.input()
        sources = list(input_['source'].values())
        license_tag = input_['license']['uk_ogl']
        uk = input_['sections']['uk']
        subsections = input_['subsections']
        units = input_['units']

        columns = OrderedDict()
        for key, column in COLUMNS_DEFINITION.items():
            columns[key] = OBSColumn(
                id=column['id'],
                name=column['name'],
                description='',
                type='Numeric',
                weight=column['weight'],
                aggregate='sum',
                targets={columns[denom]: DENOMINATOR for denom in column['denominators']},
                tags=sources + [uk, license_tag, units[column['units']], subsections[column['subsection']]]
            )

        return columns class PostcodeAreas(TableTask):

    def requires(self):
        return {
            'geom_columns': PostcodeAreasColumns(),
            'data': SimplifiedUniquePostcodeAreas(),
        }

    def version(self):
        return 2

    def table_timespan(self):
        return get_timespan('2017')

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

    def columns(self):
        input_ = self.input()
        cols = OrderedDict()
        cols.update(input_['geom_columns'])
        return cols

    def populate(self):
        session = current_session()
        session.execute('INSERT INTO {output} '
                        'SELECT ST_MakeValid(wkb_geometry) as the_geom, label pa_id, name '
                        'FROM {input}'.format(
                            output=self.output().table,
                            input=self.input()['data'].table,
                        ))

    @staticmethod
    def geoid_column():
        return PostcodeAreasColumns.geoid_column()

    @staticmethod
    def geoname_column():
        return PostcodeAreasColumns.geoname_column() class PostcodeAreasColumns(ColumnsTask):

    def version(self):
        return 1

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

    def columns(self):
        input_ = self.input()
        license = input_['license']['uk_ed_datashare']
        source = input_['source']['uk_ed_datashare_source']
        boundary_type = input_['boundary']
        geom = OBSColumn(
            id='pa_geo',
            type='Geometry',
            name='GB Postcode Areas',
            description='''
                        Dataset contains polygons corresponding to the top level (Postcode Area) of the GB Postcode Hierarchy.
                        Thiessen polygons were generated from all records in OS OpenData Code-Point Open.
                        Resulting coverage was dissolved up to postcode area level and then clipped to GB extent of the realm using OS OpenData Boundary-Line.
                        Names of postcode areas were added from Wikipedia. GIS vector data.
                        This dataset was first accessioned in the EDINA ShareGeo Open repository on 2010-07-22 and migrated to Edinburgh DataShare on 2017-02-21.
                        ''',
            weight=8,
            tags=[input_['subsections']['boundary'], input_['sections']['uk'], source, license,
                  boundary_type['cartographic_boundary'], boundary_type['interpolation_boundary']]
        )
        geomref = OBSColumn(
            id='pa_id',
            type='Text',
            name='GB Postcode Area ID',
            weight=0,
            targets={geom: GEOM_REF}
        )
        geomname = OBSColumn(
            id='pa_name',
            type='Text',
            name='GB Postcode Area Name',
            weight=1,
            targets={geom: GEOM_REF},
            tags=[input_['sections']['uk'], input_['subsections']['boundary'], source, license]
        )

        return OrderedDict([
            ('the_geom', geom),
            ('pa_id', geomref),
            ('name', geomname),
        ])

    @staticmethod
    def geoname_column():
        return 'name'

    @staticmethod
    def geoid_column():
        return 'pa_id' class PostcodeDistricts(TableTask):

    def requires(self):
        return {
            'columns': PostcodeDistrictsColumns(),
            'data': SimplifiedPostcodeDistricts(),
        }

    def version(self):
        return 1

    def table_timespan(self):
        return get_timespan('2015')

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

    def columns(self):
        input_ = self.input()
        cols = OrderedDict()
        cols.update(input_['columns'])
        return cols

    def populate(self):
        session = current_session()

        query = '''
                INSERT INTO {output}
                SELECT ST_MakeValid(wkb_geometry), name, name
                FROM {input}
                '''.format(output=self.output().table,
                           input=self.input()['data'].table)

        session.execute(query)

    @staticmethod
    def geoid_column():
        return PostcodeDistrictsColumns.geoid_column()

    @staticmethod
    def geoname_column():
        return PostcodeDistrictsColumns.geoname_column() class PostcodeDistrictsColumns(ColumnsTask):

    def version(self):
        return 1

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

    def columns(self):
        input_ = self.input()
        license = input_['license']['uk_ogl']
        source = input_['source']['uk_os_opendata_source']
        boundary_type = input_['boundary']
        geom = OBSColumn(
            id='pd_geo',
            type='Geometry',
            name='Reconstructed UK Postcode Districts',
            description='Reconstructed UK Postcode Districts',
            weight=8,
            tags=[input_['subsections']['boundary'], input_['sections']['uk'], source, license,
                  boundary_type['cartographic_boundary'], boundary_type['interpolation_boundary']]
        )
        geomref = OBSColumn(
            id='pd_id',
            type='Text',
            name='Postcode District ID',
            weight=0,
            targets={geom: GEOM_REF},
        )
        geomname = OBSColumn(
            id='pd_name',
            type='Text',
            name='Postcode District ID',
            weight=1,
            targets={geom: GEOM_REF},
            tags=[input_['sections']['uk'], input_['subsections']['boundary'], source, license]
        )

        return OrderedDict([
            ('the_geom', geom),
            ('geographycode', geomref),
            ('name', geomname)
        ])

    @staticmethod
    def geoname_column():
        return 'geographycode'

    @staticmethod
    def geoid_column():
        return 'geographycode' class PostcodeSectors(TableTask):

    def requires(self):
        return {
            'columns': PostcodeSectorsColumns(),
            'data': SimplifiedPostcodeSectors(),
        }

    def version(self):
        return 1

    def table_timespan(self):
        return get_timespan('2015')

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

    def columns(self):
        input_ = self.input()
        cols = OrderedDict()
        cols.update(input_['columns'])
        return cols

    def populate(self):
        session = current_session()

        query = '''
                INSERT INTO {output}
                SELECT ST_MakeValid(wkb_geometry), name, name
                FROM {input}
                '''.format(output=self.output().table,
                           input=self.input()['data'].table)

        session.execute(query)

    @staticmethod
    def geoid_column():
        return PostcodeSectorsColumns.geoid_column()

    @staticmethod
    def geoname_column():
        return PostcodeSectorsColumns.geoname_column() class PostcodeSectorsColumns(ColumnsTask):

    def version(self):
        return 1

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

    def columns(self):
        input_ = self.input()
        license = input_['license']['uk_ogl']
        source = input_['source']['uk_os_opendata_source']
        boundary_type = input_['boundary']
        geom = OBSColumn(
            id='ps_geo',
            type='Geometry',
            name='Reconstructed UK Postcode Sectors',
            description='Reconstructed UK Postcode Sectors',
            weight=8,
            tags=[input_['subsections']['boundary'], input_['sections']['uk'], source, license,
                  boundary_type['cartographic_boundary'], boundary_type['interpolation_boundary']]
        )
        geomref = OBSColumn(
            id='ps_id',
            type='Text',
            name='Postcode Sector ID',
            weight=0,
            targets={geom: GEOM_REF},
        )
        geomname = OBSColumn(
            id='ps_name',
            type='Text',
            name='Postcode Sector ID',
            weight=1,
            targets={geom: GEOM_REF},
            tags=[input_['sections']['uk'], input_['subsections']['boundary'], source, license]
        )

        return OrderedDict([
            ('the_geom', geom),
            (self.geoid_column(), geomref),
            (self.geoname_column(), geomname)
        ])

    @staticmethod
    def geoname_column():
        return 'geographyname'

    @staticmethod
    def geoid_column():
        return 'geographycode' class LowerLayerSuperOutputAreas(TableTask):

    def requires(self):
        return {
            'columns': LowerLayerSuperOutputAreasColumns(),
            'lsoa-data': SimplifiedImportLowerLayerSuperOutputAreas(),
            'dz-data': SimplifiedImportDataZones(),
            'ni-data': SimplifiedImportNISuperOutputAreas(),
        }

    def version(self):
        return 1

    def table_timespan(self):
        return get_timespan('2011')

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

    def columns(self):
        input_ = self.input()
        cols = OrderedDict()
        cols.update(input_['columns'])
        return cols

    def populate(self):
        session = current_session()

        query = '''
                INSERT INTO {output}
                SELECT ST_MakeValid(wkb_geometry), lsoa11cd, lsoa11nm
                FROM {input_lsoa}
                UNION ALL
                SELECT ST_MakeValid(wkb_geometry), datazone, name
                FROM {input_dz}
                UNION ALL
                SELECT ST_MakeValid(wkb_geometry), soa_code, soa_label
                FROM {input_ni}
                '''.format(output=self.output().table,
                           input_lsoa=self.input()['lsoa-data'].table,
                           input_dz=self.input()['dz-data'].table,
                           input_ni=self.input()['ni-data'].table,)

        session.execute(query)

    @staticmethod
    def geoid_column():
        return LowerLayerSuperOutputAreasColumns.geoid_column()

    @staticmethod
    def geoname_column():
        return LowerLayerSuperOutputAreasColumns.geoname_column() class LowerLayerSuperOutputAreasColumns(ColumnsTask):

    def version(self):
        return 1

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

    def columns(self):
        input_ = self.input()
        license = input_['license']['uk_ogl']
        source = input_['source']['uk-gov-source']
        boundary_type = input_['boundary']
        geom = OBSColumn(
            id='lsoa_geo',
            type='Geometry',
            name='Lower Layer Super Output Areas (LSOA) and Data Zones',
            description='Full Clipped Lower Layer Super Output Area Boundaries '
                        'in England and Wales. '
                        'Data Zone Boundaries in Scotland. '
                        'Lower Layer Super Output Areas (LSOAs) typically contain '
                        '4 to 6 OAs with a population of around 1500. '
                        '-`Wikipedia <https://en.wikipedia.org/'
                        'wiki/ONS_coding_system#Neighbourhood_Statistics_Geography>`_'
                        'Data zones are the key geography for the dissemination '
                        'of small area statistics in Scotland and are widely used '
                        'across the public and private sector. '
                        'Composed of aggregates of Census Output Areas, data zones '
                        'are large enough that statistics can be presented accurately '
                        'without fear of disclosure and yet small enough that they '
                        'can be used to represent communities. They are designed to '
                        'have roughly standard populations of 500 to 1,000 household '
                        'residents, nest within Local Authorities, have compact '
                        'shapes that respect physical boundaries where possible, '
                        'and to contain households with similar social characteristics. '
                        'Aggregations of data zones are often used to approximate a '
                        'larger area of interest or a higher level geography that '
                        'statistics wouldn’t normally be available for. '
                        'Data zones also represent a relatively stable geography '
                        'that can be used to analyse change over time, '
                        'with changes only occurring after a Census. '
                        'Following the update to data zones using 2011 Census data, '
                        'there are now 6,976 data zones covering the whole of Scotland. '
                        '-`data.gov.uk <https://data.gov.uk/dataset/'
                        'ab9f1f20-3b7f-4efa-9bd2-239acf63b540/data-zone-boundaries-2011>`_'
                        'Super Output Areas (SOAs) were a new geography that were developed '
                        'NISRA to improve the reporting of small area statistics. '
                        'A set of slightly revised Super Output Areas (SOAs) were created '
                        'for the 2011 census outputs. '
                        '-`Northern Ireland Statistics and Research Agency <https://www.nisra.gov.uk/'
                        'support/geography/northern-ireland-super-output-areas>`_',
            weight=8,
            tags=[input_['subsections']['boundary'], input_['sections']['uk'], source, license,
                  boundary_type['cartographic_boundary'], boundary_type['interpolation_boundary']]
        )
        geomref = OBSColumn(
            id='lsoa_id',
            type='Text',
            name='Lower Layer Super Output Area ID',
            weight=0,
            targets={geom: GEOM_REF}
        )
        geomname = OBSColumn(
            id='lsoa_name',
            type='Text',
            name='Lower Layer Super Output Area Name',
            weight=0,
            targets={geom: GEOM_REF}
        )

        return OrderedDict([
            ('the_geom', geom),
            ('geographycode', geomref),
            ('name', geomname),
        ])

    @staticmethod
    def geoname_column():
        return 'name'

    @staticmethod
    def geoid_column():
        return 'geographycode' class MiddleLayerSuperOutputAreas(TableTask):

    def requires(self):
        return {
            'columns': MiddleLayerSuperOutputAreasColumns(),
            'msoa-data': SimplifiedImportMiddleLayerSuperOutputAreas(),
            'iz-data': SimplifiedImportIntermediateZones(),
        }

    def version(self):
        return 1

    def table_timespan(self):
        return get_timespan('2011')

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

    def columns(self):
        input_ = self.input()
        cols = OrderedDict()
        cols.update(input_['columns'])
        return cols

    def populate(self):
        session = current_session()

        query = '''
                INSERT INTO {output}
                SELECT ST_MakeValid(wkb_geometry), msoa11cd, msoa11nm
                FROM {input_msoa}
                UNION ALL
                SELECT ST_MakeValid(wkb_geometry), interzone, name
                FROM {input_iz}
                '''.format(output=self.output().table,
                           input_msoa=self.input()['msoa-data'].table,
                           input_iz=self.input()['iz-data'].table,)

        session.execute(query)

    @staticmethod
    def geoid_column():
        return MiddleLayerSuperOutputAreasColumns.geoid_column()

    @staticmethod
    def geoname_column():
        return MiddleLayerSuperOutputAreasColumns.geoname_column() class MiddleLayerSuperOutputAreasColumns(ColumnsTask):

    def version(self):
        return 1

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

    def columns(self):
        input_ = self.input()
        license = input_['license']['uk_ogl']
        source = input_['source']['uk-gov-source']
        boundary_type = input_['boundary']
        geom = OBSColumn(
            id='msoa_geo',
            type='Geometry',
            name='Middle Layer Super Output Areas (LSOA) and Intermediate Zones',
            description='Full Clipped Middle Layer Super Output Area Boundaries '
                        'in England and Wales. '
                        'Middle Layer Super Output Areas (MSOAs) on average '
                        'have a population of 7,200. '
                        '-`Wikipedia <https://en.wikipedia.org/'
                        'wiki/ONS_coding_system#Neighbourhood_Statistics_Geography>`_'
                        'Intermediate zones are a statistical geography that sit '
                        'between data zones and local authorities, created for use '
                        'with the Scottish Neighbourhood Statistics (SNS) programme '
                        'and the wider public sector. Intermediate zones are used '
                        'for the dissemination of statistics that are not suitable '
                        'for release at the data zone level because of the sensitive '
                        'nature of the statistic, or for reasons of reliability. '
                        'Intermediate Zones were designed to meet constraints on '
                        'population thresholds (2,500 - 6,000 household residents), '
                        'to nest within local authorities, and to be built up from '
                        'aggregates of data zones. Intermediate zones also represent '
                        'a relatively stable geography that can be used to analyse '
                        'change over time, with changes only occurring after a Census. '
                        'Following the update to intermediate zones using 2011 Census '
                        'data, there are now 1,279 Intermediate Zones covering the '
                        'whole of Scotland. '
                        '-`data.gov.uk <https://data.gov.uk/dataset/'
                        '133d4983-c57d-4ded-bc59-390c962ea280/intermediate-zone-boundaries-2011>`_',
            weight=8,
            tags=[input_['subsections']['boundary'], input_['sections']['uk'], source, license,
                  boundary_type['cartographic_boundary'], boundary_type['interpolation_boundary']]
        )
        geomref = OBSColumn(
            id='msoa_id',
            type='Text',
            name='Middle Layer Super Output Area ID',
            weight=0,
            targets={geom: GEOM_REF}
        )
        geomname = OBSColumn(
            id='msoa_name',
            type='Text',
            name='Middle Layer Super Output Area Name',
            weight=0,
            targets={geom: GEOM_REF}
        )

        return OrderedDict([
            ('the_geom', geom),
            ('geographycode', geomref),
            ('name', geomname),
        ])

    @staticmethod
    def geoname_column():
        return 'name'

    @staticmethod
    def geoid_column():
        return 'geographycode' 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 class ImportUKOutputAreas(ImportUK):
    def requires(self):
        return DownloadUKOutputAreas(self.table) class ImportEnglandWalesLocal(TempTableTask):
    table = Parameter()

    def requires(self):
        return DownloadEnglandWalesLocal()

    @staticmethod
    def id_to_column(colid):
        return colid

    def run(self):
        infile = os.path.join(self.input().path, self.table + 'DATA.CSV')
        cols = OrderedDict({'geographycode': 'TEXT PRIMARY KEY'})
        with open(infile) as csvfile:
            reader = csv.reader(csvfile)
            header = next(reader)

            for c in header[1:]:
                cols[self.id_to_column(c)] = 'NUMERIC'

        with open(infile) as csvfile:
            copy_from_csv(
                current_session(),
                self.output().table,
                cols,
                csvfile
            ) class ImportScotland(TempTableTask):
    table = Parameter()

    def requires(self):
        return DownloadScotlandLocal()

    @staticmethod
    def id_to_column(colid):
        return sanitize_identifier(colid)

    def run(self):
        infile = os.path.join(self.input().path, self.table + '.csv')

        cols = OrderedDict({'geographycode': 'TEXT PRIMARY KEY'})
        with open(infile) as csvfile:
            reader = csv.reader(csvfile)
            header = next(reader)

            for c in header[1:]:
                cols[self.id_to_column(c)] = 'NUMERIC'

        with open(infile) as csvfile:
            copy_from_csv(
                current_session(),
                self.output().table,
                cols,
                CSVNormalizerStream(csvfile, lambda row: ['0' if f == '-' else f.replace(',', '') for f in row])
            ) def parse_table(table_id):
    '''
    Returns a tuple like (KS, NI) from a table string like KS201NI
    '''
    return TABLE_CODE_PARTS_RE.match(table_id).groups() COLUMNS_DEFINITION = load_definition()
from collections import OrderedDict, defaultdict
from luigi import WrapperTask, Parameter
from tasks.meta import current_session, GEOM_REF
from lib.timespan import get_timespan
from tasks.base_tasks import TableTask, InterpolationTask, CoupledInterpolationTask
from tasks.uk.cdrc import OutputAreas, OutputAreaColumns
from tasks.uk.census.metadata import CensusColumns
from tasks.uk.datashare import PostcodeAreas, PostcodeAreasColumns
from tasks.uk.odl import PostcodeDistricts, PostcodeDistrictsColumns, PostcodeSectors, PostcodeSectorsColumns
from tasks.uk.gov import (LowerLayerSuperOutputAreas, LowerLayerSuperOutputAreasColumns,
                          MiddleLayerSuperOutputAreas, MiddleLayerSuperOutputAreasColumns)
from lib.logger import get_logger
from .ons import ImportUKOutputAreas, ImportEnglandWalesLocal
from .scotland import ImportScotland
from .metadata import parse_table, COLUMNS_DEFINITION
        from_tables = set()  # Set of all source tables / CTEs (FROM generation)
        ctes = defaultdict(lambda: defaultdict(list))  # CTEs to be generated. {t1t2: { t1: [c1, c2], t2: [d1, d2]}}

        # Generate SQL parts for each column
        for k, v in COLUMNS_DEFINITION.items():
            data = v['data']
            if len(data) == 1:
                # Single table source
                table, col_expression = table_column_expression(data[0])

                from_tables.add(table)
                in_colnames.append(col_expression)
                out_colnames.append(k)
            else:
                # Multi data source
                cte_name = '_'.join([d['table'] for d in data]).lower()
                cte = ctes[cte_name]
                for d in data:
                    table, col_expression = table_column_expression(d)
                    cte[table].append('{expression} AS {id}'.format(expression=col_expression, id=k))

                in_colnames.append('{cte_name}.{id}'.format(cte_name=cte_name, id=k))
                out_colnames.append(k)

        # Generate SQL for CTEs
        ctes_sql = []
        for name, cte in ctes.items():
            selects = []
            for table, columns in cte.items():
                selects.append('SELECT geographycode, {cols} FROM {table}'.format(cols=', '.join(columns), table=table))
            ctes_sql.append('{name} AS ({union})'.format(name=name, union=' UNION '.join(selects)))
            from_tables.add(name)

        # Generate FROM clause. Uses FULL JOIN because not all tables are complete.
        tables = list(from_tables)
        from_part = tables[0]
        for t in tables[1:]:
            from_part += ' FULL JOIN {} USING (geographycode)'.format(t)

        stmt = 'WITH {ctes} INSERT INTO {output} (geographycode, {out_colnames}) ' \
               'SELECT geographycode, {in_colnames} ' \
               'FROM {from_part} ' \
               'WHERE geographycode LIKE \'_00%\''.format(
                   ctes=', '.join(ctes_sql),
                   output=self.output().table,
                   out_colnames=', '.join(out_colnames),
                   in_colnames=', '.join(in_colnames),
                   from_part=from_part)

        current_session().execute(stmt)


class CensusPostcodeAreas(InterpolationTask):
    def table_timespan(self):
        return get_timespan('2011')

    def columns(self):
        cols = OrderedDict()
        input_ = self.input()
        cols['pa_id'] = input_['target_geom_columns']['pa_id']
        cols.update(input_['source_data_columns'])
        return cols

    def requires(self):
        deps = {
            'source_geom_columns': OutputAreaColumns(),
            'source_geom': OutputAreas(),
            'source_data_columns': CensusColumns(),
            'source_data': CensusOutputAreas(),
            'target_geom_columns': PostcodeAreasColumns(),
            'target_geom': PostcodeAreas(),
            'target_data_columns': CensusColumns(),
        }

        return deps

    def get_interpolation_parameters(self):
        params = {
            'source_data_geoid': 'geographycode',
            'source_geom_geoid': 'oa_sa',
            'target_data_geoid': 'pa_id',
            'target_geom_geoid': 'pa_id',
            'source_geom_geomfield': 'the_geom',
            'target_geom_geomfield': 'the_geom',
        }

        return params


class CensusPostcodeEntitiesFromOAs(InterpolationTask):
    def table_timespan(self):
        return get_timespan('2011')

    def columns(self):
        cols = OrderedDict()
        input_ = self.input()
        cols['GeographyCode'] = input_['target_geom_columns']['geographycode']
        cols.update(input_['target_data_columns'])
        return cols

    def get_interpolation_parameters(self):
        params = {
            'source_data_geoid': 'geographycode',
            'source_geom_geoid': 'oa_sa',
            'target_data_geoid': 'geographycode',
            'target_geom_geoid': 'geographycode',
            'source_geom_geomfield': 'the_geom',
            'target_geom_geomfield': 'the_geom',
        }

        return params


class CensusPostcodeDistricts(CensusPostcodeEntitiesFromOAs):
    def requires(self):
        deps = {
            'source_geom_columns': OutputAreaColumns(),
            'source_geom': OutputAreas(),
            'source_data_columns': CensusColumns(),
            'source_data': CensusOutputAreas(),

            'target_geom_columns': PostcodeDistrictsColumns(),