internship/swash/processing/read_swash.py

import logging
import subprocess
import tempfile

import numpy as np


log = logging.getLogger("read_swash")

class ReadSwash:
    def __init__(self):
        self._n_x = None
        self._n_t = None
        self._t = None
        self._x = None

    @classmethod
    def read_nohead(cls, path):
        log.info(f"Replacing \\s with \\n in '{path}'")
        subprocess.run(("sed", "-i", r"s/\s\+/\n/g", path))
        log.info(f"Loading '{path}'")
        return np.loadtxt(path)

    def read_time(self, path):
        self._t = np.unique(self.read_nohead(path))
        self._n_t = self._t.size
        return self.t

    def read_x(self, path):
        self._x = np.unique(self.read_nohead(path))
        self._n_x = self._x.size
        return self.x

    def read_scalar(self, path, const=False):
        if const:
            return self.read_nohead(path).reshape((self._n_t, self._n_x))[0, :]
        return self.read_nohead(path).reshape((self._n_t, self._n_x))

    def read_const(self, path):
        return self.read_scalar(path, const=True)

    def read_vector(self, path):
        return self.read_nohead(path).reshape((self._n_t, 2, self._n_x))

    def read_scalar_lay(self, path):
        return self.read_nohead(path).reshape((self._n_t, -1, self._n_x))

    def read_vector_lay(self, path):
        return self.read_nohead(path).reshape((self._n_t, -1, 2, self._n_x))

    @property
    def t(self):
        return self._t

    @property
    def x(self):
        return self._x