<jupyter_start><jupyter_text>Provide geometries from CARTO with geostore IDsThis script reads a CARTO table provides each feature with a geostore ID and writes back the table to CARTO. Tables to read:- [River basins - level 3](https://resourcewatch.carto.com/u/wri-rw/tables/wat_068_rw0_watersheds_edit/public?redirected=true) `SELECT * FROM "wri-rw".wat_068_rw0_watersheds_edit WHERE level = 3`- [Country geometries - coastal](https://resourcewatch.carto.com/u/wri-rw/dataset/gadm36_0) `SELECT * FROM "wri-rw".gadm36_0 WHERE coastal = True`- [EEZ geometries](https://resourcewatch.carto.com/u/wri-rw/tables/com_011_1_maritime_boundaries_territorial_waters/public?redirected=true): `com_011_1_maritime_boundaries_territorial_waters`  WRI still working out a few discrepancies with the country iso codes used in the two datasets. We want to ensure consistency between the code columns so a country geometry can be matched with its EEZ geometry. We plan to make some edits to the gid_0 and iso_ter1 fields by next week (May31st week), but the underlying geometries should remain the same. Methodology1. Create a geojson from the CARTO table in new row2. Get ID for geojson from Geostore in new row3. Publish to CARTO ResultsPopulated columns within each table: `geostore_prod` and `geostore_staging` Issues found:* Some of the geometries will require sanitization as there is a mix between Polygons, Multipolygons and GeometryCollections.* Geostore size limits for complex Polygons is 16mb, which is very small for a geometry that cover a full country.* API inconsistency and random errors 503/504* Carto 429: too many requests at the same time<jupyter_code>import os
import sys
from dotenv import load_dotenv
import json
import geojson
import logging
import math
from functools import partial 
import asyncio
import nest_asyncio
from concurrent.futures import ProcessPoolExecutor
from decimal import Decimal
from typing import List, Tuple
import requests
import pandas as pd

load_dotenv("../.env")
logger = logging.getLogger('Script')
logger.setLevel(logging.INFO)
nest_asyncio.apply()<jupyter_output><empty_output><jupyter_text>Global Vars<jupyter_code>token = os.environ.get("RW_CARTO_KEY")
account = os.environ.get("RW_CARTO_ACCOUNT")
MAX_WORKERS = 20
SLEEP_TIME = 5<jupyter_output><empty_output><jupyter_text>Functions<jupyter_code>def postGeostore(geojson, cartodb_id, env):
    '''
    Register valid geojson to the geostore service. Return the geostore id.
    '''
    serverUrl = {
    'prod': 'https://api.resourcewatch.org/v1/geostore',
    'staging': 'https://staging-api.globalforestwatch.org/v1/geostore'
    }
    try:
        header = {
                'Content-Type':'application/json'
                }
        body = {
                'geojson': json.loads(geojson)
                }
        
        response = requests.post(serverUrl[env], headers=header, json=body)
        response.raise_for_status()
        
        return {
                'geostore_id': response.json().get('data',{}).get('id',''),
                'cartodb_id': cartodb_id,
                'metadata': None
               }
    
    except requests.exceptions.HTTPError as e:
        logger.error(f'Api fetch error in {env}!')
        logger.error(e)
        logger.error(response.text)
        logger.error(response.url)
        status_code = e.response.status_code
        return {
                'cartodb_id': cartodb_id, 
                'geostore_id': None,
                'metadata': {
                    'errror': e,
                    'url': response.url,
                    'text': response.text,
                }
               }
            
    except Exception as e:
        logger.error('is the api!')
        logger.error(e)
        return {
                'cartodb_id': cartodb_id, 
                'geostore_id': None,
                'metadata': {
                    'errror': e,
                }
               }

def get(payload: dict)-> list:
    '''
    
    '''
    cartoUrl = f'https://{account}.carto.com/api/v2/sql'
    try:
        response = requests.get(cartoUrl, params = payload)
        response.raise_for_status()
        logger.info(f'sucess get from page {str(payload.get("page", ""))}')
        return response.json().get('rows',[])        
    except Exception as e:
        logger.error('carto fetch error for call:')
        logger.error(payload)
        logger.error(e)
        return []

def getCount(q: str)-> int:
    '''
    '''
    
    payload = {
        'api_key': token,
        'q': f'select count(*) from ({q}) st'    
    }
    
    result = get(payload)
    
    return result[0]['count'] if len(result) > 0 else None


def composeCallsTask(payload, table_name):
    '''
    
    '''
    partialData = get(payload)
    if len(partialData) > 0:
        partialGeostoreIdsProd = pd.DataFrame.from_dict(registerInGeostore(partialData, 'prod'), orient='columns')
        partialGeostoreIdsStaging = pd.DataFrame.from_dict(registerInGeostore(partialData, 'staging'), orient='columns')

        return updateCarto(partialGeostoreIdsProd, partialGeostoreIdsStaging, table_name)
    return {}


def cartoWorker(q: str, table_name: str):
    '''
    Carto worker that orchestrate and execute
    '''
    #
    paginationSize = 20
     
    nRows = getCount(q)
    nPages = math.ceil(nRows/paginationSize)

    eventLoop = asyncio.get_event_loop()
    executor = ProcessPoolExecutor(max_workers=MAX_WORKERS)
    tasks = [eventLoop.run_in_executor(executor, partial(composeCallsTask, payload = {
                                                                        'api_key': token,
                                                                        'sort_order': 'asc',
                                                                        'q': q,
                                                                        'rows_per_page': paginationSize,
                                                                        'sort_order': 'asc',
                                                                        'page': i,
                                                                        'order_by': 'cartodb_id'
                                                                        },
                                                             table_name = table_name 
                                                         )
                                       ) for i in range(0, nPages) ]
    data=[]
    data = asyncio.gather(*tasks)
    return eventLoop.run_until_complete(data)
        

def registerInGeostore(data, env = 'prod'):
    '''
    There is a different approach that will be passing the table and the carto account and a id filter to generate the geostore
    '''
    dataResult = []
    for row in data:
        result = postGeostore(row.get('geometry'), 
                              row.get('cartodb_id'), 
                              env)
        
        dataResult.append(result)
    
    return dataResult

def updateCarto(produnctionList, stagingList, table_name):
    '''
    
    '''
    data = pd.merge(produnctionList, stagingList, on='cartodb_id', suffixes=['_prod','_staging'])
    values = [f"({str(row['cartodb_id'])}, '{str(row['geostore_id_prod'] or '')}', '{str(row['geostore_id_staging'] or 'null')}')" for row in data.to_dict('records') 
              if (row['geostore_id_prod'] is not None 
              or row['geostore_id_staging'] is not None)]
    
    updateQuery = f'''update {table_name} set geostore_prod=tmp.geostore_prod, geostore_staging=tmp.geostore_staging 
        from (values {', '.join(values)}) as tmp (cartodb_id, geostore_prod, geostore_staging) 
        where {table_name}.cartodb_id=tmp.cartodb_id
        '''
    logger.info(updateQuery)
    try:
        payload = {
        'api_key': token,
        'q': updateQuery   
        }
        
        
        cartoUpdateResponse = get(payload)
        
        return {
            'data': data,
            'sqlUpdate': updateQuery,
            'updateResponse': cartoUpdateResponse
        }
    
    except Exception as e:
        logger.error('is updating carto!')
        logger.error(e)
        return {}
    
def getGeometriesFromCarto(table_name):
    '''
    
    '''
    dataQuery = f'''
    select cartodb_id, 
        CASE
           WHEN GeometryType(the_geom) like 'GeometryCollection' THEN st_asgeojson(st_makevalid(ST_Simplify(ST_CollectionExtract(the_geom, 3), 0.1, true)))
           Else st_asgeojson(st_makevalid(ST_Simplify(ST_RemoveRepeatedPoints(the_geom, 0.1), 0.5, true)))
        END  as geometry
    from {table_name}
    where geostore_prod is null
    '''
    
    dataCarto = cartoWorker(dataQuery, table_name)
    
    return dataCarto<jupyter_output><empty_output><jupyter_text>Processing<jupyter_code>getGeometriesFromCarto('gadm36_0')
#getGeometriesFromCarto('wat_068_rw0_watersheds_edit')
getGeometriesFromCarto('com_011_1_maritime_boundaries_territorial_waters')<jupyter_output>Api fetch error in prod!
504 Server Error: Gateway Timeout for url: https://api.resourcewatch.org/v1/geostore
{"message": "Network error communicating with endpoint"}
https://api.resourcewatch.org/v1/geostore
Api fetch error in prod!
504 Server Error: Gateway Timeout for url: https://api.resourcewatch.org/v1/geostore
{"message": "Network error communicating with endpoint"}
Api fetch error in prod!
https://api.resourcewatch.org/v1/geostore
504 Server Error: Gateway Timeout for url: https://api.resourcewatch.org/v1/geostore
{"message": "Network error communicating with endpoint"}
https://api.resourcewatch.org/v1/geostore
Api fetch error in prod!
504 Server Error: Gateway Timeout for url: https://api.resourcewatch.org/v1/geostore
{"message": "Network error communicating with endpoint"}
https://api.resourcewatch.org/v1/geostore
Api fetch error in prod!
503 Server Error: first byte timeout for url: https://api.resourcewatch.org/v1/geostore

<?xml version="1.0" encoding="utf-8"?>
<!DOCTYPE h[...]<jupyter_text>Old methodology using cartoframes  Accessing the tables Countries geometry<jupyter_code>countries_df = cf.io.carto.read_carto('SELECT * FROM "wri-rw".gadm36_0 WHERE coastal = True', credentials=creds)
countries_df.head()<jupyter_output><empty_output><jupyter_text>Watersheds<jupyter_code>watersheds_df = cf.io.carto.read_carto('SELECT * FROM "wri-rw".wat_068_rw0_watersheds_edit WHERE level = 3', credentials=creds)
watersheds_df.head()<jupyter_output><empty_output><jupyter_text>EEZ<jupyter_code>eez_table = 'com_011_1_maritime_boundaries_territorial_waters'
eez_df = cf.io.carto.read_carto(eez_table, credentials=creds)

eez_df.head()<jupyter_output><empty_output><jupyter_text>Creating geojson and getting geostore id  Countries geometries<jupyter_code>countries_g_id_dict = {}
countries_geojson_dict = {}
for index, row in countries_df.iterrows():
    g = shpToGeojson(row.the_geom)
    g_id = registerGeostore(g)
    if re.search('^Error ', g_id)!=None: #if there is an error with geostore
        countries_geojson_dict[row.cartodb_id] = g
    else: 
        countries_g_id_dict[row.cartodb_id] = g_id
Api fetch error in prod!
Api fetch error in prod!
Api fetch error in prod!print(len(countries_g_id_dict))
print(len(countries_geojson_dict))
len(countries_g_id_dict)+len(countries_geojson_dict)
countries_df.shape
# save dicts locally
with open('countries_g_id_dict.json', 'w') as fp:
    json.dump(countries_g_id_dict, fp)
with open('countries_geojson_dict.json', 'w') as fp:
    json.dump(countries_geojson_dict, fp)
countries_df.to_csv(f'./countries_df_v20210528.csv')<jupyter_output><empty_output><jupyter_text>Watersheds<jupyter_code>watershed_g_id_dict = {}
watershed_geojson_dict = {}
for index, row in watersheds_df.iterrows():
    g = shpToGeojson(row.the_geom)
    g_id = registerGeostore(g)
    if re.search('^Error ', g_id)!=None: #if there is an error with geostore
        watershed_geojson_dict[row.cartodb_id] = g
    else: 
        watershed_g_id_dict[row.cartodb_id] = g_id
print(len(watershed_g_id_dict))
print(len(watershed_geojson_dict))
len(watershed_g_id_dict)+len(watershed_geojson_dict)
watersheds_df.shape
# save dicts locally
with open('watershed_g_id_dict.json', 'w') as fp:
    json.dump(watershed_g_id_dict, fp)

# transform dictionary to table with two columns and merge
watershed_id_df = pd.DataFrame.from_dict(watershed_g_id_dict, orient = 'index', columns = ["geo_id"])
watershed_id_df.reset_index(inplace=True)
watershed_id_df = watershed_id_df.rename(columns = {'index':'cartodb_id'})
watershed_id_df.merge(watersheds_df, left_on='cartodb_id', right_on='cartodb_id')
watershed_id_df.head()
#save locally
watershed_id_df.to_csv(f'./watersheds_df_id_v20210528.csv')
watershed_id_df = pd.read_csv(f'./watersheds_df_id_v20210528.csv')
watershed_id_df.head()<jupyter_output><empty_output><jupyter_text>EEZ<jupyter_code>eez_g_id_dict = {}
eez_geojson_dict = {}
for index, row in eez_df.iterrows():
    g = shpToGeojson(row.the_geom)
    g_id = registerGeostore(g)
    if re.search('^Error ', g_id)!=None: #if there is an error with geostore
        eez_geojson_dict[row.cartodb_id] = g
    else: 
        eez_g_id_dict[row.cartodb_id] = g_id
eez_df.shape
print(len(eez_g_id_dict))
print(len(eez_geojson_dict))
len(eez_g_id_dict)+len(eez_geojson_dict)
# save dicts locally
with open('eez_g_id_dict.json', 'w') as fp:
    json.dump(eez_g_id_dict, fp)
with open('eez_geojson_dict.json', 'w') as fp:
    json.dump(eez_geojson_dict, fp)
## save locally
eez_df.to_csv(f'./{eez_table}_v20210528.csv')<jupyter_output><empty_output><jupyter_text>check failed registration, those elements in geojson_dict- for countries- for eez  EEZProcess to validate the geometry. Use PostGIS and `ST_MakeValid` ([documentation](https://postgis.net/docs/ST_MakeValid.html)). The geometries are returned as hex and transformed into wkb. The valid geometry can be sent to geostore and added to the dictionary that contains the cartodb_id and the geostore id.<jupyter_code>with open('eez_geojson_dict.json') as json_file:
    eez_geojson_dict = json.load(json_file)
', '.join(eez_geojson_dict.keys())
eez_subset = cf.io.carto.read_carto(f'SELECT the_geom AS before_geom, ST_MakeValid(the_geom) AS after_geom FROM com_011_1_maritime_boundaries_territorial_waters WHERE cartodb_id in ({", ".join(eez_geojson_dict.keys())})', 
                                    credentials=creds)
#eez_subset.to_csv("valid_eez_subset.csv")
eez_subset = gpd.read_file('valid_eez_subset.csv')
eez_subset
eez_df.info()
type(eez_df.the_geom[0])
eez_subset['before_geom']
for g in eez_subset.index:
    try: 
        initial_string2 = eez_subset.loc[[g]].after_geom        
        initial_byte2 = bytes(bytearray.fromhex(initial_string2.values[0]))
        geom_shape = shapely.wkb.loads(initial_byte2)
        if geom_shape.is_valid:
            print("is valid")
            gjson = shpToGeojson(geom_shape)
            print("geojson")           
            g_id = registerGeostore(gjson)
            print(g_id)

            
    except:
        print(len(initial_string2.values))
for g in eez_subset.index:
    try: 
        initial_string2 = eez_subset.loc[[g]].after_geom        
        initial_byte2 = bytes(bytearray.fromhex(initial_string2.values[0]))
        geom_shape = shapely.wkb.loads(initial_byte2)
        if geom_shape.is_valid:
            print("is valid")
            gjson = shpToGeojson(geom_shape)
            print("geojson")           
            g_id = registerGeostore(gjson)
            print(g_id)
            time.sleep(10)
    except:
        print(len(initial_string2.values))
time.sleep(10)
xx = gpd.GeoSeries([geom_shape]).to_json()
body = xx
header = {
    'Content-Type':'application/json'
                }
r = requests.post('http://api.resourcewatch.org/v1/geostore', headers=header, json=body)

r
g_id = registerGeostore(gjson)
for g in eez_geojson_dict:
    rr = eez_df.loc[eez_df['cartodb_id']== int(g)]
    
    #s = rr.the_geom
    print(rr.sovereign1)
    #s2 = s.buffer(0)
    #print(s2.is_valid)
    #g.is_valid
    #break
len(eez_geojson_dict)
## check failed registration, geostore_id  == ''
## s.is_valid, s.make_valid / buffer(0)<jupyter_output><empty_output><jupyter_text>Countries<jupyter_code>with open('countries_geojson_dict.json') as json_file:
    countries_geojson_dict = json.load(json_file)
', '.join(countries_geojson_dict.keys())
countries_subset = cf.io.carto.read_carto(f'SELECT * FROM "wri-rw".gadm36_0 WHERE cartodb_id in ({", ".join(countries_geojson_dict.keys())})', 
                                    credentials=creds)
countries_subset.head()
for g in countries_geojson_dict:
    rr = countries_subset.loc[countries_subset['cartodb_id']==int(g)]   
    s = rr.the_geom
    if s.is_valid:
        print(rr)
    else:
        print(g)
    #g.is_valid
    #break
## check failed registration, geostore_id  == ''
## s.is_valid, s.make_valid / buffer(0)<jupyter_output><empty_output><jupyter_text>write carto table  Watersheds<jupyter_code>watershed_id_df = pd.read_csv(f'./watersheds_df_id_v20210528.csv')
watershed_id_df.head()
watershed_id_df.rename(columns = {"geo_id":"geostore_prod"}, inplace=True)
watershed_id_df.head()<jupyter_output><empty_output><jupyter_text>Using requests code<jupyter_code>sql = 'SELECT * FROM wat_068_rw0_watersheds_edit WHERE level = 3'
def get_query_carto(sql, creds=None, account='wri-rw'):
    urlCarto = f"https://{account}.carto.com/api/v2/sql"
    params = {"q": sql}
    if creds: params["api_key"] = creds
    r = requests.get(urlCarto, params=params)
    if r.status_code==200:
        return r.json().get('rows', None)
    return f'Error {r.status_code}'
%%time
watersheds_df = get_query_carto(sql=sql)
print(watersheds_df[0].keys())
sql_alter = '''
    ALTER TABLE wat_068_rw0_watersheds_edit
    ADD COLUMN geostore_prod VARCHAR,
    ADD COLUMN geostore_staging VARCHAR;
'''
%%time
get_query_carto(sql_alter, creds=creds.api_key)
watershed_update_list = watershed_id_df[["cartodb_id", "geostore_prod"]].to_dict(orient = "records")
print(watershed_update_list[0])
%%time
records = []
for d in watershed_update_list:
    geo_id = d['geostore_prod']
    cartodb_id = d['cartodb_id']
    update_sql = f"""
        UPDATE wat_068_rw0_watersheds_edit 
           SET geostore_prod = '{geo_id}'
         WHERE level = 3 
           AND cartodb_id = {cartodb_id}
        """
    response = get_query_carto(update_sql, creds=creds.api_key)
    records += [{
        **d,
        'response':response
    }]
len([r for r in records if r.get("response", None)])<jupyter_output><empty_output><jupyter_text>Using cartoframes<jupyter_code># cf.io.carto.to_carto(df, <tablename>, if_exists='replace', credentials=creds)
# cf.update_privacy_table(<tablename>, privacy='public', credentials=creds)
# api sql by CARTO, alter table: two new fields (geostore_prod, and geostore_staging), 
# https://carto.com/developers/sql-api/reference/#operation/postSQLStatement
# update row
# bulk update: insert table and then update table using cartodb_id<jupyter_output><empty_output><jupyter_text>Countries<jupyter_code># cf.io.carto.to_carto(df, <tablename>, if_exists='replace', credentials=creds)
# cf.update_privacy_table(<tablename>, privacy='public', credentials=creds)<jupyter_output><empty_output><jupyter_text>EEZ<jupyter_code># cf.io.carto.to_carto(df, <tablename>, if_exists='replace', credentials=creds)
# cf.update_privacy_table(<tablename>, privacy='public', credentials=creds)
#simplify, reduce number of the points or reduce the precision of the points (decimals in the lat long).<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11009221050486526, 'fraction_numerical': 0.010290896123083194, 'mean_word_length': 3.208833619210978, 'pattern_counts': {'":': 6, '<': 76, '<?xml version=': 1, '>': 79, 'https://': 18, 'lorem ipsum': 0, 'www.': 0, 'xml': 1}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '17', 'dir': 'jupyter-structured-clean-dedup', 'id': '655376', 'license': 'MIT', 'n_tokens_mistral': 6786, 'n_tokens_neox': 6484, 'n_words': 1284, 'path': 'ResourceWatch/Script_geostore_IDs_OW-79.ipynb', 'repo_name': 'resource-watch/notebooks', 'size': '19629'}