10691143420

Committed 03 Sep 2024 09:48PM UTC coverage: 76.813% (+0.8%) from 76.01%

Build # 10691143420

Build Type

Pull #35

github

Committed by

web-flow

Commit Message

Merge 1ebfe99fb into 81a2299af

Pull Request Pull Request #35: Bump actions/download-artifact from 2 to 4.1.7 in /.github/workflows

Run Details

1928 of 2510 relevant lines covered (76.81%)

10.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.67

/src/welltestpy/data/data_io.py

"""welltestpy subpackage providing input-output routines."""
import csv
import numbers
import os
import shutil
import tempfile
import zipfile
from io import BytesIO as BytIO
from io import TextIOWrapper as TxtIO

import numpy as np
from packaging.version import parse as version_parse

from . import campaignlib, testslib, varlib

try:
    from .._version import __version__
except ImportError:  # pragma: nocover
    # package is not installed
    __version__ = "0.0.0.dev0"


# TOOLS ###


class LoadError(Exception):
    """Loading error for all reading routines."""

    pass


def _formstr(string):
    # remove spaces, tabs, linebreaks and other separators
    return "".join(str(string).split())


def _formname(string):
    # remove slashes
    string = "".join(str(string).split(os.path.sep))
    # remove spaces, tabs, linebreaks and other separators
    return _formstr(string)


def _nextr(data):
    return tuple(filter(None, next(data)))


def _check_version(version):
    """At least check major version."""
    if version.major > version_parse(__version__).major:
        raise ValueError(f"Unknown version '{version.public}'")


# SAVE ###


def save_var(var, path="", name=None):
    """Save a variable to file.

    This writes the variable to a csv file.

    Parameters
    ----------
    path : :class:`str`, optional
        Path where the variable should be saved. Default: ``""``
    name : :class:`str`, optional
        Name of the file. If ``None``, the name will be generated by
        ``"Var_"+name``. Default: ``None``

    Notes
    -----
    The file will get the suffix ``".var"``.
    """
    path = os.path.normpath(path)
    # create the path if not existing
    if not os.path.exists(path):
        os.makedirs(path)
    # create a standard name if None is given
    if name is None:
        name = "Var_" + var.name
    # ensure the name ends with '.var'
    if name[-4:] != ".var":
        name += ".var"
    name = _formname(name)
    file_path = os.path.join(path, name)
    # write the csv-file
    with open(file_path, "w") as csvf:
        writer = csv.writer(
            csvf, quoting=csv.QUOTE_NONNUMERIC, lineterminator="\n"
        )
        writer.writerow(["wtp-version", __version__])
        writer.writerow(["Variable"])
        writer.writerow(["name", var.name])
        writer.writerow(["symbol", var.symbol])
        writer.writerow(["units", var.units])
        writer.writerow(["description", var.description])
        if issubclass(np.asanyarray(var.value).dtype.type, numbers.Integral):
            writer.writerow(["integer"])
        else:
            writer.writerow(["float"])
        if var.scalar:
            writer.writerow(["scalar"])
            writer.writerow(["value", var.value])
        else:
            writer.writerow(["shape"] + list(np.shape(var.value)))
            tmpvalue = np.reshape(var.value, -1)
            writer.writerow(["values", len(tmpvalue)])
            for val in tmpvalue:
                writer.writerow([val])
    return file_path


def save_obs(obs, path="", name=None):
    """Save an observation to file.

    This writes the observation to a csv file.

    Parameters
    ----------
    path : :class:`str`, optional
        Path where the variable should be saved. Default: ``""``
    name : :class:`str`, optional
        Name of the file. If ``None``, the name will be generated by
        ``"Obs_"+name``. Default: ``None``

    Notes
    -----
    The file will get the suffix ``".obs"``.
    """
    path = os.path.normpath(path)
    # create the path if not existing
    if not os.path.exists(path):
        os.makedirs(path)
    # create a standard name if None is given
    if name is None:
        name = "Obs_" + obs.name
    # ensure the name ends with '.obs'
    if name[-4:] != ".obs":
        name += ".obs"
    name = _formname(name)
    # create temporal directory for the included files
    patht = tempfile.mkdtemp(dir=path)
    # write the csv-file
    with open(os.path.join(patht, "info.csv"), "w") as csvf:
        writer = csv.writer(
            csvf, quoting=csv.QUOTE_NONNUMERIC, lineterminator="\n"
        )
        writer.writerow(["wtp-version", __version__])
        writer.writerow(["Observation"])
        writer.writerow(["name", obs.name])
        writer.writerow(["state", obs.state])
        writer.writerow(["description", obs.description])
        if obs.state == "steady":
            obsname = name[:-4] + "_ObsVar.var"
            writer.writerow(["observation", obsname])
            obs._observation.save(patht, obsname)
        else:
            timname = name[:-4] + "_TimVar.var"
            obsname = name[:-4] + "_ObsVar.var"
            writer.writerow(["time", timname])
            writer.writerow(["observation", obsname])
            obs._time.save(patht, timname)
            obs._observation.save(patht, obsname)
    # compress everything to one zip-file
    file_path = os.path.join(path, name)
    with zipfile.ZipFile(file_path, "w") as zfile:
        zfile.write(os.path.join(patht, "info.csv"), "info.csv")
        if obs.state == "transient":
            zfile.write(os.path.join(patht, timname), timname)
        zfile.write(os.path.join(patht, obsname), obsname)
    shutil.rmtree(patht, ignore_errors=True)
    return file_path


def save_well(well, path="", name=None):
    """Save a well to file.

    This writes the variable to a csv file.

    Parameters
    ----------
    path : :class:`str`, optional
        Path where the variable should be saved. Default: ``""``
    name : :class:`str`, optional
        Name of the file. If ``None``, the name will be generated by
        ``"Well_"+name``. Default: ``None``

    Notes
    -----
    The file will get the suffix ``".wel"``.
    """
    path = os.path.normpath(path)
    # create the path if not existing
    if not os.path.exists(path):
        os.makedirs(path)
    # create a standard name if None is given
    if name is None:
        name = "Well_" + well.name
    # ensure the name ends with '.wel'
    if name[-4:] != ".wel":
        name += ".wel"
    name = _formname(name)
    # create temporal directory for the included files
    patht = tempfile.mkdtemp(dir=path)
    # write the csv-file
    with open(os.path.join(patht, "info.csv"), "w") as csvf:
        writer = csv.writer(
            csvf, quoting=csv.QUOTE_NONNUMERIC, lineterminator="\n"
        )
        writer.writerow(["wtp-version", __version__])
        writer.writerow(["Well"])
        writer.writerow(["name", well.name])
        # define names for the variable-files
        radiuname = name[:-4] + "_RadVar.var"
        coordname = name[:-4] + "_CooVar.var"
        welldname = name[:-4] + "_WedVar.var"
        aquifname = name[:-4] + "_AqdVar.var"
        screename = name[:-4] + "_ScrVar.var"
        # save variable-files
        writer.writerow(["radius", radiuname])
        well.wellradius.save(patht, radiuname)
        writer.writerow(["coordinates", coordname])
        well.coordinates.save(patht, coordname)
        writer.writerow(["welldepth", welldname])
        well.welldepth.save(patht, welldname)
        writer.writerow(["aquiferdepth", aquifname])
        well.aquiferdepth.save(patht, aquifname)
        writer.writerow(["screensize", screename])
        well.screensize.save(patht, screename)
    # compress everything to one zip-file
    file_path = os.path.join(path, name)
    with zipfile.ZipFile(file_path, "w") as zfile:
        zfile.write(os.path.join(patht, "info.csv"), "info.csv")
        zfile.write(os.path.join(patht, radiuname), radiuname)
        zfile.write(os.path.join(patht, coordname), coordname)
        zfile.write(os.path.join(patht, welldname), welldname)
        zfile.write(os.path.join(patht, aquifname), aquifname)
        zfile.write(os.path.join(patht, screename), screename)
    # delete the temporary directory
    shutil.rmtree(patht, ignore_errors=True)
    return file_path


def save_campaign(campaign, path="", name=None):
    """Save the campaign to file.

    This writes the campaign to a csv file.

    Parameters
    ----------
    path : :class:`str`, optional
        Path where the variable should be saved. Default: ``""``
    name : :class:`str`, optional
        Name of the file. If ``None``, the name will be generated by
        ``"Cmp_"+name``. Default: ``None``

    Notes
    -----
    The file will get the suffix ``".cmp"``.
    """
    path = os.path.normpath(path)
    # create the path if not existing
    if not os.path.exists(path):
        os.makedirs(path)
    # create a standard name if None is given
    if name is None:
        name = "Cmp_" + campaign.name
    # ensure the name ends with '.cmp'
    if name[-4:] != ".cmp":
        name += ".cmp"
    name = _formname(name)
    # create temporal directory for the included files
    patht = tempfile.mkdtemp(dir=path)
    # write the csv-file
    with open(os.path.join(patht, "info.csv"), "w") as csvf:
        writer = csv.writer(
            csvf, quoting=csv.QUOTE_NONNUMERIC, lineterminator="\n"
        )
        writer.writerow(["wtp-version", __version__])
        writer.writerow(["Campaign"])
        writer.writerow(["name", campaign.name])
        writer.writerow(["description", campaign.description])
        writer.writerow(["timeframe", campaign.timeframe])
        # define names for the variable-files
        if campaign.fieldsite is not None:
            fieldsname = name[:-4] + "_Fieldsite.fds"
            # save variable-files
            writer.writerow(["fieldsite", fieldsname])
            campaign.fieldsite.save(patht, fieldsname)
        else:
            writer.writerow(["fieldsite", "None"])

        wkeys = tuple(campaign.wells.keys())
        writer.writerow(["Wells", len(wkeys)])
        wellsname = {}
        for k in wkeys:
            wellsname[k] = name[:-4] + "_" + k + "_Well.wel"
            writer.writerow([k, wellsname[k]])
            campaign.wells[k].save(patht, wellsname[k])

        tkeys = tuple(campaign.tests.keys())
        writer.writerow(["Tests", len(tkeys)])
        testsname = {}
        for k in tkeys:
            testsname[k] = name[:-4] + "_" + k + "_Test.tst"
            writer.writerow([k, testsname[k]])
            campaign.tests[k].save(patht, testsname[k])

    # compress everything to one zip-file
    file_path = os.path.join(path, name)
    with zipfile.ZipFile(file_path, "w") as zfile:
        zfile.write(os.path.join(patht, "info.csv"), "info.csv")
        if campaign.fieldsite is not None:
            zfile.write(os.path.join(patht, fieldsname), fieldsname)
        for k in wkeys:
            zfile.write(os.path.join(patht, wellsname[k]), wellsname[k])
        for k in tkeys:
            zfile.write(os.path.join(patht, testsname[k]), testsname[k])
    # delete the temporary directory
    shutil.rmtree(patht, ignore_errors=True)
    return file_path


def save_fieldsite(fieldsite, path="", name=None):
    """Save a field site to file.

    This writes the field site to a csv file.

    Parameters
    ----------
    path : :class:`str`, optional
        Path where the variable should be saved. Default: ``""``
    name : :class:`str`, optional
        Name of the file. If ``None``, the name will be generated by
        ``"Field_"+name``. Default: ``None``

    Notes
    -----
    The file will get the suffix ``".fds"``.
    """
    path = os.path.normpath(path)
    # create the path if not existing
    if not os.path.exists(path):
        os.makedirs(path)
    # create a standard name if None is given
    if name is None:
        name = "Field_" + fieldsite.name
    # ensure the name ends with '.fds'
    if name[-4:] != ".fds":
        name += ".fds"
    name = _formname(name)
    # create temporal directory for the included files
    patht = tempfile.mkdtemp(dir=path)
    # write the csv-file
    with open(os.path.join(patht, "info.csv"), "w") as csvf:
        writer = csv.writer(
            csvf, quoting=csv.QUOTE_NONNUMERIC, lineterminator="\n"
        )
        writer.writerow(["wtp-version", __version__])
        writer.writerow(["Fieldsite"])
        writer.writerow(["name", fieldsite.name])
        writer.writerow(["description", fieldsite.description])
        # define names for the variable-files
        if fieldsite.coordinates is not None:
            coordname = name[:-4] + "_CooVar.var"
            # save variable-files
            writer.writerow(["coordinates", coordname])
            fieldsite.coordinates.save(patht, coordname)
        else:
            writer.writerow(["coordinates", "None"])
    # compress everything to one zip-file
    file_path = os.path.join(path, name)
    with zipfile.ZipFile(file_path, "w") as zfile:
        zfile.write(os.path.join(patht, "info.csv"), "info.csv")
        if fieldsite.coordinates is not None:
            zfile.write(os.path.join(patht, coordname), coordname)
    # delete the temporary directory
    shutil.rmtree(patht, ignore_errors=True)
    return file_path


def save_pumping_test(pump_test, path="", name=None):
    """Save a pumping test to file.

    This writes the variable to a csv file.

    Parameters
    ----------
    path : :class:`str`, optional
        Path where the variable should be saved. Default: ``""``
    name : :class:`str`, optional
        Name of the file. If ``None``, the name will be generated by
        ``"Test_"+name``. Default: ``None``

    Notes
    -----
    The file will get the suffix ``".tst"``.
    """
    path = os.path.normpath(path)
    # create the path if not existing
    if not os.path.exists(path):
        os.makedirs(path)
    # create a standard name if None is given
    if name is None:
        name = "Test_" + pump_test.name
    # ensure the name ends with '.tst'
    if name[-4:] != ".tst":
        name += ".tst"
    name = _formname(name)
    # create temporal directory for the included files
    patht = tempfile.mkdtemp(dir=path)
    # write the csv-file
    with open(os.path.join(patht, "info.csv"), "w") as csvf:
        writer = csv.writer(
            csvf, quoting=csv.QUOTE_NONNUMERIC, lineterminator="\n"
        )
        writer.writerow(["wtp-version", __version__])
        writer.writerow(["Testtype", "PumpingTest"])
        writer.writerow(["name", pump_test.name])
        writer.writerow(["description", pump_test.description])
        writer.writerow(["timeframe", pump_test.timeframe])
        writer.writerow(["pumpingwell", pump_test.pumpingwell])
        # define names for the variable-files (file extension added autom.)
        pumprname = name[:-4] + "_PprVar"
        aquidname = name[:-4] + "_AqdVar"
        aquirname = name[:-4] + "_AqrVar"
        # save variable-files
        pumpr_path = pump_test.pumpingrate.save(patht, pumprname)
        pumpr_base = os.path.basename(pumpr_path)
        writer.writerow(["pumpingrate", pumpr_base])
        aquid_path = pump_test.aquiferdepth.save(patht, aquidname)
        aquid_base = os.path.basename(aquid_path)
        writer.writerow(["aquiferdepth", aquid_base])
        aquir_path = pump_test.aquiferradius.save(patht, aquirname)
        aquir_base = os.path.basename(aquir_path)
        writer.writerow(["aquiferradius", aquir_base])
        okeys = tuple(pump_test.observations.keys())
        writer.writerow(["Observations", len(okeys)])
        obsname = {}
        for k in okeys:
            obsname[k] = name[:-4] + "_" + k + "_Obs.obs"
            writer.writerow([k, obsname[k]])
            pump_test.observations[k].save(patht, obsname[k])
    # compress everything to one zip-file
    file_path = os.path.join(path, name)
    with zipfile.ZipFile(file_path, "w") as zfile:
        zfile.write(os.path.join(patht, "info.csv"), "info.csv")
        zfile.write(pumpr_path, pumpr_base)
        zfile.write(aquir_path, aquir_base)
        zfile.write(aquid_path, aquid_base)
        for k in okeys:
            zfile.write(os.path.join(patht, obsname[k]), obsname[k])
    # delete the temporary directory
    shutil.rmtree(patht, ignore_errors=True)
    return file_path


# LOAD ###


def _load_var_data(data):
    # default version string
    version_string = "1.0.0"
    first_line = _nextr(data)
    if first_line[0] == "wtp-version":
        version_string = first_line[1]
        header = _nextr(data)
    else:
        header = first_line
    version = version_parse(version_string)
    _check_version(version)
    if header[0] != "Variable":
        raise ValueError(
            f"load_var: expected 'Variable' but got '{header[0]}'"
        )
    name = next(data)[1]
    symbol = next(data)[1]
    units = next(data)[1]
    description = next(data)[1]
    integer = next(data)[0] == "integer"
    shapenfo = _nextr(data)
    if shapenfo[0] == "scalar":
        if integer:
            value = int(next(data)[1])
        else:
            value = float(next(data)[1])
    else:
        shape = tuple(np.array(shapenfo[1:], dtype=int))
        vcnt = int(next(data)[1])
        vlist = []
        for __ in range(vcnt):
            vlist.append(next(data)[0])
        if integer:
            value = np.array(vlist, dtype=int).reshape(shape)
        else:
            value = np.array(vlist, dtype=float).reshape(shape)

    return varlib.Variable(name, value, symbol, units, description)


def load_var(varfile):
    """Load a variable from file.

    This reads a variable from a csv file.

    Parameters
    ----------
    varfile : :class:`str`
        Path to the file
    """
    cleanup = False
    try:
        # read file
        data_file = open(varfile, "r")
    except TypeError:  # if it is an instance of TextIOWrapper
        try:
            # read stream
            data = csv.reader(varfile)
        except Exception as exc:
            raise LoadError(
                f"load_var: couldn't read file '{varfile}'"
            ) from exc
    else:
        data = csv.reader(data_file)
        cleanup = True

    try:
        var = _load_var_data(data)
    except Exception as exc:
        raise LoadError(
            f"load_var: couldn't load variable '{varfile}'"
        ) from exc

    if cleanup:
        data_file.close()

    return var


def load_obs(obsfile):
    """Load an observation from file.

    This reads a observation from a csv file.

    Parameters
    ----------
    obsfile : :class:`str`
        Path to the file
    """
    # default version string
    version_string = "1.0.0"
    try:
        with zipfile.ZipFile(obsfile, "r") as zfile:
            info = TxtIO(zfile.open("info.csv"))
            data = csv.reader(info)
            first_line = _nextr(data)
            if first_line[0] == "wtp-version":
                version_string = first_line[1]
                header = _nextr(data)
            else:
                header = first_line
            version = version_parse(version_string)
            _check_version(version)
            if header[0] != "Observation":
                raise ValueError(
                    f"load_obs: expected 'Observation' but got '{header[0]}'"
                )
            name = next(data)[1]
            steady = next(data)[1] == "steady"
            description = next(data)[1]
            if not steady:
                timef = next(data)[1]
            obsf = next(data)[1]
            # read time if not steady
            time = None
            if not steady:
                time = load_var(TxtIO(zfile.open(timef)))
            # read observation
            obs = load_var(TxtIO(zfile.open(obsf)))
        # generate observation object
        observation = varlib.Observation(name, obs, time, description)
    except Exception as exc:
        raise LoadError(
            f"load_obs: couldn't load observation '{obsfile}'"
        ) from exc
    return observation


def load_well(welfile):
    """Load a well from file.

    This reads a well from a csv file.

    Parameters
    ----------
    welfile : :class:`str`
        Path to the file
    """
    # default version string
    version_string = "1.0.0"
    try:
        with zipfile.ZipFile(welfile, "r") as zfile:
            info = TxtIO(zfile.open("info.csv"))
            data = csv.reader(info)
            first_line = _nextr(data)
            if first_line[0] == "wtp-version":
                version_string = first_line[1]
                header = _nextr(data)
            else:
                header = first_line
            version = version_parse(version_string)
            _check_version(version)
            if header[0] != "Well":
                raise ValueError(
                    f"load_well: expected 'Well' but got '{header[0]}'"
                )
            name = next(data)[1]
            # radius
            radf = next(data)[1]
            rad = load_var(TxtIO(zfile.open(radf)))
            # coordinates
            coordf = next(data)[1]
            coord = load_var(TxtIO(zfile.open(coordf)))
            # well depth
            welldf = next(data)[1]
            welld = load_var(TxtIO(zfile.open(welldf)))
            # aquifer depth
            aquidf = next(data)[1]
            aquid = load_var(TxtIO(zfile.open(aquidf)))
            # read screensize implemented in v1.1
            screend = None
            if version.release >= (1, 1):
                screenf = next(data)[1]
                screend = load_var(TxtIO(zfile.open(screenf)))

        well = varlib.Well(name, rad, coord, welld, aquid, screend)
    except Exception as exc:
        raise LoadError(f"load_well: couldn't load well '{welfile}'") from exc
    return well


def load_campaign(cmpfile):
    """Load a campaign from file.

    This reads a campaign from a csv file.

    Parameters
    ----------
    cmpfile : :class:`str`
        Path to the file
    """
    # default version string
    version_string = "1.0.0"
    try:
        with zipfile.ZipFile(cmpfile, "r") as zfile:
            info = TxtIO(zfile.open("info.csv"))
            data = csv.reader(info)
            first_line = _nextr(data)
            if first_line[0] == "wtp-version":
                version_string = first_line[1]
                header = _nextr(data)
            else:
                header = first_line
            version = version_parse(version_string)
            _check_version(version)
            if header[0] != "Campaign":
                raise ValueError(
                    f"load_campaign: expected 'Campaign' but got '{header[0]}'"
                )
            name = next(data)[1]
            description = next(data)[1]
            timeframe = next(data)[1]
            row = _nextr(data)
            if row[1] == "None":
                fieldsite = None
            else:
                fieldsite = load_fieldsite(BytIO(zfile.read(row[1])))
            wcnt = int(next(data)[1])
            wells = {}
            for __ in range(wcnt):
                row = _nextr(data)
                wells[row[0]] = load_well(BytIO(zfile.read(row[1])))

            tcnt = int(next(data)[1])
            tests = {}
            for __ in range(tcnt):
                row = _nextr(data)
                tests[row[0]] = load_test(BytIO(zfile.read(row[1])))

        campaign = campaignlib.Campaign(
            name, fieldsite, wells, tests, timeframe, description
        )
    except Exception as exc:
        raise LoadError(
            f"load_campaign: couldn't load campaign '{cmpfile}'"
        ) from exc
    return campaign


def load_fieldsite(fdsfile):
    """Load a field site from file.

    This reads a field site from a csv file.

    Parameters
    ----------
    fdsfile : :class:`str`
        Path to the file
    """
    # default version string
    version_string = "1.0.0"
    try:
        with zipfile.ZipFile(fdsfile, "r") as zfile:
            info = TxtIO(zfile.open("info.csv"))
            data = csv.reader(info)
            first_line = _nextr(data)
            if first_line[0] == "wtp-version":
                version_string = first_line[1]
                header = _nextr(data)
            else:
                header = first_line
            version = version_parse(version_string)
            _check_version(version)
            if header[0] != "Fieldsite":
                raise ValueError(
                    "load_fieldsite: expected 'Fieldsite' "
                    f"but got '{header[0]}'"
                )
            name = next(data)[1]
            description = next(data)[1]
            coordinfo = next(data)[1]
            if coordinfo == "None":
                coordinates = None
            else:
                coordinates = load_var(TxtIO(zfile.open(coordinfo)))
        fieldsite = campaignlib.FieldSite(name, description, coordinates)
    except Exception as exc:
        raise LoadError(
            f"load_fieldsite: couldn't load fieldsite '{fdsfile}'"
        ) from exc
    return fieldsite


def load_test(tstfile):
    """Load a test from file.

    This reads a test from a csv file.

    Parameters
    ----------
    tstfile : :class:`str`
        Path to the file
    """
    # default version string
    version_string = "1.0.0"
    try:
        with zipfile.ZipFile(tstfile, "r") as zfile:
            info = TxtIO(zfile.open("info.csv"))
            data = csv.reader(info)
            first_line = _nextr(data)
            if first_line[0] == "wtp-version":
                version_string = first_line[1]
                header = _nextr(data)
            else:
                header = first_line
            version = version_parse(version_string)
            _check_version(version)
            if header[0] != "Testtype":
                raise ValueError(
                    f"load_test: expected 'Testtype' but got '{header[0]}'"
                )
            if header[1] == "PumpingTest":
                routine = _load_pumping_test
            else:
                raise ValueError(f"load_test: unknown test type '{header[1]}'")
    except Exception as exc:
        raise LoadError(f"load_test: couldn't load test '{tstfile}'") from exc
    return routine(tstfile)


def _load_pumping_test(tstfile):
    """Load a pumping test from file.

    This reads a pumping test from a csv file.

    Parameters
    ----------
    tstfile : :class:`str`
        Path to the file
    """
    # default version string
    version_string = "1.0.0"
    try:
        with zipfile.ZipFile(tstfile, "r") as zfile:
            info = TxtIO(zfile.open("info.csv"))
            data = csv.reader(info)
            first_line = _nextr(data)
            if first_line[0] == "wtp-version":
                version_string = first_line[1]
                header = _nextr(data)
            else:
                header = first_line
            version = version_parse(version_string)
            _check_version(version)
            if header[1] != "PumpingTest":
                raise ValueError(
                    f"load_test: expected 'PumpingTest' but got '{header[1]}'"
                )
            name = next(data)[1]
            description = next(data)[1]
            timeframe = next(data)[1]
            pumpingwell = next(data)[1]
            rate_raw = TxtIO(zfile.open(next(data)[1]))
            try:
                pumpingrate = load_var(rate_raw)
            except Exception:
                pumpingrate = load_obs(rate_raw)
            aquiferdepth = load_var(TxtIO(zfile.open(next(data)[1])))
            aquiferradius = load_var(TxtIO(zfile.open(next(data)[1])))
            obscnt = int(next(data)[1])
            observations = {}
            for __ in range(obscnt):
                row = _nextr(data)
                observations[row[0]] = load_obs(BytIO(zfile.read(row[1])))

        pumpingtest = testslib.PumpingTest(
            name,
            pumpingwell,
            pumpingrate,
            observations,
            aquiferdepth,
            aquiferradius,
            description,
            timeframe,
        )
    except Exception as exc:
        raise LoadError(
            f"load_test: couldn't load pumpingtest '{tstfile}'"
        ) from exc
    return pumpingtest

GeoStat-Framework / welltestpy / 10691143420

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous