zele-utils/functions/helpers.py

import geopandas, os, datetime
from sqlalchemy import create_engine
from geoalchemy2 import Geometry, WKTElement

def buffer_creator(file,divider,start_line, chunk_size):
  buffer = []
  line_number = start_line
  current_line = 0
  divider_count = 0
  with open(file,'r',encoding='utf-8') as f:
    for line in f:
      current_line += 1
      if (current_line <= line_number): continue
      if (line.strip()== divider):
        divider_count = divider_count + 1
        if divider_count == chunk_size: break
        continue
      buffer.append(line.strip())
  return current_line,(' ').join(buffer)
  
def push_to_db_coords(name,data,mode):
    GDF = geopandas.GeoDataFrame(data, crs='EPSG:4326')
    GDF['geom'] = GDF['coordinates'].apply(lambda x: WKTElement(x.wkt, srid=os.getenv("SRID")))
    engine = create_engine(f'postgresql://{os.getenv("USER")}:{os.getenv("PASS")}@{os.getenv("HOST_NAME")}/{os.getenv("DATA_BASE")}', echo=False)
    GDF.to_sql(
        name=name,
        con=engine,
        if_exists=mode,
        chunksize=os.getenv("CHUNK_SIZE"),
        dtype={'geom': Geometry('Point', srid=os.getenv("SRID"))},
        index=False
    )

def write_to_csv(name,data, file):
    directory = file.parent
    id = datetime.datetime.now().strftime("%Y%m%d")
    csv = directory / (file.stem + f"-{name}-{id}.csv")
    if csv.exists():
         data.to_csv(csv, mode='a',index=False)
    else:
        data.to_csv(csv,index=False)