"""
Extract the contents of the FAULTS keyword into
a DataFrame
"""
from __future__ import annotations
import argparse
import logging
from itertools import product
# Needed for mypy
import opm.io
import pandas as pd
from .common import parse_opmio_deckrecord, write_dframe_stdout_file
from .res2csvlogger import getLogger_res2csv
from .resdatafiles import ResdataFiles
logger = logging.getLogger(__name__)
RECORD_COLUMNS = ["NAME", "IX1", "IX2", "IY1", "IY2", "IZ1", "IZ2", "FACE"]
COLUMNS = ["NAME", "I", "J", "K", "FACE"]
ALLOWED_FACES = ["X", "Y", "Z", "I", "J", "K", "X-", "Y-", "Z-", "I-", "J-", "K-"]
[docs]
def df(deck: ResdataFiles | opm.opmcommon_python.Deck) -> pd.DataFrame:
"""Produce a dataframe of fault data from a :term:`deck`
All data for the keyword FAULTS will be returned.
Args:
deck: A :term:`deck`
"""
if isinstance(deck, ResdataFiles):
deck = deck.get_deck()
# In[91]: list(deck['FAULTS'][0])
# Out[91]: [[u'F1'], [36], [36], [41], [42], [1], [14], [u'I']]
data: list[list[str | int]] = []
# It is allowed in Eclipse to use the keyword FAULTS
# as many times as needed. Thus we need to loop in some way:
for keyword in deck:
if keyword.name == "FAULTS":
for rec in keyword:
# Each record now has a range potentially in three
# dimensions for the fault, unroll this:
frec_dict = parse_opmio_deckrecord(rec, "FAULTS")
faultname = frec_dict["NAME"]
faultface = frec_dict["FACE"]
indices = product(
range(frec_dict["IX1"], frec_dict["IX2"] + 1),
range(frec_dict["IY1"], frec_dict["IY2"] + 1),
range(frec_dict["IZ1"], frec_dict["IZ2"] + 1),
)
data.extend([faultname, i, j, k, faultface] for i, j, k in indices)
dframe = pd.DataFrame(columns=COLUMNS, data=data)
logger.info("Extracted %i faults", len(dframe["NAME"].unique()))
return dframe
[docs]
def fill_parser(parser: argparse.ArgumentParser) -> argparse.ArgumentParser:
"""Set up sys.argv parsers.
Arguments:
parser: argparse.ArgumentParser or argparse.subparser
"""
parser.add_argument(
"DATAFILE", help="Name of the .DATA input file for the reservoir simulator"
)
parser.add_argument(
"-o",
"--output",
type=str,
help="Name of output csv file.",
default="faults.csv",
)
parser.add_argument("-v", "--verbose", action="store_true", help="Be verbose")
return parser
[docs]
def faults_main(args: argparse.Namespace) -> None:
"""Read from disk and write CSV back to disk"""
logger = getLogger_res2csv(__name__, vars(args))
resdatafiles = ResdataFiles(args.DATAFILE)
deck = resdatafiles.get_deck()
faults_df = df(deck)
write_dframe_stdout_file(
faults_df,
args.output,
index=False,
caller_logger=logger,
logstr=f"Wrote to {args.output}",
)