data_for_test.py

import numpy as np
import pytest
import utilities as u
import weno_reference_implementation as ref


# {{{ FluxDataSingle

class FluxDataSingle:
    nvars = 5
    ndim = 3
    dirs = {"x": 1, "y": 2, "z": 3}

    def __init__(self, states_str, direction):

        self.direction = self.dirs[direction]
        self.dir_internal = self.direction-1

        self.metrics = np.array([np.identity(self.ndim) for i in range(6)],
                dtype=np.float64, order="F")
        self.jacobians = np.repeat(1.0, 6)

        # FIXME: should be computed directly from the metrics and jacobians
        self.frozen_metrics = np.mean(self.metrics[2:4], axis=0)
        self.frozen_jacobian = np.mean(self.jacobians[2:4], axis=0)
        self.combined_frozen_metrics = 1.0

        self.state_pair = self.swap_array_rows(
                u.transposed_array_from_string(states_str), self.dir_internal)
        self.states = u.expand_to_6(self.state_pair)

        # FIXME: these should be generalized fluxes
        # FIXME: make a clear distinction between fluxes in physical and
        # generalized coordinates
        self.flux_pair = ref.pointwise_fluxes(
                self.state_pair)[:,:,self.dir_internal].T.copy(order="F")
        self.fluxes = ref.pointwise_fluxes(
                self.states)[:,:,self.dir_internal].T.copy(order="F")

        self.lam_pointwise = ref.lambda_pointwise(
                self.states, self.metrics, self.dir_internal)
        self.R, self.R_inv, self.lam_roe = ref.roe_eigensystem(
                self.state_pair, self.frozen_metrics, self.direction)
        self.wavespeeds = ref.wavespeeds(self.lam_pointwise, self.lam_roe)

        self.char_fluxes_pos, self.char_fluxes_neg = ref.split_char_fluxes(
                self.states, self.wavespeeds,
                self.frozen_metrics[self.dir_internal], self.frozen_jacobian,
                self.R_inv)

        self.oscillation_pos = ref.oscillation(self.char_fluxes_pos)
        self.oscillation_neg = ref.oscillation(self.char_fluxes_neg[:,::-1])

        self.weno_weights_pos = ref.weno_weights(
                self.oscillation_pos, self.combined_frozen_metrics)
        self.weno_weights_neg = ref.weno_weights(
                self.oscillation_neg, self.combined_frozen_metrics)

        self.consistent = ref.consistent_part(self.fluxes)
        self.dissipation_pos = ref.dissipation_part(
                self.R, self.char_fluxes_pos, self.weno_weights_pos, 1)
        self.dissipation_neg = ref.dissipation_part(
                self.R, self.char_fluxes_neg, self.weno_weights_neg, -1)

        self.weno_flux = ref.weno_flux(
                self.consistent, self.dissipation_pos, self.dissipation_neg)

    def swap_array_rows(self, arr, d):
        p = self.permutation(d)
        arr[p, :] = arr[[1, 2, 3], :]
        return arr

    def permutation(self, d):
        return [(d+i) % 3 + 1 for i in range(3)]

# }}}


single_data = {}

single_data["Case flat:x"] = FluxDataSingle(
        states_str="1 1 1 1 5.5,1 1 1 1 5.5", direction="x")
single_data["Case flat:y"] = FluxDataSingle(
        states_str="1 1 1 1 5.5,1 1 1 1 5.5", direction="y")
single_data["Case flat:z"] = FluxDataSingle(
        states_str="1 1 1 1 5.5,1 1 1 1 5.5", direction="z")

single_data["Case a:x"] = FluxDataSingle(
        states_str="2 4 4 4 20,1 1 1 1 5.5", direction="x")
single_data["Case a:y"] = FluxDataSingle(
        states_str="2 4 4 4 20,1 1 1 1 5.5", direction="y")
single_data["Case a:z"] = FluxDataSingle(
        states_str="2 4 4 4 20,1 1 1 1 5.5", direction="z")

single_data["Case b:x"] = FluxDataSingle(
        states_str="1 -1 -1 -1 5.5,2 -4 -4 -4 20", direction="x")
single_data["Case b:y"] = FluxDataSingle(
        states_str="1 -1 -1 -1 5.5,2 -4 -4 -4 20", direction="y")
single_data["Case b:z"] = FluxDataSingle(
        states_str="1 -1 -1 -1 5.5,2 -4 -4 -4 20", direction="z")

single_data["Case c:x"] = FluxDataSingle(
        states_str="2 4 8 12 64,1 1 2 3 11", direction="x")
single_data["Case c:y"] = FluxDataSingle(
        states_str="2 8 12 4 64,1 2 3 1 11", direction="y")
single_data["Case c:z"] = FluxDataSingle(
        states_str="2 12 4 8 64,1 3 1 2 11", direction="z")

single_data["Case d:x"] = FluxDataSingle(
        states_str="1 -1 -2 -3 11,2 -4 -8 -12 64", direction="x")
single_data["Case d:y"] = FluxDataSingle(
        states_str="1 -2 -3 -1 11,2 -8 -12 -4 64", direction="y")
single_data["Case d:z"] = FluxDataSingle(
        states_str="1 -3 -1 -2 11,2 -12 -4 -8 64", direction="z")


@pytest.fixture(scope="session", params=[
    "Case flat:x", "Case flat:y", "Case flat:z",
    "Case a:x", "Case a:y", "Case a:z",
    "Case b:x", "Case b:y", "Case b:z",
    "Case c:x", "Case c:y", "Case c:z",
    "Case d:x", "Case d:y", "Case d:z"])
def flux_test_data_fixture(request):
    return single_data[request.param]


# vim: foldmethod=marker