1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298
|
from . import _version
from . import exceptions as ex
from . import utils
from ._graphblas import ffi, lib
import struct
import platform
_is_osx_arm64 = platform.machine() == "arm64"
_is_ppc64le = platform.machine() == "ppc64le"
_c_float = ffi.typeof("float")
_c_double = ffi.typeof("double")
# It is strongly recommended to use the non-variadic version of functions to be
# compatible with the most number of architectures. For example, you should use
# GxB_Matrix_Option_get_INT32 instead of GxB_Matrix_Option_get.
if _is_osx_arm64 or _is_ppc64le:
def vararg(val):
# Interpret float as int32 and double as int64
# https://devblogs.microsoft.com/oldnewthing/20220823-00/?p=107041
tov = ffi.typeof(val)
if tov == _c_float:
val = struct.unpack("l", struct.pack("f", val))[0]
val = ffi.cast("int64_t", val)
elif tov == _c_double:
val = struct.unpack("q", struct.pack("d", val))[0]
val = ffi.cast("int64_t", val)
# Cast variadic argument as char * to force it onto the stack where ARM64 expects it
# https://developer.apple.com/documentation/xcode/writing-arm64-code-for-apple-platforms
#
# The same fix *may* work for ppc64le
return ffi.cast("char *", val)
else:
def vararg(val):
return val
def is_initialized():
"""Is GraphBLAS initialized via GrB_init or GxB_init?"""
mode = ffi.new("int32_t*")
return lib.GxB_Global_Option_get_INT32(lib.GxB_MODE, mode) != lib.GrB_PANIC
def supports_complex():
"""Does this package support complex numbers?"""
return hasattr(lib, "GrB_FC64") or hasattr(lib, "GxB_FC64")
def initialize(*, blocking=False, memory_manager="numpy"):
"""Initialize GraphBLAS via GrB_init or GxB_init.
This must be called before any other GraphBLAS functions are called.
A RuntimeError will be raised if called more than once.
Parameters
----------
blocking : bool, optional
Whether to call init with GrB_BLOCKING or GrB_NONBLOCKING.
Default is False.
memory_manager : {'numpy', 'c'}, optional
Choose which malloc/free functions to use. 'numpy' uses numpy's
allocators, which makes it safe to perform zero-copy to and from numpy,
and allows Python to track memory usage via tracemalloc (if enabled).
'c' uses the default allocators. Default is 'numpy'.
The global variable `suitesparse_graphblas.is_initialized` indicates whether
GraphBLAS has been initialized.
"""
if is_initialized():
raise RuntimeError("GraphBLAS is already initialized! Unable to initialize again.")
blocking = lib.GrB_BLOCKING if blocking else lib.GrB_NONBLOCKING
memory_manager = memory_manager.lower()
if memory_manager == "numpy":
utils.call_gxb_init(ffi, lib, blocking)
elif memory_manager == "c":
lib.GrB_init(blocking)
else:
raise ValueError(f'memory_manager argument must be "numpy" or "c"; got: {memory_manager!r}')
# See: https://github.com/GraphBLAS/python-suitesparse-graphblas/issues/40
for attr in dir(lib):
getattr(lib, attr)
def libget(name):
"""Helper to get items from GraphBLAS which might be GrB or GxB"""
try:
return getattr(lib, name)
except AttributeError:
ext_name = f"GxB_{name[4:]}"
try:
return getattr(lib, ext_name)
except AttributeError:
pass
raise
bool_types = frozenset((lib.GrB_BOOL,))
signed_integer_types = frozenset(
(
lib.GrB_INT8,
lib.GrB_INT16,
lib.GrB_INT32,
lib.GrB_INT64,
)
)
unsigned_integer_types = frozenset(
(
lib.GrB_UINT8,
lib.GrB_UINT16,
lib.GrB_UINT32,
lib.GrB_UINT64,
)
)
integer_types = signed_integer_types | unsigned_integer_types
real_types = frozenset(
(
lib.GrB_FP32,
lib.GrB_FP64,
)
)
if supports_complex():
complex_types = frozenset(
(
lib.GxB_FC32,
lib.GxB_FC64,
)
)
else:
complex_types = frozenset()
grb_types = bool_types | integer_types | real_types | complex_types
_error_code_lookup = {
# Warning
lib.GrB_NO_VALUE: ex.NoValue,
# API Errors
lib.GrB_UNINITIALIZED_OBJECT: ex.UninitializedObject,
lib.GrB_INVALID_OBJECT: ex.InvalidObject,
lib.GrB_NULL_POINTER: ex.NullPointer,
lib.GrB_INVALID_VALUE: ex.InvalidValue,
lib.GrB_INVALID_INDEX: ex.InvalidIndex,
lib.GrB_DOMAIN_MISMATCH: ex.DomainMismatch,
lib.GrB_DIMENSION_MISMATCH: ex.DimensionMismatch,
lib.GrB_OUTPUT_NOT_EMPTY: ex.OutputNotEmpty,
lib.GrB_EMPTY_OBJECT: ex.EmptyObject,
# Execution Errors
lib.GrB_OUT_OF_MEMORY: ex.OutOfMemory,
lib.GrB_INSUFFICIENT_SPACE: ex.InsufficientSpace,
lib.GrB_INDEX_OUT_OF_BOUNDS: ex.IndexOutOfBound,
lib.GrB_PANIC: ex.Panic,
lib.GrB_NOT_IMPLEMENTED: ex.NotImplementedException,
# GxB Errors
lib.GxB_EXHAUSTED: StopIteration,
}
GrB_SUCCESS = lib.GrB_SUCCESS
GrB_NO_VALUE = lib.GrB_NO_VALUE
_error_func_lookup = {
"struct GB_Type_opaque *": lib.GrB_Type_error,
"struct GB_UnaryOp_opaque *": lib.GrB_UnaryOp_error,
"struct GB_BinaryOp_opaque *": lib.GrB_BinaryOp_error,
"struct GB_SelectOp_opaque *": lib.GxB_SelectOp_error,
"struct GB_Monoid_opaque *": lib.GrB_Monoid_error,
"struct GB_Semiring_opaque *": lib.GrB_Semiring_error,
"struct GB_Scalar_opaque *": lib.GxB_Scalar_error,
"struct GB_Matrix_opaque *": lib.GrB_Matrix_error,
"struct GB_Vector_opaque *": lib.GrB_Vector_error,
"struct GB_Descriptor_opaque *": lib.GrB_Descriptor_error,
}
def check_status(obj, response_code):
"""Check the return code of the GraphBLAS function.
If the operation was successful, return None.
If the operation returned no value return `exceptions.NoValue`.
Otherwise it is an error, lookup the exception and the error
description, and throw the exception.
"""
if response_code == GrB_SUCCESS:
return
if response_code == GrB_NO_VALUE:
return ex.NoValue
if ffi.typeof(obj).item.kind == "pointer":
obj = obj[0]
cname = ffi.typeof(obj).cname
error_func = _error_func_lookup.get(cname)
if error_func is None:
raise TypeError(f"Unknown cname {cname} looking up error string.")
string = ffi.new("char**")
error_func(string, obj)
text = ffi.string(string[0]).decode()
raise _error_code_lookup[response_code](text)
class burble:
"""Control diagnostic output, and may be used as a context manager.
Set up and simple usage:
>>> from suitesparse_graphblas import burble, lib, matrix
>>>
>>> A = matrix.new(lib.GrB_BOOL, 3, 3)
>>> burble.is_enabled
False
>>> burble.enable()
>>> burble.is_enabled
True
>>> burble.disable()
Example with explicit enable and disable:
>>> burble.enable()
>>> n = matrix.nvals(A)
[ GrB_Matrix_nvals
1.91e-06 sec ]
>>> burble.disable()
Example as a context manager:
>>> with burble():
>>> n = matrix.nvals(A)
[ GrB_Matrix_nvals
1.91e-06 sec ]
"""
def __init__(self):
self._states = []
@property
def is_enabled(self):
"""Is burble enabled?"""
val_ptr = ffi.new("int32_t*")
info = lib.GxB_Global_Option_get_INT32(lib.GxB_BURBLE, val_ptr)
if info != lib.GrB_SUCCESS:
raise _error_code_lookup[info](
"Failed to get burble status (has GraphBLAS been initialized?"
)
return val_ptr[0]
def enable(self):
"""Enable diagnostic output"""
info = lib.GxB_Global_Option_set_INT32(lib.GxB_BURBLE, ffi.cast("int32_t", 1))
if info != lib.GrB_SUCCESS:
raise _error_code_lookup[info](
"Failed to enable burble (has GraphBLAS been initialized?"
)
def disable(self):
"""Disable diagnostic output"""
info = lib.GxB_Global_Option_set_INT32(lib.GxB_BURBLE, ffi.cast("int32_t", 0))
if info != lib.GrB_SUCCESS:
raise _error_code_lookup[info](
"Failed to disable burble (has GraphBLAS been initialized?"
)
def __enter__(self):
is_enabled = self.is_enabled
if not is_enabled:
self.enable()
self._states.append(is_enabled)
return self
def __exit__(self, type_, value, traceback):
is_enabled = self._states.pop()
if not is_enabled:
self.disable()
def __reduce__(self):
return "burble"
def __repr__(self):
return f"<burble is_enabled={self.is_enabled}>"
burble = burble()
__version__ = _version.get_versions()["version"]
|