import itertools

import numpy as np

from .. import backend, binary, monoid, select, semiring, unary

from ..dtypes import _INDEX, FP64, INT64, lookup_dtype, unify

from ..exceptions import DimensionMismatch, GrB_NO_VALUE, NoValue, check_status, check_status_carg

from . import _supports_udfs, automethods, ffi, lib, utils

from .base import BaseExpression, BaseType, _check_mask, _is_recording, call

from .descriptor import lookup as descriptor_lookup

from .expr import _ALL_INDICES, AmbiguousAssignOrExtract, IndexerResolver, InfixExprBase, Updater

from .mask import Mask, StructuralMask, ValueMask

from .operator import (

UNKNOWN_OPCLASS,

_get_typed_op_from_exprs,

find_opclass,

get_semiring,

get_typed_op,

op_from_string,

)

from .scalar import (

_COMPLETE,

_MATERIALIZE,

Scalar,

ScalarExpression,

ScalarIndexExpr,

_as_scalar,

_scalar_index,

)

from .utils import (

_CArray,

_Pointer,

class_property,

ints_to_numpy_buffer,

normalize_values,

output_type,

values_to_numpy_buffer,

wrapdoc,

)

if backend == "suitesparse":

from .ss.vector import ss

ffi_new = ffi.new

# Custom recipes

def _v_add_m(updater, left, right, op):

full = Vector(left.dtype, right._ncols, name="v_full")

full(**updater.opts)[:] = 0

temp = left.outer(full, binary.first).new(

name="M_temp", mask=updater.kwargs.get("mask"), **updater.opts

)

updater << temp.ewise_add(right, op)

def _v_mult_m(updater, left, right, op):

updater << left.diag(name="M_temp").mxm(right, get_semiring(monoid.any, op))

def _v_union_m(updater, left, right, left_default, right_default, op):

full = Vector(left.dtype, right._ncols, name="v_full")

full(**updater.opts)[:] = 0

temp = left.outer(full, binary.first).new(

name="M_temp", mask=updater.kwargs.get("mask"), **updater.opts

)

updater << temp.ewise_union(right, op, left_default=left_default, right_default=right_default)

def _v_union_v(updater, left, right, left_default, right_default, op):

mask = updater.kwargs.get("mask")

opts = updater.opts

new_left = left.dup(op.type, clear=True)

new_left(mask=mask, **opts) << binary.second(right, left_default)

new_left(mask=mask, **opts) << binary.first(left | new_left)

new_right = right.dup(op.type2, clear=True)

new_right(mask=mask, **opts) << binary.second(left, right_default)

new_right(mask=mask, **opts) << binary.first(right | new_right)

updater << op(new_left & new_right)

def _reposition(updater, indices, chunk):

updater[indices] = chunk

def _select_mask(updater, obj, mask):

if updater.kwargs.get("mask") is None:

orig_kwargs = updater.kwargs

try:

if updater.kwargs.get("accum") is None:

updater.kwargs = dict(orig_kwargs, mask=mask, replace=True)

else:

updater.kwargs = dict(orig_kwargs, mask=mask)

updater << obj

finally:

updater.kwargs = orig_kwargs

else:

# Can we do any better depending on accum, replace, and type of masks?

updater << obj.dup(mask=mask)

def _isclose_recipe(self, other, rel_tol, abs_tol, **opts):

# x == y or abs(x - y) <= max(rel_tol * max(abs(x), abs(y)), abs_tol)

isequal = self.ewise_mult(other, binary.eq).new(bool, name="isclose", **opts)

if isequal._nvals != self._nvals:

return False

if type(isequal) is Vector:

val = isequal.reduce(monoid.land, allow_empty=False).new(**opts).value

else:

val = isequal.reduce_scalar(monoid.land, allow_empty=False).new(**opts).value

if val:

return True

# So we can use structural mask below

isequal(**opts) << select.value(isequal == True) # noqa: E712

# abs(x)

x = self.apply(unary.abs).new(FP64, mask=~isequal.S, **opts)

# abs(y)

y = other.apply(unary.abs).new(FP64, mask=~isequal.S, **opts)

# max(abs(x), abs(y))

x(**opts) << x.ewise_mult(y, binary.max)

max_x_y = x

# rel_tol * max(abs(x), abs(y))

max_x_y(**opts) << max_x_y.apply(binary.times, rel_tol)

# max(rel_tol * max(abs(x), abs(y)), abs_tol)

max_x_y(**opts) << max_x_y.apply(binary.max, abs_tol)

# x - y

y(~isequal.S, replace=True, **opts) << self.ewise_mult(other, binary.minus)

abs_x_y = y

# abs(x - y)

abs_x_y(**opts) << abs_x_y.apply(unary.abs)

# abs(x - y) <= max(rel_tol * max(abs(x), abs(y)), abs_tol)

isequal(**opts) << abs_x_y.ewise_mult(max_x_y, binary.le)

if isequal.ndim == 1:

return isequal.reduce(monoid.land, allow_empty=False).new(**opts).value

return isequal.reduce_scalar(monoid.land, allow_empty=False).new(**opts).value

class Vector(BaseType):

"""Create a new GraphBLAS Sparse Vector.

Parameters

----------

dtype :

Data type for elements in the Vector.

size : int

Size of the Vector.

name : str, optional

Name to give the Vector. This will be displayed in the ``__repr__``.

"""

__slots__ = "_size", "_parent"

ndim = 1

_name_counter = itertools.count()

def __new__(cls, dtype=FP64, size=0, *, name=None):

self = object.__new__(cls)

self.dtype = lookup_dtype(dtype)

size = _as_scalar(size, _INDEX, is_cscalar=True)

self.name = f"v_{next(Vector._name_counter)}" if name is None else name

self.gb_obj = ffi_new("GrB_Vector*")

call("GrB_Vector_new", [_Pointer(self), self.dtype, size])

self._size = size.value

self._parent = None

return self

@classmethod

def _from_obj(cls, gb_obj, dtype, size, *, parent=None, name=None):

self = object.__new__(cls)

self.name = f"v_{next(Vector._name_counter)}" if name is None else name

self.gb_obj = gb_obj

self.dtype = dtype

self._size = size

self._parent = parent

return self

def __del__(self):

parent = getattr(self, "_parent", None)

if parent is not None:

return

gb_obj = getattr(self, "gb_obj", None)

if gb_obj is not None and lib is not None:

# it's difficult/dangerous to record the call, b/c `self.name` may not exist

check_status(lib.GrB_Vector_free(gb_obj), self)

def _as_matrix(self, *, name=None):

"""Cast this Vector to a Matrix (such as a column vector).

This is SuiteSparse-specific and may change in the future.

This does not copy the vector.

"""

from .matrix import Matrix

if backend == "suitesparse":

return Matrix._from_obj(

ffi.cast("GrB_Matrix*", self.gb_obj),

self.dtype,

self._size,

1,

parent=self,

name=f"(GrB_Matrix){self.name}" if name is None else name,

)

rv = Matrix(self.dtype, self._size, 1, name=self.name if name is None else name)

rv[:, 0] = self

return rv

def __repr__(self, mask=None, expr=None):

from .formatting import format_vector

from .recorder import skip_record

with skip_record:

return format_vector(self, mask=mask, expr=expr)

def _repr_html_(self, mask=None, collapse=False, expr=None):

if self._parent is not None and mask is None:

# Scalars repr can't handle mask

return self._parent._repr_html_(collapse=collapse)

from .formatting import format_vector_html

from .recorder import skip_record

with skip_record:

return format_vector_html(self, mask=mask, collapse=collapse, expr=expr)

@property

def _name_html(self):

if self._parent is not None:

return self._parent._name_html

return super()._name_html

def __reduce__(self):

# TODO: we should probably use (or compare to) GraphBLAS serialize methods

if backend == "suitesparse":

pieces = self.ss.export(raw=True)

else:

indices, values = self.to_coo(sort=False)

pieces = (indices, values, self.dtype, self._size)

return self._deserialize, (pieces, self.name)

@staticmethod

def _deserialize(pieces, name):

if backend == "suitesparse":

return Vector.ss.import_any(name=name, **pieces)

indices, values, dtype, size = pieces

return Vector.from_coo(indices, values, dtype, size=size, name=name)

@property

def S(self):

"""Create a Mask based on the structure of the Vector."""

return StructuralMask(self)

@property

def V(self):

"""Create a Mask based on the values in the Vector (treating each value as truthy)."""

return ValueMask(self)

def __delitem__(self, keys, **opts):

"""Delete a single element or subvector.

Examples

--------

>>> del v[1:-1]

"""

del Updater(self, opts=opts)[keys]

def __getitem__(self, keys):

"""Extract a single element or subvector.

See the `Extract section <../user_guide/operations.html#extract>`__

in the User Guide for more details.

Examples

--------

.. code-block:: python

sub_v = v[[1, 3, 5]].new()

"""

resolved_indexes = IndexerResolver(self, keys)

shape = resolved_indexes.shape

if not shape:

return ScalarIndexExpr(self, resolved_indexes)

return VectorIndexExpr(self, resolved_indexes, *shape)

def __setitem__(self, keys, expr, **opts):

"""Assign values to a single element or subvector.

See the `Assign section <../user_guide/operations.html#assign>`__

in the User Guide for more details.

Examples

--------

.. code-block:: python

# This makes a dense iso-value vector

v[:] = 1

"""

# Fast path for `v[i] = scalar`: a plain integer index and an exact-fit

# Python scalar, with no mask/accum/opts, a non-UDT dtype, and no active

# Recorder. This mirrors what Updater -> _assign_element does for a

# single element, but skips building the resolver, Updater, and Scalar.

# Only int/float/bool/complex values are taken here so the dtype

# inference and cffi coercion match _assign_element exactly, and only

# int and np.integer keys (bools excluded) so the accepted indices

# match parse_index; everything else (slices, fancy indexing, 0-d

# arrays and other __index__ objects, numpy or Scalar values,

# `v(mask)[i] << x`) falls back to the full assign path, leaving

# mask/accum, coercion, and index errors unchanged.

if (

not opts

and type(expr) in (int, float, bool, complex)

and (type(keys) is int or isinstance(keys, np.integer))

and not self.dtype._is_udt

and not _is_recording()

):

idx = keys.__index__()

size = self._size

if idx < 0:

idx += size

if idx < 0 or idx >= size:

raise IndexError(f"Index out of range: index={keys}, size={size}")

vdtype = lookup_dtype(type(expr), expr)

cvalue = ffi_new(f"{vdtype.c_type}*")

cvalue[0] = expr # cffi coercion, identical to the Scalar.value setter

err_code = utils.libget(f"GrB_Vector_setElement_{vdtype.name}")(

self.gb_obj[0], cvalue[0], idx

)

if err_code:

check_status_carg(err_code, "Vector", self.gb_obj[0])

return

Updater(self, opts=opts)[keys] = expr

def __contains__(self, index):

"""Indicates whether a value is present at the index.

Examples

--------

.. code-block:: python

# Check if v[15] is non-empty

15 in v

"""

# Fast path for a plain integer index: probe with

# GrB_Vector_extractElement directly instead of building an extract

# expression and Scalar. An out-of-range index falls through to the

# expression path so it raises the same IndexError as the slow path.

# Only int and np.integer take the fast lane (bools excluded), matching

# parse_index; other __index__ objects fall through to the expression

# path and its canonical errors. Fall back for a UDT dtype and an active

# Recorder (so the call is recorded), mirroring Vector.get.

if (

(type(index) is int or isinstance(index, np.integer))

and not self.dtype._is_udt

and not _is_recording()

):

idx = index.__index__()

size = self._size

if idx < 0:

idx += size

if 0 <= idx < size:

dtype = self.dtype

res = ffi_new(f"{dtype.c_type}*")

err_code = utils.libget(f"GrB_Vector_extractElement_{dtype.name}")(

res, self.gb_obj[0], idx

)

if err_code:

if err_code == GrB_NO_VALUE:

return False

check_status_carg(err_code, "Vector", self.gb_obj[0])

return True

extractor = self[index]

if not extractor._is_scalar:

raise TypeError(

f"Invalid index to Vector contains: {index!r}. An integer is expected. "

"Doing `index in my_vector` checks whether a value is present at that index."

)

scalar = extractor.new(name="s_contains")

return not scalar._is_empty

def __iter__(self):

"""Iterate over indices which are present in the vector."""

indices, _ = self.to_coo(values=False)

return indices.flat

def __sizeof__(self):

if backend == "suitesparse":

size = ffi_new("size_t*")

check_status(lib.GxB_Vector_memoryUsage(size, self.gb_obj[0]), self)

return size[0] + object.__sizeof__(self)

raise TypeError("Unable to get size of Vector with backend: {backend}")

def isequal(self, other, *, check_dtype=False, **opts):

"""Check for exact equality (same size, same structure).

Parameters

----------

other : Vector

The vector to compare against

check_dtypes : bool, default=False

If True, also checks that dtypes match

Returns

-------

bool

See Also

--------

:meth:`isclose` : For equality check of floating point dtypes

"""

other = self._expect_type(other, Vector, within="isequal", argname="other")

if check_dtype and self.dtype != other.dtype:

return False

if self._size != other._size:

return False

if self._nvals != other._nvals:

return False

if check_dtype:

# dtypes are equivalent, so not need to unify

op = binary.eq[self.dtype]

else:

op = get_typed_op(binary.eq, self.dtype, other.dtype, kind="binary")

matches = Vector(bool, self._size, name="v_isequal")

matches(**opts) << self.ewise_mult(other, op)

# ewise_mult performs intersection, so nvals will indicate mismatched empty values

if matches._nvals != self._nvals:

return False

# Check if all results are True

return matches.reduce(monoid.land, allow_empty=False).new(**opts).value

def isclose(self, other, *, rel_tol=1e-7, abs_tol=0.0, check_dtype=False, **opts):

"""Check for approximate equality (including same size and same structure).

Equivalent to: ``abs(a-b) <= max(rel_tol * max(abs(a), abs(b)), abs_tol)``.

Parameters

----------

other : Vector

Vector to compare against

rel_tol : float

Relative tolerance

abs_tol : float

Absolute tolerance

check_dtype : bool

If True, also checks that dtypes match

Returns

-------

bool

"""

other = self._expect_type(other, Vector, within="isclose", argname="other")

if check_dtype and self.dtype != other.dtype:

return False

if self._size != other._size:

return False

if self._nvals != other._nvals:

return False

if not _supports_udfs:

return _isclose_recipe(self, other, rel_tol, abs_tol, **opts)

matches = self.ewise_mult(other, binary.isclose(rel_tol, abs_tol)).new(

bool, name="M_isclose", **opts

)

# ewise_mult performs intersection, so nvals will indicate mismatched empty values

if matches._nvals != self._nvals:

return False

# Check if all results are True

return matches.reduce(monoid.land, allow_empty=False).new(**opts).value

@property

def size(self):

"""Size of the Vector."""

scalar = _scalar_index("s_size")

call("GrB_Vector_size", [_Pointer(scalar), self])

return scalar.gb_obj[0]

@property

def shape(self):

"""A tuple of ``(size,)``."""

return (self._size,)

@property

def nvals(self):

"""Number of non-empty values in the Vector."""

scalar = _scalar_index("s_nvals")

call("GrB_Vector_nvals", [_Pointer(scalar), self])

return scalar.gb_obj[0]

@property

def _nvals(self):

"""Like nvals, but doesn't record calls."""

n = ffi_new("GrB_Index*")

check_status(lib.GrB_Vector_nvals(n, self.gb_obj[0]), self)

return n[0]

def clear(self):

"""In-place operation which clears all values in the Vector.

After the call, :attr:`nvals` will return 0. The :attr:`size` will not change.

"""

call("GrB_Vector_clear", [self])

def resize(self, size):

"""In-place operation which changes the :attr:`size`.

| Increasing :attr:`size` will expand with empty values.

| Decreasing :attr:`size` will drop existing values above the new maximum index.

"""

size = _as_scalar(size, _INDEX, is_cscalar=True)

call("GrB_Vector_resize", [self, size])

self._size = size.value

def to_coo(self, dtype=None, *, indices=True, values=True, sort=True):

"""Extract the indices and values as a 2-tuple of numpy arrays.

Parameters

----------

dtype :

Requested dtype for the output values array.

indices :bool, default=True

Whether to return indices; will return ``None`` for indices if ``False``

values : bool, default=True

Whether to return values; will return ``None`` for values if ``False``

sort : bool, default=True

Whether to require sorted indices.

See Also

--------

to_dense

to_dict

from_coo

Returns

-------

np.ndarray[dtype=uint64] : Indices

np.ndarray : Values

"""

if sort and backend == "suitesparse":

self.wait() # sort in SS

nvals = self._nvals

if indices or backend != "suitesparse":

c_indices = _CArray(size=nvals, name="&index_array")

else:

c_indices = None

if values or backend != "suitesparse":

c_values = _CArray(size=nvals, dtype=self.dtype, name="&values_array")

else:

c_values = None

scalar = _scalar_index("s_nvals")

scalar.value = nvals

dtype_name = "UDT" if self.dtype._is_udt else self.dtype.name

call(

f"GrB_Vector_extractTuples_{dtype_name}", [c_indices, c_values, _Pointer(scalar), self]

)

if values:

c_values = normalize_values(self, c_values.array, dtype)

if sort and backend != "suitesparse":

c_indices = c_indices.array

ind = np.argsort(c_indices)

return (

c_indices[ind] if indices else None,

c_values[ind] if values else None,

)

return (

c_indices.array if indices else None,

c_values if values else None,

)

def build(self, indices, values, *, dup_op=None, clear=False, size=None):

"""Rarely used method to insert values into an existing Vector. The typical use case

is to create a new Vector and insert values at the same time using :meth:`from_coo`.

All the arguments are used identically in :meth:`from_coo`, except for ``clear``, which

indicates whether to clear the Vector prior to adding the new values.

"""

# TODO: accept `dtype` keyword to match the dtype of `values`?

indices = ints_to_numpy_buffer(indices, np.uint64, name="indices")

values, _dtype = values_to_numpy_buffer(values, self.dtype)

n = values.shape[0]

if indices.size != n:

raise ValueError(

f"`indices` and `values` lengths must match: {indices.size} != {values.size}"

)

if clear:

self.clear()

if size is not None:

self.resize(size)

if n == 0:

return

dup_op_given = dup_op is not None

if not dup_op_given:

if not self.dtype._is_udt:

dup_op = binary.plus

elif backend != "suitesparse":

dup_op = binary.any

# SS:SuiteSparse-specific: we use NULL for dup_op

if dup_op is not None:

dup_op = get_typed_op(dup_op, self.dtype, kind="binary")

if dup_op.opclass == "Monoid":

dup_op = dup_op.binaryop

else:

self._expect_op(dup_op, "BinaryOp", within="build", argname="dup_op")

indices = _CArray(indices)

values = _CArray(values, self.dtype)

dtype_name = "UDT" if self.dtype._is_udt else self.dtype.name

call(

f"GrB_Vector_build_{dtype_name}",

[self, indices, values, _as_scalar(n, _INDEX, is_cscalar=True), dup_op],

)

# Check for duplicates when dup_op was not provided

if not dup_op_given and self._nvals < n:

raise ValueError("Duplicate indices found, must provide `dup_op` BinaryOp")

def dup(self, dtype=None, *, clear=False, mask=None, name=None, **opts):

"""Create a duplicate of the Vector.

This is a full copy, not a view on the original.

Parameters

----------

dtype :

Data type of the new Vector. Normal typecasting rules apply.

clear : bool, default=False

If True, the returned Vector will be empty.

mask : Mask, optional

Mask controlling which elements of the original to include in the copy.

name : str, optional

Name to give the Vector.

Returns

-------

Vector

"""

if dtype is not None or mask is not None or clear:

if dtype is None:

dtype = self.dtype

rv = Vector(dtype, size=self._size, name=name)

if not clear:

rv(mask=mask, **opts)[...] = self

else:

if opts:

# Ignore opts for now

desc = descriptor_lookup(**opts) # noqa: F841 (keep desc in scope for context)

rv = Vector._from_obj(ffi_new("GrB_Vector*"), self.dtype, self._size, name=name)

call("GrB_Vector_dup", [_Pointer(rv), self])

return rv

def diag(self, k=0, *, name=None):

"""Return a Matrix with values on the diagonal built from the Vector.

Parameters

----------

k : int

Off-diagonal offset in the Matrix.

dtype :

Data type of the new Matrix. Normal typecasting rules apply.

name : str, optional

Name to give the new Matrix.

Returns

-------

:class:`~graphblas.Matrix`

"""

from .matrix import Matrix

k = _as_scalar(k, INT64, is_cscalar=True)

n = self._size + abs(k.value)

rv = Matrix._from_obj(ffi_new("GrB_Matrix*"), self.dtype, n, n, name=name)

call("GrB_Matrix_diag", [_Pointer(rv), self, k])

return rv

def wait(self, how="materialize"):

"""Wait for a computation to complete or establish a "happens-before" relation.

Parameters

----------

how : {"materialize", "complete"}

"materialize" fully computes an object.

"complete" establishes a "happens-before" relation useful with multi-threading.

See GraphBLAS documentation for more details.

In `non-blocking mode <../user_guide/init.html#graphblas-modes>`__,

the computations may be delayed and not yet safe to use by multiple threads.

Use wait to force completion of the Vector.

Has no effect in `blocking mode <../user_guide/init.html#graphblas-modes>`__.

"""

how = how.lower()

if how == "materialize":

mode = _MATERIALIZE

elif how == "complete":

mode = _COMPLETE

else:

raise ValueError(f'`how` argument must be "materialize" or "complete"; got {how!r}')

call("GrB_Vector_wait", [self, mode])

return self

def get(self, index, default=None):

"""Get an element at ``index`` as a Python scalar.

Parameters

----------

index : int

Vector index

default :

Value returned if no element exists at index

Returns

-------

Python scalar

"""

# Fast path for a plain integer index: call GrB_Vector_extractElement

# directly instead of building an extract expression, which costs ~10x

# more than the C call for single-element access. Fall back when a

# Recorder is active (so the call is recorded) and for UDTs (whose

# values need numpy-based conversion in Scalar.value). Only int and

# np.integer take the fast lane (bools excluded), matching parse_index;

# other __index__ objects fall through to the expression path and its

# canonical errors.

if (

(type(index) is int or isinstance(index, np.integer))

and not self.dtype._is_udt

and not _is_recording()

):

idx = index.__index__()

size = self._size

if idx < 0:

idx += size

if idx < 0 or idx >= size:

raise IndexError(f"Index out of range: index={index}, size={size}")

dtype = self.dtype

res = ffi_new(f"{dtype.c_type}*")

err_code = utils.libget(f"GrB_Vector_extractElement_{dtype.name}")(

res, self.gb_obj[0], idx

)

if err_code:

if err_code == GrB_NO_VALUE:

return default

check_status_carg(err_code, "Vector", self.gb_obj[0])

return res[0]

expr = self[index]

if expr._is_scalar:

rv = expr.new().value

return default if rv is None else rv

raise ValueError(

"Bad index in Vector.get(...). "

"A single index should be given, and the result will be a Python scalar."

)

@classmethod

def from_coo(cls, indices, values=1.0, dtype=None, *, size=None, dup_op=None, name=None):

"""Create a new Vector from indices and values.

.. warning::

When ``size`` is omitted, it is inferred from the largest index, so

trailing empty positions are dropped. Pass ``size`` explicitly to

pin the length.

Parameters

----------

indices : list or np.ndarray

Vector indices.

values : list or np.ndarray or scalar, default 1.0

List of values. If a scalar is provided, all values will be set to this single value.

dtype :

Data type of the Vector. If not provided, the values will be inspected

to choose an appropriate dtype.

size : int, optional

Size of the Vector. If not provided, ``size`` is computed from

the maximum index found in ``indices``.

dup_op : BinaryOp, optional

Function used to combine values if duplicate indices are found.

Leaving ``dup_op=None`` will raise an error if duplicates are found.

name : str, optional

Name to give the Vector.

See Also

--------

from_dense

from_dict

from_pairs

to_coo

Returns

-------

Vector

"""

indices = ints_to_numpy_buffer(indices, np.uint64, name="indices")

values, dtype = values_to_numpy_buffer(values, dtype, subarray_after=1)

# Compute size if not provided

if size is None:

if indices.size == 0:

raise ValueError("No indices provided. Unable to infer size.")

size = int(indices.max()) + 1

# Create the new vector

w = cls(dtype, size, name=name)

if values.ndim == 0:

if dup_op is not None:

raise ValueError(

"dup_op must be None if values is a scalar so that all "

"values can be identical. Duplicate indices will be ignored."

)

if backend == "suitesparse":

w.ss.build_scalar(indices, values.tolist())

else:

w.build(indices, np.broadcast_to(values, indices.size), dup_op=binary.any)

else:

# This needs to be the original data to get proper error messages

w.build(indices, values, dup_op=dup_op)

return w

@classmethod

def from_pairs(cls, pairs, dtype=None, *, size=None, dup_op=None, name=None):

"""Create a new Vector from indices and values.

This transforms the data and calls ``Vector.from_coo``.

Parameters

----------

pairs : list or iterable

A sequence of ``(index, value)`` pairs.

dtype :

Data type of the Vector. If not provided, the values will be inspected

to choose an appropriate dtype.

size : int, optional

Size of the Vector. If not provided, ``size`` is computed from

the maximum index found in ``pairs``.

dup_op : BinaryOp, optional

Function used to combine values if duplicate indices are found.

Leaving ``dup_op=None`` will raise an error if duplicates are found.

name : str, optional

Name to give the Vector.

See Also

--------

from_coo

from_dense

from_dict

to_coo

Returns

-------

Vector

"""

if isinstance(pairs, np.ndarray):

raise TypeError("pairs as NumPy array is not supported; use `Vector.from_coo` instead")

unzipped = list(zip(*pairs, strict=True))

if len(unzipped) == 2:

indices, values = unzipped

elif not unzipped:

# Empty pairs (size should be given)

indices = values = unzipped

else:

raise ValueError(

"Each item in the pairs must have two elements (for index and value); "

f"got {len(unzipped)}"

)

return cls.from_coo(indices, values, dtype, size=size, dup_op=dup_op, name=name)

@classmethod

def from_scalar(cls, value, size, dtype=None, *, name=None, **opts):

"""Create a fully dense Vector filled with a scalar value.

For SuiteSparse:GraphBLAS backend, this creates an iso-valued full Vector

that stores a single value regardless of the size of the Vector, so large

vectors created by ``Vector.from_scalar`` will use very low memory.

If instead you want to create a new iso-valued Vector with the same structure

as an existing Vector, you may do: ``w = binary.second(v, value).new()``.

Parameters

----------

value : scalar

Scalar value used to fill the Vector.

nrows : int

Number of rows.

ncols : int

Number of columns.

dtype : DataType, optional

Data type of the Vector. If not provided, the scalar value will be

inspected to choose an appropriate dtype.

name : str, optional

Name to give the Vector.

See Also

--------

from_coo

from_dense

from_dict

from_pairs

Returns

-------

Vector

"""

if type(value) is not Scalar:

try:

value = Scalar.from_value(value, dtype, is_cscalar=None, name="")

except TypeError:

value = cls()._expect_type(

value,

Scalar,

within="from_scalar",

keyword_name="value",

extra_message="Literal scalars also accepted.",

)

dtype = value.dtype

elif dtype is None:

dtype = value.dtype

else:

dtype = lookup_dtype(dtype)

if backend == "suitesparse" and not dtype._is_udt:

# `Vector.ss.import_full` does not yet handle all cases with UDTs

return cls.ss.import_full(value, dtype=dtype, size=size, is_iso=True, name=name)

rv = cls(dtype, size, name=name)

rv(**opts) << value

return rv

@classmethod

def from_dense(cls, values, missing_value=None, *, dtype=None, name=None, **opts):

"""Create a Vector from a NumPy array or list.

Parameters

----------

values : list or np.ndarray

List of values.

missing_value : scalar, optional

A scalar value to consider "missing"; elements of this value will be dropped.

If None, then the resulting Vector will be dense.

dtype : DataType, optional

Data type of the Vector. If not provided, the values will be inspected

to choose an appropriate dtype.

name : str, optional

Name to give the Vector.

See Also

--------

from_coo

from_dict

from_pairs

from_scalar

to_dense

Returns

-------

Vector

"""

values, dtype = values_to_numpy_buffer(values, dtype, subarray_after=1)

if values.ndim == 0:

raise TypeError(

"values must be an array or list, not a scalar. "

"To create a dense Vector from a scalar, use `Vector.from_scalar`."

)

if values.ndim == 1 and dtype.np_type.subdtype is not None:

raise ValueError("A >1d array is required to create a dense Vector with subdtype")

if values.ndim > 1 and dtype.np_type.subdtype is None:

raise ValueError(f"values array must be 1d to create dense Vector with dtype {dtype}")

if backend == "suitesparse":

rv = cls.ss.import_full(values, dtype=dtype, name=name)

else:

# TODO: GraphBLAS needs a better way to import or assign dense

rv = cls.from_coo(

np.arange(values.shape[0], dtype=np.uint64),

values,

dtype,

size=values.shape[0],

name=name,

)

if missing_value is not None:

rv(**opts) << select.valuene(rv, missing_value)

return rv

def to_dense(self, fill_value=None, dtype=None, **opts):

"""Convert Vector to NumPy array of the same shape with missing values filled.

.. warning::

This can create very large arrays that require a lot of memory; please use caution.

Parameters