criar_uso_solo.py

def handle(self, *args, **options):
        val = options['município/estado/ano']
        slug = slugify(val.split('/')[0])
        sigla_uf = val.split('/')[1].strip().upper()
        ano = int(val.split('/')[2].strip())
        municipio = Municipio.objects.get(slug=slug, sigla_uf=sigla_uf)

        SnciUsoOcupacao.objects.filter(snci__municipio=municipio.cod_ibge_m, ano=ano).delete()
        SigefUsoOcupacao.objects.filter(sigef__municipio=municipio.cod_ibge_m, ano=ano).delete()

        municipio_uso_ocupacao = MunicipioUsoOcupacao.objects.filter(cod_ibge_m=municipio.cod_ibge_m)
        data = [{'val': obj.val, 'desc': obj.desc, 'geom': obj.geom.wkt} for obj in municipio_uso_ocupacao if obj.geom]
        df = pd.DataFrame(data)
        df_uso_ocupacao = gpd.GeoDataFrame(df, geometry=df['geom'].apply(wkt.loads), crs=4326)
        del df_uso_ocupacao['geom']

        val_list = df_uso_ocupacao['val'].unique()
        desc_list = df_uso_ocupacao['desc'].unique()

        classes_dict = {}
        for val, desc in zip(val_list, desc_list):
            classes_dict[val] = desc

        municipio_incra_sigef = IncraSigef.objects.filter(municipio=municipio.cod_ibge_m)
        data = [
            {'sigef_id': obj.id, 'geom': obj.geom.wkt} for obj in municipio_incra_sigef if obj.geom
        ]
        df = pd.DataFrame(data)
        df_sigef = gpd.GeoDataFrame(df, geometry=df['geom'].apply(wkt.loads), crs=4326)
        del df_sigef['geom']

        df_sigef_clip = df_uso_ocupacao.overlay(df_sigef, how='intersection')
        df_sigef_clip = df_sigef_clip.to_crs(5880)
        df_sigef_clip['geometry'] = df_sigef_clip['geometry'].to_crs(5880)
        df_sigef_clip['area_ha'] = df_sigef_clip['geometry'].area / 10000

        df_sigef_clip = df_sigef_clip.to_crs(4326)
        df_sigef_clip['geometry'] = df_sigef_clip['geometry'].to_crs(4326)

        print('Criando Uso e Ocupação para os SIGEF...')
        sigef_list = []
        for idx, row in df_sigef_clip.iterrows():
            sigef = IncraSigef.objects.get(pk=row['sigef_id'])

            geom = GEOSGeometry(row['geometry'].wkt)
            if geom.geom_type == 'Polygon':
                geom = MultiPolygon(geom)

            sigef_list.append(
                SigefUsoOcupacao(
                    sigef=sigef,
                    val=row['val'],
                    desc=classes_dict[row['val']],
                    area=row['area_ha'],
                    ano=ano,
                    geom=geom
                )
            )

        SigefUsoOcupacao.objects.bulk_create(sigef_list)

        municipio_incra_snci = IncraSnci.objects.filter(municipio=municipio.cod_ibge_m)

        data = [
            {'snci_id': obj.id, 'geom': obj.geom.wkt} for obj in municipio_incra_snci if obj.geom
        ]
        df = pd.DataFrame(data)
        df_snci = gpd.GeoDataFrame(df, geometry=df['geom'].apply(wkt.loads), crs=4326)
        del df_snci['geom']

        df_snci_clip = df_uso_ocupacao.overlay(df_snci, how='intersection')
        df_snci_clip = df_snci_clip.to_crs(5880)
        df_snci_clip['geometry'] = df_snci_clip['geometry'].to_crs(5880)
        df_snci_clip['area_ha'] = df_snci_clip['geometry'].area / 10000

        df_snci_clip = df_snci_clip.to_crs(4326)
        df_snci_clip['geometry'] = df_snci_clip['geometry'].to_crs(4326)

        print('Criando Uso e Ocupação para os SNCI...')
        snci_list = []
        for idx, row in df_snci_clip.iterrows():
            snci = IncraSnci.objects.get(pk=row['snci_id'])

            geom = GEOSGeometry(row['geometry'].wkt)
            if geom.geom_type == 'Polygon':
                geom = MultiPolygon(geom)

            snci_list.append(
                SnciUsoOcupacao(
                    snci=snci,
                    val=row['val'],
                    desc=classes_dict[row['val']],
                    area=row['area_ha'],
                    ano=ano,
                    geom=geom
                )
            )

        SnciUsoOcupacao.objects.bulk_create(snci_list)

        print('Uso e Ocupação Finalizado com Sucesso!')