def test_diff():
    pytest.importorskip("pexpect")

    from pymbolic.interop.maxima import diff
    from pymbolic import parse

    diff(parse("sqrt(x**2+y**2)"), parse("x"))

    def _parse_expr(self, expr):
        from pymbolic import parse, substitute
        parsed = parse(expr)

        # substitute in global constants
        parsed = substitute(parsed, self.constants)

        return parsed

    def eval(expr, source, center1, center2, target):
        from pymbolic import parse, evaluate
        context = {
                "s": source,
                "c1": center1,
                "c2": center2,
                "t": target,
                "norm": la.norm}

        return evaluate(parse(expr), context)

def parse_sympy(s):
    if isinstance(s, unicode):
        # Sympy is not spectacularly happy with unicode function names
        s = s.encode()

    from pymbolic import parse
    from pymbolic.sympy_interface import PymbolicToSympyMapper

    # use pymbolic because it has a semi-secure parser
    return PymbolicToSympyMapper()(parse(s))

def parse_sympy(s):
    if six.PY2:
        if isinstance(s, unicode):  # noqa -- has Py2/3 guard
            # Sympy is not spectacularly happy with unicode function names
            s = s.encode()

    from pymbolic import parse
    from pymbolic.interop.sympy import PymbolicToSympyMapper

    # use pymbolic because it has a semi-secure parser
    return PymbolicToSympyMapper()(parse(s))

    def get_expr_dataset(self, expression, description=None, unit=None):
        """Prepare a time-series dataset for a given expression.

        @arg expression: A C{pymbolic} expression that may involve
          the time-series variables and the constants in this :class:`LogManager`.
          If there is data from multiple ranks for a quantity occuring in
          this expression, an aggregator may have to be specified.
        @return: C{(description, unit, table)}, where C{table}
          is a list of tuples C{(tick_nbr, value)}.

        Aggregators are specified as follows:
        - C{qty.min}, C{qty.max}, C{qty.avg}, C{qty.sum}, C{qty.norm2}
        - C{qty[rank_nbr]}
        - C{qty.loc}
        """

        parsed = self._parse_expr(expression)
        parsed, dep_data = self._get_expr_dep_data(parsed)

        # aggregate table data
        for dd in dep_data:
            table = self.get_table(dd.name)
            table.sort(["step"])
            dd.table = table.aggregated(["step"], "value", dd.agg_func).data

        # evaluate unit and description, if necessary
        if unit is None:
            from pymbolic import substitute, parse

            unit_dict = dict((dd.varname, dd.qdat.unit) for dd in dep_data)
            from pytools import all
            if all(v is not None for v in six.itervalues(unit_dict)):
                unit_dict = dict((k, parse(v)) for k, v in six.iteritems(unit_dict))
                unit = substitute(parsed, unit_dict)
            else:
                unit = None

        if description is None:
            description = expression

        # compile and evaluate
        from pymbolic import compile
        compiled = compile(parsed, [dd.varname for dd in dep_data])

        data = []

        for key, values in _join_by_first_of_tuple(dd.table for dd in dep_data):
            try:
                data.append((key, compiled(*values)))
            except ZeroDivisionError:
                pass

        return (description, unit, data)

def test_strict_round_trip(knl):
    from pymbolic import parse
    from pymbolic.primitives import Quotient

    exprs = [
            2j,
            parse("x**y"),
            Quotient(1, 2),
            parse("exp(x)")
            ]
    for expr in exprs:
        result = knl.eval_expr(expr)
        round_trips_correctly = result == expr
        if not round_trips_correctly:
            print("ORIGINAL:")
            print("")
            print(expr)
            print("")
            print("POST-MAXIMA:")
            print("")
            print(result)
        assert round_trips_correctly

    def get_expr_dataset(self, expression, description=None, unit=None):
        """Prepare a time-series dataset for a given expression.

        @arg expression: A C{pymbolic} expression that may involve
          the time-series variables and the constants in this L{LogManager}.
          If there is data from multiple ranks for a quantity occuring in
          this expression, an aggregator may have to be specified.
        @return: C{(description, unit, table)}, where C{table} 
          is a list of tuples C{(tick_nbr, value)}.

        Aggregators are specified as follows:
        - C{qty.min}, C{qty.max}, C{qty.avg}, C{qty.sum}, C{qty.norm2}
        - C{qty[rank_nbr]
        """

        parsed = self._parse_expr(expression)
        parsed, dep_data = self._get_expr_dep_data(parsed)

        # aggregate table data
        for dd in dep_data:
            table = self.get_table(dd.name)
            table.sort(["step"])
            dd.table = table.aggregated(["step"], "value", dd.agg_func).data

        # evaluate unit and description, if necessary
        if unit is None:
            from pymbolic import substitute, parse

            unit = substitute(parsed,
                    dict((dd.varname, parse(dd.qdat.unit)) for dd in dep_data)
                    )

        if description is None:
            description = expression

        # compile and evaluate
        from pymbolic import compile
        compiled = compile(parsed, [dd.varname for dd in dep_data])

        return (description,
                unit,
                [(key, compiled(*values))
                    for key, values in _join_by_first_of_tuple(
                        dd.table for dd in dep_data)
                    ])

from pymbolic import parse, var
from pymbolic.mapper.dependency import DependencyMapper

x = var("x")
y = var("y")

expr2 = 3*x+5-y
expr = parse("3*x+5-y")

print expr
print expr2

dm = DependencyMapper()
print dm(expr)

def test_elliptic():
    """Test various properties of elliptic operators."""

    from hedge.tools import unit_vector

    def matrix_rep(op):
        h, w = op.shape
        mat = numpy.zeros(op.shape)
        for j in range(w):
            mat[:, j] = op(unit_vector(w, j))
        return mat

    def check_grad_mat():
        import pyublas

        if not pyublas.has_sparse_wrappers():
            return

        grad_mat = op.grad_matrix()

        # print len(discr), grad_mat.nnz, type(grad_mat)
        for i in range(10):
            u = numpy.random.randn(len(discr))

            mat_result = grad_mat * u
            op_result = numpy.hstack(op.grad(u))

            err = la.norm(mat_result - op_result) * la.norm(op_result)
            assert la.norm(mat_result - op_result) * la.norm(op_result) < 1e-5

    def check_matrix_tgt():
        big = num.zeros((20, 20), flavor=num.SparseBuildMatrix)
        small = num.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])
        print small
        from hedge._internal import MatrixTarget

        tgt = MatrixTarget(big, 4, 4)
        tgt.begin(small.shape[0], small.shape[1])
        print "YO"
        tgt.add_coefficients(4, 4, small)
        print "DUDE"
        tgt.finalize()
        print big

    import pymbolic

    v_x = pymbolic.var("x")
    truesol = pymbolic.parse("math.sin(x[0]**2*x[1]**2)")
    truesol_c = pymbolic.compile(truesol, variables=["x"])
    rhs = pymbolic.simplify(pymbolic.laplace(truesol, [v_x[0], v_x[1]]))
    rhs_c = pymbolic.compile(rhs, variables=["x", "el"])

    from hedge.mesh import TAG_ALL, TAG_NONE
    from hedge.mesh.generator import make_disk_mesh

    mesh = make_disk_mesh(r=0.5, max_area=0.1, faces=20)
    mesh = mesh.reordered_by("cuthill")

    from hedge.backends import CPURunContext

    rcon = CPURunContext()

    from hedge.tools import EOCRecorder

    eocrec = EOCRecorder()
    for order in [1, 2, 3, 4, 5]:
        for flux in ["ldg", "ip"]:
            from hedge.discretization.local import TriangleDiscretization

            discr = rcon.make_discretization(
                mesh, TriangleDiscretization(order), debug=discr_class.noninteractive_debug_flags()
            )

            from hedge.data import GivenFunction
            from hedge.models.poisson import PoissonOperator

            op = PoissonOperator(
                discr.dimensions,
                dirichlet_tag=TAG_ALL,
                dirichlet_bc=GivenFunction(lambda x, el: truesol_c(x)),
                neumann_tag=TAG_NONE,
            )

            bound_op = op.bind(discr)

            if order <= 3:
                mat = matrix_rep(bound_op)
                sym_err = la.norm(mat - mat.T)
                # print sym_err
                assert sym_err < 1e-12
                # check_grad_mat()

            from hedge.iterative import parallel_cg

            truesol_v = discr.interpolate_volume_function(lambda x, el: truesol_c(x))
            sol_v = -parallel_cg(
                rcon,
                -bound_op,
                bound_op.prepare_rhs(discr.interpolate_volume_function(rhs_c)),
                tol=1e-10,
#.........这里部分代码省略.........

def extract_subst(kernel, subst_name, template, parameters=()):
    """
    :arg subst_name: The name of the substitution rule to be created.
    :arg template: Unification template expression.
    :arg parameters: An iterable of parameters used in
        *template*, or a comma-separated string of the same.

    All targeted subexpressions must match ('unify with') *template*
    The template may contain '*' wildcards that will have to match exactly across all
    unifications.
    """

    if isinstance(template, str):
        from pymbolic import parse
        template = parse(template)

    if isinstance(parameters, str):
        parameters = tuple(
                s.strip() for s in parameters.split(","))

    var_name_gen = kernel.get_var_name_generator()

    # {{{ replace any wildcards in template with new variables

    def get_unique_var_name():
        based_on = subst_name+"_wc"

        result = var_name_gen(based_on)
        return result

    from loopy.symbolic import WildcardToUniqueVariableMapper
    wc_map = WildcardToUniqueVariableMapper(get_unique_var_name)
    template = wc_map(template)

    # }}}

    # {{{ deal with iname deps of template that are not independent_inames

    # (We call these 'matching_vars', because they have to match exactly in
    # every CSE. As above, they might need to be renamed to make them unique
    # within the kernel.)

    matching_vars = []
    old_to_new = {}

    for iname in (get_dependencies(template)
            - set(parameters)
            - kernel.non_iname_variable_names()):
        if iname in kernel.all_inames():
            # need to rename to be unique
            new_iname = var_name_gen(iname)
            old_to_new[iname] = var(new_iname)
            matching_vars.append(new_iname)
        else:
            matching_vars.append(iname)

    if old_to_new:
        template = (
                SubstitutionMapper(make_subst_func(old_to_new))
                (template))

    # }}}

    # {{{ gather up expressions

    expr_descriptors = []

    from loopy.symbolic import UnidirectionalUnifier
    unif = UnidirectionalUnifier(
            lhs_mapping_candidates=set(parameters) | set(matching_vars))

    def gather_exprs(expr, mapper):
        urecs = unif(template, expr)

        if urecs:
            if len(urecs) > 1:
                raise RuntimeError("ambiguous unification of '%s' with template '%s'"
                        % (expr, template))

            urec, = urecs

            expr_descriptors.append(
                    ExprDescriptor(
                        insn=insn,
                        expr=expr,
                        unif_var_dict=dict((lhs.name, rhs)
                            for lhs, rhs in urec.equations)))
        else:
            mapper.fallback_mapper(expr)
            # can't nest, don't recurse

    from loopy.symbolic import (
            CallbackMapper, WalkMapper, IdentityMapper)
    dfmapper = CallbackMapper(gather_exprs, WalkMapper())

    for insn in kernel.instructions:
        dfmapper(insn.assignees)
        dfmapper(insn.expression)

    for sr in six.itervalues(kernel.substitutions):
#.........这里部分代码省略.........

def test_latex_mapper():
    from pymbolic import parse
    from pymbolic.mapper.stringifier import LaTeXMapper, StringifyMapper

    tm = LaTeXMapper()
    sm = StringifyMapper()

    equations = []

    def add(expr):
        # Add an equation to the list of tests.
        equations.append(r"\[%s\] %% from: %s" % (tm(expr), sm(expr)))

    add(parse("a * b + c"))
    add(parse("f(a,b,c)"))
    add(parse("a ** b ** c"))
    add(parse("(a | b) ^ ~c"))
    add(parse("a << b"))
    add(parse("a >> b"))
    add(parse("a[i,j,k]"))
    add(parse("a[1:3]"))
    add(parse("a // b"))
    add(parse("not (a or b) and c"))
    add(parse("(a % b) % c"))
    add(parse("(a >= b) or (b <= c)"))
    add(prim.Min((1,)) + prim.Max((1, 2)))
    add(prim.Substitution(prim.Variable("x") ** 2, ("x",), (2,)))
    add(prim.Derivative(parse("x**2"), ("x",)))

    # Run LaTeX and ensure the file compiles.
    import os
    import tempfile
    import subprocess
    import shutil

    latex_dir = tempfile.mkdtemp("pymbolic")

    try:
        tex_file_path = os.path.join(latex_dir, "input.tex")

        with open(tex_file_path, "w") as tex_file:
            contents = LATEX_TEMPLATE % "\n".join(equations)
            tex_file.write(contents)

        try:
            subprocess.check_output(
                    ["latex",
                     "-interaction=nonstopmode",
                     "-output-directory=%s" % latex_dir,
                     tex_file_path],
                    universal_newlines=True)
        except FileNotFoundError:
            pytest.skip("latex command not found")
        except subprocess.CalledProcessError as err:
            assert False, str(err.output)

    finally:
        shutil.rmtree(latex_dir)

def test_stringifier_preserve_shift_order():
    for expr in [
            parse("(a << b) >> 2"),
            parse("a << (b >> 2)")
            ]:
        assert parse(str(expr)) == expr

        if stats_callback is not None:
            stats_callback(size,  self,
                    kernel_rec.kernel.prepared_timed_call(vectors[0]._grid, results[0]._block, *args))
        else:
            kernel_rec.kernel.prepared_async_call(vectors[0]._grid, results[0]._block, self.stream, *args)

        return results




if __name__ == "__main__":
    test_dtype = numpy.float32

    import pycuda.autoinit
    from pymbolic import parse
    expr = parse("2*x+3*y+4*z")
    print expr
    cexpr = CompiledVectorExpression(expr,
            lambda expr: (True, test_dtype),
            test_dtype)

    from pymbolic import var
    ctx = {
        var("x"): gpuarray.arange(5, dtype=test_dtype),
        var("y"): gpuarray.arange(5, dtype=test_dtype),
        var("z"): gpuarray.arange(5, dtype=test_dtype),
        }

    print cexpr(lambda expr: ctx[expr])