# These tests are very slow, since they force creation of all

# numpy unary, binary, monoid, and semiring objects.

import itertools

import sys

import numpy as np

import pytest

from packaging.version import parse

import graphblas as gb

import graphblas.binary.numpy as npbinary

import graphblas.monoid.numpy as npmonoid

import graphblas.op.numpy as npop

import graphblas.semiring.numpy as npsemiring

import graphblas.unary.numpy as npunary

from graphblas import Vector, backend, config

from graphblas.core import _supports_udfs as supports_udfs

from graphblas.dtypes import _supports_complex

from .conftest import compute, shouldhave

is_win = sys.platform.startswith("win")

suitesparse = backend == "suitesparse"

def test_numpyops_dir():

udf_or_mapped = supports_udfs or config["mapnumpy"]

assert ("exp2" in dir(npunary)) == udf_or_mapped

assert ("logical_and" in dir(npbinary)) == udf_or_mapped

assert ("logaddexp" in dir(npmonoid)) == supports_udfs

assert ("add_add" in dir(npsemiring)) == udf_or_mapped

@pytest.mark.skipif("not supports_udfs")

@pytest.mark.slow

def test_bool_doesnt_get_too_large():

a = Vector.from_coo([0, 1, 2, 3], [True, False, True, False])

b = Vector.from_coo([0, 1, 2, 3], [True, True, False, False])

if gb.config["mapnumpy"]:

with pytest.raises(KeyError, match="plus does not work with BOOL"):

z = a.ewise_mult(b, gb.monoid.numpy.add).new()

else:

z = a.ewise_mult(b, gb.monoid.numpy.add).new()

_x, y = z.to_coo()

np.testing.assert_array_equal(y, (True, True, True, False))

def func(x): # pragma: no cover (numba)

return np.add(x, x)

op = gb.core.operator.UnaryOp.register_anonymous(func)

z = a.apply(op).new()

_x, y = z.to_coo()

np.testing.assert_array_equal(y, (True, False, True, False))

@pytest.mark.skipif("not supports_udfs")

# On a broken numba x numpy combo (numba #8478, still within our supported floor)

# the module sets _fmin_is_float=True and does not populate the integer fmax/fmin

# identities, so skip rather than KeyError on that config.

@pytest.mark.skipif("npmonoid._fmin_is_float")

@pytest.mark.parametrize("dtype", ["INT16", "INT32", "INT64"])

def test_numpy_fmax_fmin_identity_outside_int8(dtype):

# Regression: the fmax identity for INT16/INT32/INT64 was set to int8's min

# (-128) instead of the dtype's own min, so a fmax-reduce over values all

# below -128 wrongly returned -128 rather than the true maximum. fmin always

# used the correct per-dtype max; it is checked here as a control. Only the

# mapnumpy=False path (UDF monoids) reads these identities; mapnumpy=True

# resolves fmax/fmin to the builtin max/min, which were already correct.

np_dtype = getattr(np, dtype.lower())

lo = int(np.iinfo(np_dtype).min)

hi = int(np.iinfo(np_dtype).max)

# Pin the identity table directly (deterministic, independent of config).

assert npmonoid._monoid_identities["fmax"][dtype] == lo

assert npmonoid._monoid_identities["fmin"][dtype] == hi

# Exercise the UDF monoids end-to-end. Force a fresh mapnumpy=False build so

# this hits the affected path regardless of the session's random mapnumpy

# (the resolved monoid is cached on the module after first access).

orig = config["mapnumpy"]

# The monoid is built from whatever binary.numpy has cached (monoid/numpy.py

# does getattr(_binary.numpy, name)), so the binary cache must be cleared too:

# under mapnumpy=True it holds the *builtin* binary.max/min, whose typed ops

# are TypedBuiltinBinaryOp, which has no `_monoid` slot for Monoid to write

# its back-reference to. Leaving it cached makes this test raise AttributeError

# whenever an earlier test in the session already resolved binary.numpy.fmax.

# op.numpy is cleared as well because registering the UDF binary op populates

# it as a side effect. Save every entry first so the exact prior state, cached

# or absent, is restored for later tests.

names = ("fmax", "fmin")

modules = (npmonoid, npbinary, npop)

saved = [(module, name, module.__dict__.get(name)) for module in modules for name in names]

config.set(mapnumpy=False)

for module, name, _ in saved:

module.__dict__.pop(name, None)

try:

below = [-1000, -2000, -3000] # all below int8's min, valid for INT16+

above = [1000, 2000, 3000] # all above int8's max

v_below = Vector.from_coo([0, 1, 2], below, dtype=dtype)

v_above = Vector.from_coo([0, 1, 2], above, dtype=dtype)

assert v_below.reduce(npmonoid.fmax).new().value == max(below) # -1000

assert v_above.reduce(npmonoid.fmin).new().value == min(above) # 1000

assert npmonoid.fmax[dtype].identity == lo

assert npmonoid.fmin[dtype].identity == hi

finally:

config.set(mapnumpy=orig)

for module, name, obj in saved:

if obj is None:

module.__dict__.pop(name, None)

else:

module.__dict__[name] = obj

def test_numpy_monoid_identity_matches_builtin():

# Invariant: mapnumpy must not change a numpy monoid's identity. For every op

# in the numpy->builtin mapping, the numpy identity (read straight from

# _monoid_identities, which register_new uses verbatim on the mapnumpy=False

# path) must equal the builtin monoid's identity for every shared dtype,

# compared cast-to-dtype so benign encodings agree (bitwise_and's -1 and

# band's 255 are both all-ones); only a genuinely wrong value (e.g. logical_or

# True vs lor False) is flagged. This guards the whole table and caught both

# fmax/fmin and logical_or. Reading the table directly keeps it deterministic

# and mutates no operator caches, so it cannot desync monoid/binary/op.numpy.

identities = npmonoid._monoid_identities

mismatches = []

for np_name, gb_name in npmonoid._numpy_to_graphblas.items():

table = identities[np_name]

gb_monoid = getattr(gb.monoid, gb_name)

for dtype in gb_monoid.types:

if isinstance(table, dict):

if dtype.name not in table:

continue # numpy op does not define this dtype

np_id = table[dtype.name]

else:

np_id = table # scalar identity applies to every dtype

gb_id = gb_monoid[dtype].identity

np_cast = np.asarray(np_id).astype(dtype.np_type)

gb_cast = np.asarray(gb_id).astype(dtype.np_type)

if np_cast != gb_cast:

mismatches.append(f"{np_name}[{dtype.name}]={np_id!r} vs {gb_name}={gb_id!r}")

assert not mismatches, mismatches

def test_numpy_monoid_unmapped_identity_consistency():

# The six numpy monoids with no builtin counterpart (gcd, hypot, logaddexp,

# logaddexp2, maximum, minimum) are absent from _numpy_to_graphblas, so the

# sibling test above never reads their identities. Guard the two that have a

# byte-identical builtin-backed twin: numpy's maximum/minimum share fmax's/

# fmin's identity table exactly (nan vs. non-nan changes only propagation of

# a present value, not the neutral element). fmax/fmin are themselves pinned

# against builtin max/min by the sibling test, so asserting maximum == fmax

# and minimum == fmin transitively validates maximum/minimum. Read the table

# directly and compare cast-to-dtype, matching the sibling test: no config

# toggle, no operator-cache mutation, no monoid resolution.

identities = npmonoid._monoid_identities

mismatches = []

for np_name, twin in (("maximum", "fmax"), ("minimum", "fmin")):

table = identities[np_name]

twin_table = identities[twin]

# fmax/fmin drop their integer keys on a broken numba/numpy combo

# (_fmin_is_float), so intersect rather than assume every dtype is present.

shared = table.keys() & twin_table.keys()

assert shared, f"{np_name}/{twin} share no dtypes"

for name in sorted(shared):

np_type = gb.dtypes.lookup_dtype(name).np_type

np_cast = np.asarray(table[name]).astype(np_type)

twin_cast = np.asarray(twin_table[name]).astype(np_type)

if np_cast != twin_cast:

mismatches.append(

f"{np_name}[{name}]={table[name]!r} vs {twin}[{name}]={twin_table[name]!r}"

)

assert not mismatches, mismatches

# Light self-check for the log-add monoids: their neutral element is -inf

# (logaddexp(-inf, x) == x). gcd and hypot are deliberately left out of any

# functional identity check: their conventional identity is 0, but numpy's

# gcd/hypot are not sign-preserving (gcd(0, -5) == 5 != -5), so

# f(identity, x) == x is not a valid invariant for them even though 0 is the

# correct identity. So only their presence, not a value, is asserted here.

for np_name in ("logaddexp", "logaddexp2"):

table = identities[np_name]

assert table, f"{np_name} identity table is empty"

for name, value in table.items():

assert value == -np.inf, f"{np_name}[{name}]={value!r} != -inf"

for np_name in ("gcd", "hypot"):

assert identities[np_name], f"{np_name} identity table is empty"

@pytest.mark.slow

def test_npunary():

L = list(range(5))

data = [

[Vector.from_coo([0, 1], [True, False]), np.array([True, False])],

[Vector.from_coo(L, L), np.array(L, dtype=np.int64)],

[Vector.from_coo(L, L, dtype="float64"), np.array(L, dtype=np.float64)],

]

if _supports_complex:

data.append(

[Vector.from_coo(L, L, dtype="FC64"), np.array(L, dtype=np.complex128)],

)

blocklist = {

"BOOL": {"negative", "positive", "reciprocal", "sign"},

"INT64": {"reciprocal"},

"FC64": {"ceil", "floor", "trunc"},

}

if suitesparse and is_win and gb.config["mapnumpy"]:

# asin and asinh are known to be wrong in SuiteSparse:GraphBLAS

# due to limitation of MSVC with complex

blocklist["FC64"].update({"arcsin", "arcsinh"})

blocklist["FC32"] = {"arcsin", "arcsinh"}

if shouldhave(gb.binary, "isclose"):

isclose = gb.binary.isclose(1e-6, 0)

else:

isclose = None

for gb_input, np_input in data:

for unary_name in sorted(npunary._unary_names & npunary.__dir__()):

op = getattr(npunary, unary_name)

if gb_input.dtype not in op.types or unary_name in blocklist.get(

gb_input.dtype.name, ()

):

continue

if gb_input.dtype.name.startswith("FC"):

# There are some nasty branch cuts as 1

gb_input = gb_input.dup()

gb_input[1] = 1.1 + 1.2j

np_input = np_input.copy()

np_input[1] = 1.1 + 1.2j

with np.errstate(divide="ignore", over="ignore", under="ignore", invalid="ignore"):

gb_result = gb_input.apply(op).new()

if gb_input.dtype == "BOOL" and gb_result.dtype == "FP32":

np_result = getattr(np, unary_name)(np_input, dtype="float32")

compare_op = isclose

else:

np_result = getattr(np, unary_name)(np_input)

if gb_result.dtype.name.startswith("F"):

compare_op = isclose

else:

compare_op = npbinary.equal

np_result = Vector.from_coo(

list(range(np_input.size)), list(np_result), dtype=gb_result.dtype

)

assert gb_result.nvals == np_result.size

if compare_op is None:

continue # FLAKY COVERAGE

match = gb_result.ewise_mult(np_result, compare_op).new()

if gb_result.dtype.name.startswith("F"):

match(accum=gb.binary.lor) << gb_result.apply(npunary.isnan)

compare = match.reduce(gb.monoid.land).new()

if not compare: # pragma: no cover (debug)

import numba

if (

unary_name == "sign"

and np.__version__.startswith("2.")

and parse(numba.__version__) < parse("0.61.0")

):

# numba <0.61.0 does not match numpy 2.0

continue

print(unary_name, gb_input.dtype)

print(compute(gb_result))

print(np_result)

assert compare

@pytest.mark.slow

def test_npbinary():

values1 = [0, 0, 1, 1, 2, 5]

values2 = [0, 1, 0, 1, 3, 8]

index = list(range(len(values1)))

data = [

[

[Vector.from_coo(index, values1), Vector.from_coo(index, values2)],

[np.array(values1, dtype=np.int64), np.array(values2, dtype=np.int64)],

],

[

[

Vector.from_coo(index, values1, dtype="float64"),

Vector.from_coo(index, values2, dtype="float64"),

],

[np.array(values1, dtype=np.float64), np.array(values2, dtype=np.float64)],

],

[

[

Vector.from_coo([0, 1, 2, 3], [True, False, True, False]),

Vector.from_coo([0, 1, 2, 3], [True, True, False, False]),

],

[np.array([True, False, True, False]), np.array([True, True, False, False])],

],

]

if _supports_complex:

data.append(

[

[

Vector.from_coo(index, values1, dtype="FC64"),

Vector.from_coo(index, values2, dtype="FC64"),

],

[np.array(values1, dtype=np.complex128), np.array(values2, dtype=np.complex128)],

],

)

blocklist = {

"FP64": {"floor_divide"}, # numba/numpy difference for 1.0 / 0.0

"BOOL": {"gcd", "lcm", "subtract"}, # not supported by numpy

}

if shouldhave(gb.binary, "isclose"):

isclose = gb.binary.isclose(1e-7, 0)

else:

isclose = None

if shouldhave(npbinary, "equal"):

equal = npbinary.equal

else:

equal = gb.binary.eq

if shouldhave(npbinary, "isnan"):

isnan = npunary.isnan

else:

isnan = gb.unary.isnan

if shouldhave(npbinary, "isinf"):

isinf = npunary.isinf

else:

isinf = gb.unary.isinf

for (gb_left, gb_right), (np_left, np_right) in data:

for binary_name in sorted(npbinary._binary_names & npbinary.__dir__()):

op = getattr(npbinary, binary_name)

if gb_left.dtype not in op.types or binary_name in blocklist.get(

gb_left.dtype.name, ()

):

continue

if is_win and binary_name == "ldexp":

# On Windows, the second argument must be int32 or less (I'm not sure why)

np_right = np_right.astype(np.int32)

with np.errstate(divide="ignore", over="ignore", under="ignore", invalid="ignore"):

gb_result = gb_left.ewise_mult(gb_right, op).new()

try:

if gb_left.dtype == "BOOL" and gb_result.dtype == "FP32":

np_result = getattr(np, binary_name)(np_left, np_right, dtype="float32")

compare_op = isclose

else:

np_result = getattr(np, binary_name)(np_left, np_right)

if binary_name == "arctan2":

compare_op = isclose

else:

compare_op = equal

except Exception: # pragma: no cover (debug)

print(f"Error computing numpy result for {binary_name}")

print(f"dtypes: ({gb_left.dtype}, {gb_right.dtype}) -> {gb_result.dtype}")

raise

np_result = Vector.from_coo(np.arange(np_left.size), np_result, dtype=gb_result.dtype)

assert gb_result.nvals == np_result.size

if compare_op is None:

continue # FLAKY COVERAGE

match = gb_result.ewise_mult(np_result, compare_op).new()

if gb_result.dtype.name.startswith("F"):

match(accum=gb.binary.lor) << gb_result.apply(isnan)

if gb_result.dtype.name.startswith("FC"):

# Divide by 0j sometimes result in different behavior, such as `nan` or `(inf+0j)`

match(accum=gb.binary.lor) << gb_result.apply(isinf)

compare = match.reduce(gb.monoid.land).new()

if not compare: # pragma: no cover (debug)

print(compare_op)

print(binary_name)

print(compute(gb_left))

print(compute(gb_right))

print(compute(gb_result))

print(np_result)

print((np_result - compute(gb_result)).new().to_coo()[1])

assert compare

@pytest.mark.slow

def test_npmonoid():

values1 = [0, 0, 1, 1, 2, 5]

values2 = [0, 1, 0, 1, 3, 8]

index = list(range(len(values1)))

data = [

[

[Vector.from_coo(index, values1), Vector.from_coo(index, values2)],

[np.array(values1, dtype=int), np.array(values2, dtype=int)],

],

[

[

Vector.from_coo(index, values1, dtype="float64"),

Vector.from_coo(index, values2, dtype="float64"),

],

[np.array(values1, dtype=np.float64), np.array(values2, dtype=np.float64)],

],

[

[

Vector.from_coo([0, 1, 2, 3], [True, False, True, False]),

Vector.from_coo([0, 1, 2, 3], [True, True, False, False]),

],

[np.array([True, False, True, False]), np.array([True, True, False, False])],

],

]

# Complex monoids not working yet (they segfault upon creation in gb.core.operators)

if _supports_complex:

data.append(

[

[

Vector.from_coo(index, values1, dtype="FC64"),

Vector.from_coo(index, values2, dtype="FC64"),

],

[

np.array(values1, dtype=np.complex128),

np.array(values2, dtype=np.complex128),

],

]

)

blocklist = {}

reduction_blocklist = {

"BOOL": {"add"},

}

for (gb_left, gb_right), (np_left, np_right) in data:

for binary_name in sorted(npmonoid._monoid_identities.keys() & npmonoid.__dir__()):

op = getattr(npmonoid, binary_name)

assert len(op.types) > 0, op.name

if gb_left.dtype not in op.types or binary_name in blocklist.get(

gb_left.dtype.name, ()

):

continue # FLAKY COVERAGE

with np.errstate(divide="ignore", over="ignore", under="ignore", invalid="ignore"):

gb_result = gb_left.ewise_mult(gb_right, op).new()

np_result = getattr(np, binary_name)(np_left, np_right)

np_result = Vector.from_coo(np.arange(np_left.size), np_result, dtype=gb_result.dtype)

assert gb_result.nvals == np_result.size

match = gb_result.ewise_mult(np_result, npbinary.equal).new()

if gb_result.dtype.name.startswith("F"):

match(accum=gb.binary.lor) << gb_result.apply(npunary.isnan)

compare = match.reduce(gb.monoid.land).new()

if not compare: # pragma: no cover (debug)

print(binary_name, gb_left.dtype)

print(compute(gb_result))

print(np_result)

assert compare

# numpy reductions don't have dtype-dependent identities, so results sometimes differ

if binary_name in reduction_blocklist.get(gb_left.dtype.name, ()):

continue

gb_result = gb_left.reduce(op).new()

np_result = getattr(np, binary_name).reduce(np_left)

assert gb_result.value == np_result

gb_result = gb_right.reduce(op).new()

np_result = getattr(np, binary_name).reduce(np_right)

assert gb_result.value == np_result

@pytest.mark.slow

def test_npsemiring():

for monoid_name, binary_name in itertools.product(

sorted(npmonoid._monoid_identities.keys() & npmonoid.__dir__()),

sorted(npbinary._binary_names & npbinary.__dir__()),

):

monoid = getattr(npmonoid, monoid_name)

binary = getattr(npbinary, binary_name)

name = monoid.name.split(".")[-1] + "_" + binary.name.split(".")[-1]

if name in {"eq_pow", "eq_minus"}:

continue

semiring = gb.core.operator.Semiring.register_anonymous(monoid, binary, name)

if len(semiring.types) == 0:

if not gb.config["mapnumpy"] and "logical" not in name:

assert not hasattr(npsemiring, semiring.name), name

else:

assert hasattr(npsemiring, f"{monoid_name}_{binary_name}"), (name, semiring.name)