<filename>.dataset/dvfplus/setup-3-to-csv.py
#!/usr/bin/env python
from pdb import set_trace as bp

import json
import os
import re
import sys
import glob

import pandas as pd
import typer
from typing import Optional
from sqlalchemy import create_engine
import psycopg2

debug = False
verbose = True
use_pandas = False

if use_pandas:
    print("WARNING: under developpement, this will probably fail because of memory limitation.")

if debug:
    limit = 10000
    verbose = True


def main(
        path_cfg: str = typer.Argument(..., help="Data configuration file."),
        path_sec: Optional[str] = typer.Option(None, "-sec", help="Path to PG configuration file.")
):

    with open(path_cfg) as f_cfg:
        cfg = json.load(f_cfg)
        if verbose:
            print(cfg)


    if path_sec is None:
        pat = cfg['dbname'][0:15]
        lf = glob.glob(
            os.environ['HOME'] + '/.cred/ofm/database/'+pat+"*.json"
        )
        if len(lf) > 1:
            print("ERROR: there is more than one credential file matching.")
            print("       Please use -sec to pass path_sec.")
            sys.exit(1)
        else:
            path_sec = lf[0]

    with open(path_sec) as f_sec:
        sec = json.load(f_sec)
        if verbose:
            print(sec)


    if not cfg['dbname'] == sec['POSTGRES_DBNAME']:
        typer.echo("ERROR: Something is wrong, database missmatch")
        typer.echo(cfg['dbname'])
        typer.echo(sec['POSTGRES_DBNAME'])
        raise typer.Exit(1)

    output_dir = f"{cfg['delivery']}/csv"

    if not os.path.isdir(output_dir):
        os.makedirs(output_dir)

    # A long string that contains the necessary Postgres login information
    postgres_secret = ('postgresql://{username}:{password}@{ipaddress}:{port}/{dbname}'.format(
        username=sec['POSTGRES_USERNAME'],
        password=sec['POSTGRES_PASSWORD'],
        ipaddress=sec['POSTGRES_ADDRESS'],
        port=sec['POSTGRES_PORT'],
        dbname=sec['POSTGRES_DBNAME']
    ))
    # Create the engine
    engine = create_engine(postgres_secret)
    df = pd.read_sql_query('''
    SELECT *
    FROM pg_catalog.pg_tables
    WHERE schemaname != 'pg_catalog' AND 
    schemaname != 'information_schema';''', engine)
    mask_dvf = (df.schemaname == 'dvf') | (df.schemaname == 'dvf_annexe')
    tablenames = (df[mask_dvf].schemaname + '.' + df[mask_dvf].tablename).tolist()
    del df
    
    conn = psycopg2.connect(host=sec['POSTGRES_ADDRESS'], port=sec['POSTGRES_PORT'], dbname=sec['POSTGRES_DBNAME'], user=sec['POSTGRES_USERNAME'], password=sec['<PASSWORD>'])
    cur = conn.cursor()

    def read_write_table(tablename, conn, cur):

        # Because BigQuery GIS do not support EWKB (2021).
        df = pd.read_sql(f"select * from {tablename} LIMIT 1", conn)
        lcol = df.columns.tolist()
        for i, col in enumerate(lcol):
            if re.match(r"^geom.*$", col):
                 
                #lcol[i] = f"ST_AsText(ST_Transform({col}, 4326)) as {col}"
                # If there is an importation problem try, GeoJSON
                #if col == 'geompar':
                # lcol[i] = f"REPLACE(ST_AsGeoJSON(ST_Transform({col}, 4326)), '] ], [ [', '] ] ], [ [ [') as {col}"
                # lcol[i] = f"ST_AsGeoJSON(ST_MakeValid(ST_Transform({col}, 4326))) as {col}"
                #else: 
                lcol[i] = f"ST_AsGeoJSON(ST_Transform({col}, 4326)) as {col}"
                
        lcol = ", ".join(lcol)
        
        if debug:
            subquery = f"SELECT {lcol} FROM {tablename} LIMIT {limit}"
        else:
            subquery = f"SELECT {lcol} FROM {tablename}"
            
        # Use the COPY function on the SQL we created above.
        query = "COPY ({0}) TO STDOUT WITH CSV HEADER".format(subquery)

        # Set up a variable to store our file path and name.
        opath = output_dir+os.sep+tablename+".csv"

        try:
            with open(opath, 'w') as f_output:
                if verbose:
                    print("dump with query:")
                    print(query)
                cur.copy_expert(query, f_output)
                if verbose:
                    print("finish")
        except psycopg2.Error as e:
            t_message = "\nERROR: \n" + str(e) + \
                "\nsubquery: " + \
                subquery + \
                "\nopath: " + opath + "\n\n"
            print(t_message)
            raise

    def read_write_table_pd(tablename):
        if debug:
            df = pd.read_sql_query(
                f'''SELECT * FROM {tablename} LIMIT {limit}''',
                engine
            )
        else:
            df = pd.read_sql_query(
                f'''SELECT * FROM {tablename}''',
                engine
            )
        if verbose:
            print(df)
        opath = output_dir+os.sep+tablename+".csv"
        if verbose:
            print(opath)
        if os.path.exists(opath):
            typer.echo(f"Skipping file {opath}")
        else:
            if verbose:
                print("Writting table")
            df.to_csv(opath, index=False)
        del df

    for it, tablename in enumerate(tablenames):
        if verbose:
            print(it+1, tablename)
            print("Loading table")
        if use_pandas:
            read_write_table_pd(tablename)
        else:
            read_write_table(tablename, conn, cur)
            
    # Clean up: Close the database cursor and connection
    if not use_pandas:
        cur.close()
        conn.close()

if __name__ == "__main__":
    typer.run(main)

<filename>.dataset/dvfplus/setup-3-to-csv.py
#!/usr/bin/env python
from pdb import set_trace as bp

import json
import os
import re
import sys
import glob

import pandas as pd
import typer
from typing import Optional
from sqlalchemy import create_engine
import psycopg2

debug = False
verbose = True
use_pandas = False

if use_pandas:
    print("WARNING: under developpement, this will probably fail because of memory limitation.")

if debug:
    limit = 10000
    verbose = True


def main(
        path_cfg: str = typer.Argument(..., help="Data configuration file."),
        path_sec: Optional[str] = typer.Option(None, "-sec", help="Path to PG configuration file.")
):

    with open(path_cfg) as f_cfg:
        cfg = json.load(f_cfg)
        if verbose:
            print(cfg)


    if path_sec is None:
        pat = cfg['dbname'][0:15]
        lf = glob.glob(
            os.environ['HOME'] + '/.cred/ofm/database/'+pat+"*.json"
        )
        if len(lf) > 1:
            print("ERROR: there is more than one credential file matching.")
            print("       Please use -sec to pass path_sec.")
            sys.exit(1)
        else:
            path_sec = lf[0]

    with open(path_sec) as f_sec:
        sec = json.load(f_sec)
        if verbose:
            print(sec)


    if not cfg['dbname'] == sec['POSTGRES_DBNAME']:
        typer.echo("ERROR: Something is wrong, database missmatch")
        typer.echo(cfg['dbname'])
        typer.echo(sec['POSTGRES_DBNAME'])
        raise typer.Exit(1)

    output_dir = f"{cfg['delivery']}/csv"

    if not os.path.isdir(output_dir):
        os.makedirs(output_dir)

    # A long string that contains the necessary Postgres login information
    postgres_secret = ('postgresql://{username}:{password}@{ipaddress}:{port}/{dbname}'.format(
        username=sec['POSTGRES_USERNAME'],
        password=sec['POSTGRES_PASSWORD'],
        ipaddress=sec['POSTGRES_ADDRESS'],
        port=sec['POSTGRES_PORT'],
        dbname=sec['POSTGRES_DBNAME']
    ))
    # Create the engine
    engine = create_engine(postgres_secret)
    df = pd.read_sql_query('''
    SELECT *
    FROM pg_catalog.pg_tables
    WHERE schemaname != 'pg_catalog' AND 
    schemaname != 'information_schema';''', engine)
    mask_dvf = (df.schemaname == 'dvf') | (df.schemaname == 'dvf_annexe')
    tablenames = (df[mask_dvf].schemaname + '.' + df[mask_dvf].tablename).tolist()
    del df
    
    conn = psycopg2.connect(host=sec['POSTGRES_ADDRESS'], port=sec['POSTGRES_PORT'], dbname=sec['POSTGRES_DBNAME'], user=sec['POSTGRES_USERNAME'], password=sec['<PASSWORD>'])
    cur = conn.cursor()

    def read_write_table(tablename, conn, cur):

        # Because BigQuery GIS do not support EWKB (2021).
        df = pd.read_sql(f"select * from {tablename} LIMIT 1", conn)
        lcol = df.columns.tolist()
        for i, col in enumerate(lcol):
            if re.match(r"^geom.*$", col):
                 
                #lcol[i] = f"ST_AsText(ST_Transform({col}, 4326)) as {col}"
                # If there is an importation problem try, GeoJSON
                #if col == 'geompar':
                # lcol[i] = f"REPLACE(ST_AsGeoJSON(ST_Transform({col}, 4326)), '] ], [ [', '] ] ], [ [ [') as {col}"
                # lcol[i] = f"ST_AsGeoJSON(ST_MakeValid(ST_Transform({col}, 4326))) as {col}"
                #else: 
                lcol[i] = f"ST_AsGeoJSON(ST_Transform({col}, 4326)) as {col}"
                
        lcol = ", ".join(lcol)
        
        if debug:
            subquery = f"SELECT {lcol} FROM {tablename} LIMIT {limit}"
        else:
            subquery = f"SELECT {lcol} FROM {tablename}"
            
        # Use the COPY function on the SQL we created above.
        query = "COPY ({0}) TO STDOUT WITH CSV HEADER".format(subquery)

        # Set up a variable to store our file path and name.
        opath = output_dir+os.sep+tablename+".csv"

        try:
            with open(opath, 'w') as f_output:
                if verbose:
                    print("dump with query:")
                    print(query)
                cur.copy_expert(query, f_output)
                if verbose:
                    print("finish")
        except psycopg2.Error as e:
            t_message = "\nERROR: \n" + str(e) + \
                "\nsubquery: " + \
                subquery + \
                "\nopath: " + opath + "\n\n"
            print(t_message)
            raise

    def read_write_table_pd(tablename):
        if debug:
            df = pd.read_sql_query(
                f'''SELECT * FROM {tablename} LIMIT {limit}''',
                engine
            )
        else:
            df = pd.read_sql_query(
                f'''SELECT * FROM {tablename}''',
                engine
            )
        if verbose:
            print(df)
        opath = output_dir+os.sep+tablename+".csv"
        if verbose:
            print(opath)
        if os.path.exists(opath):
            typer.echo(f"Skipping file {opath}")
        else:
            if verbose:
                print("Writting table")
            df.to_csv(opath, index=False)
        del df

    for it, tablename in enumerate(tablenames):
        if verbose:
            print(it+1, tablename)
            print("Loading table")
        if use_pandas:
            read_write_table_pd(tablename)
        else:
            read_write_table(tablename, conn, cur)
            
    # Clean up: Close the database cursor and connection
    if not use_pandas:
        cur.close()
        conn.close()

if __name__ == "__main__":
    typer.run(main)

#!/usr/bin/env python # A long string that contains the necessary Postgres login information # Create the engine SELECT *     FROM pg_catalog.pg_tables     WHERE schemaname != 'pg_catalog' AND      schemaname != 'information_schema'; # Because BigQuery GIS do not support EWKB (2021). #lcol[i] = f"ST_AsText(ST_Transform({col}, 4326)) as {col}" # If there is an importation problem try, GeoJSON #if col == 'geompar': # lcol[i] = f"REPLACE(ST_AsGeoJSON(ST_Transform({col}, 4326)), '] ], [ [', '] ] ], [ [ [') as {col}" # lcol[i] = f"ST_AsGeoJSON(ST_MakeValid(ST_Transform({col}, 4326))) as {col}" #else: # Use the COPY function on the SQL we created above. # Set up a variable to store our file path and name. SELECT * FROM {tablename} LIMIT {limit} SELECT * FROM {tablename} # Clean up: Close the database cursor and connection