internship/data/processing/subdomain.py

import argparse
import configparser
import logging
import pathlib
import sys

import numpy as np

parser = argparse.ArgumentParser(description="Pre-process bathymetry")
parser.add_argument("-v", "--verbose", action="count", default=0)
parser.add_argument("-c", "--config", default="config.ini")
args = parser.parse_args()

logging.basicConfig(level=max((10, 20 - 10 * args.verbose)))
log = logging.getLogger("bathy")

log.info("Starting bathymetry pre-processing")
config = configparser.ConfigParser()
config.read(args.config)

artha = np.array((config.getfloat("artha", "lon"), config.getfloat("artha", "lat")))
buoy = np.array((config.getfloat("buoy", "lon"), config.getfloat("buoy", "lat")))
log.debug(f"artha: {artha}")
log.debug(f"buoy: {buoy}")

domain = np.stack((artha, buoy))
domain.sort(axis=0)
log.debug(f"domain: {domain}")

margin = config.getfloat("out", "margin", fallback=0.002)
domain[0] -= margin
domain[1] += margin
log.debug(f"domain: {domain}")

inp_root = pathlib.Path(config.get("inp", "root"))
out_root = pathlib.Path(config.get("out", "root"))
out_root.mkdir(exist_ok=True)
bathy_inp = inp_root.joinpath(config.get("inp", "base"))
bathy_out = out_root.joinpath(config.get("out", "sub"))
if bathy_out.exists():
    log.error(f"'{bathy_out}' already exists!")
    sys.exit(1)
log.info(f"Reading bathymetry from '{bathy_inp}'")
raw_bathy = np.loadtxt(bathy_inp)
log.debug(f"Initial size: {raw_bathy.shape}")

bathy = raw_bathy[
    ((raw_bathy[:, :2] > domain[0]) & (raw_bathy[:, :2] < domain[1])).all(axis=1)
]
del raw_bathy
log.debug(f"Final size: {bathy.shape}")

log.info(f"Saving subdomain to 'bathy'")
np.save(bathy_out, bathy)
bathy_out.chmod(0o444)