585234361

Build Type

github

Committed by GitHub

Commit Message

Merge 2962a2fb2 into 3f04889fe

Pull Request Pull Request #67: Add `grblas.op` namespace to put all operations together.

Run Details

1290 of 1326 branches covered (97.29%)

Branch coverage included in aggregate %.

319 of 322 new or added lines in 20 files covered. (99.07%)

2 existing lines in 1 file now uncovered.

6400 of 6453 relevant lines covered (99.18%)

5.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.69

/grblas/base.py

from contextvars import ContextVar
from . import ffi
from .descriptor import lookup as descriptor_lookup
from .dtypes import lookup_dtype
from .exceptions import check_status
from .expr import AmbiguousAssignOrExtract, Updater
from .mask import Mask
from .operator import UNKNOWN_OPCLASS, find_opclass, get_typed_op
from .unary import identity
from .utils import libget, _Pointer

NULL = ffi.NULL
CData = ffi.CData
_recorder = ContextVar("recorder")
_prev_recorder = None


def call(cfunc_name, args):
    call_args = [getattr(x, "_carg", x) if x is not None else NULL for x in args]
    cfunc = libget(cfunc_name)
    try:
        err_code = cfunc(*call_args)
    except TypeError as exc:
        # We should strive to not encounter this during normal usage
        from .recorder import gbstr

        callstr = f'{cfunc.__name__}({", ".join(gbstr(x) for x in args)})'
        lines = cfunc.__doc__.splitlines()
        sig = lines[0] if lines else ""
        raise TypeError(
            f"Error calling {cfunc.__name__}:\n"
            f" - Call objects: {callstr}\n"
            f" - C signature: {sig}\n"
            f" - Error: {exc}"
        )
    rv = check_status(err_code, args)
    rec = _recorder.get(_prev_recorder)
    if rec is not None:
        rec.record(cfunc_name, args)
    return rv


def _expect_type_message(
    self, x, types, *, within, argname=None, keyword_name=None, extra_message=""
):
    if type(types) is tuple:
        if type(x) in types:
            return
    elif type(x) is types:
        return
    if argname:
        argmsg = f"for argument `{argname}` "
    elif keyword_name:
        argmsg = f"for keyword argument `{keyword_name}=` "
    else:
        argmsg = ""
    if type(types) is tuple:
        expected = ", ".join(typ.__name__ for typ in types)
    else:
        expected = types.__name__
    if extra_message:
        extra_message = f"\n{extra_message}"
    return (
        f"Bad type {argmsg}in {type(self).__name__}.{within}(...).\n"
        f"    - Expected type: {expected}.\n"
        f"    - Got: {type(x)}."
        f"{extra_message}"
    )


def _expect_type(self, x, types, **kwargs):
    message = _expect_type_message(self, x, types, **kwargs)
    if message is not None:
        raise TypeError(message) from None


def _expect_op_message(
    self, op, values, *, within, argname=None, keyword_name=None, extra_message=""
):
    if type(values) is tuple:
        if op.opclass in values:
            return
    elif op.opclass == values:
        return
    if argname:
        argmsg = f"for argument `{argname}` "
    elif keyword_name:
        argmsg = f"for keyword argument `{keyword_name}=` "
    else:  # pragma: no cover
        argmsg = ""
    if type(values) is tuple:
        expected = ", ".join(values)
    else:
        expected = values
    if extra_message:
        extra_message = f"\n{extra_message}"
    return (
        f"Bad type {argmsg}in {type(self).__name__}.{within}(...).\n"
        f"    - Expected type: {expected}.\n"
        f"    - Got: {op.opclass}."
        f"{extra_message}"
    )


def _expect_op(self, op, values, **kwargs):
    message = _expect_op_message(self, op, values, **kwargs)
    if message is not None:
        raise TypeError(message) from None


def _check_mask(mask, output=None):
    if isinstance(mask, BaseType) or type(mask) is Mask:
        raise TypeError("Mask must indicate values (M.V) or structure (M.S)")
    if not isinstance(mask, Mask):
        raise TypeError(f"Invalid mask: {type(mask)}")
    if output is not None:
        from .vector import Vector

        if type(output) is Vector and type(mask.mask) is not Vector:
            raise TypeError(f"Mask object must be type Vector; got {type(mask.mask)}")


class BaseType:
    # Flag for operations which depend on scalar vs vector/matrix
    _is_scalar = False

    def __init__(self, gb_obj, dtype, name):
        if not isinstance(gb_obj, CData):
            raise TypeError("Object passed to __init__ must be CData type")
        self.gb_obj = gb_obj
        self.dtype = lookup_dtype(dtype)
        self.name = name

    def __call__(
        self, *optional_mask_and_accum, mask=None, accum=None, replace=False, input_mask=None
    ):
        # Pick out mask and accum from positional arguments
        mask_arg = None
        accum_arg = None
        for arg in optional_mask_and_accum:
            if isinstance(arg, (BaseType, Mask)):
                if self._is_scalar:
                    raise TypeError("Mask not allowed for Scalars")
                if mask_arg is not None:
                    raise TypeError("Got multiple values for argument 'mask'")
                mask_arg = arg
            else:
                if accum_arg is not None:
                    raise TypeError("Got multiple values for argument 'accum'")
                accum_arg, opclass = find_opclass(arg)
                if opclass == UNKNOWN_OPCLASS:
                    raise TypeError(f"Invalid item found in output params: {type(arg)}")
                elif opclass == "Monoid":
                    accum_arg = accum_arg.binaryop
                elif opclass != "BinaryOp":
                    raise TypeError(f"accum must be a BinaryOp, not {opclass}")
        # Merge positional and keyword arguments
        if mask_arg is not None and mask is not None:
            raise TypeError("Got multiple values for argument 'mask'")
        if mask_arg is not None:
            mask = mask_arg
        if mask is None:
            if input_mask is None:
                pass
            elif self._is_scalar:
                raise TypeError("input_mask not allowed for Scalars")
            else:
                _check_mask(input_mask)
        elif self._is_scalar:
            raise TypeError("Mask not allowed for Scalars")
        elif input_mask is not None:
            raise TypeError("mask and input_mask arguments cannot both be given")
        else:
            _check_mask(mask)
        if accum_arg is not None and accum is not None:
            raise TypeError("Got multiple values for argument 'accum'")
        if accum_arg is not None:
            accum = accum_arg
        return Updater(self, mask=mask, accum=accum, replace=replace, input_mask=input_mask)

    def __eq__(self, other):
        raise TypeError(
            f"__eq__ not defined for objects of type {type(self)}.  Use `.isequal` method instead."
        )

    def __bool__(self):
        raise TypeError(
            f"__bool__ not defined for objects of type {type(self)}.  "
            "Perhaps use .nvals attribute instead."
        )

    def __lshift__(self, delayed):
        return self._update(delayed)

    def update(self, delayed):
        """
        Convenience function when no output arguments (mask, accum, replace) are used
        """
        return self._update(delayed)

    def _update(self, delayed, mask=None, accum=None, replace=False, input_mask=None):
        # TODO: check expected output type (now included in Expression object)
        if not isinstance(delayed, BaseExpression):
            if type(delayed) is AmbiguousAssignOrExtract:
                if delayed.resolved_indexes.is_single_element and self._is_scalar:
                    # Extract element (s << v[1])
                    if accum is not None:
                        raise TypeError(
                            "Scalar accumulation with extract element"
                            "--such as `s(accum=accum) << v[0]`--is not supported"
                        )
                    self.value = delayed.new(dtype=self.dtype, name="s_extract").value
                    return

                # Extract (C << A[rows, cols])
                if input_mask is not None:
                    if mask is not None:
                        raise TypeError("mask and input_mask arguments cannot both be given")
                    _check_mask(input_mask, output=delayed.parent)
                    mask = delayed._input_mask_to_mask(input_mask)
                    input_mask = None
                delayed = delayed._extract_delayed()
            elif type(delayed) is type(self):
                # Simple assignment (w << v)
                if self._is_scalar:
                    if accum is not None:
                        raise TypeError(
                            "Scalar update with accumulation--such as `s(accum=accum) << t`"
                            "--is not supported"
                        )
                    self.value = delayed.value
                    return

                delayed = delayed.apply(identity)
            elif self._is_scalar:
                if accum is not None:
                    raise TypeError(
                        "Scalar update with accumulation--such as `s(accum=accum) << t`"
                        "--is not supported"
                    )
                self.value = delayed
                return

            else:
                from .matrix import Matrix, TransposedMatrix, MatrixExpression

                if type(delayed) is TransposedMatrix and type(self) is Matrix:
                    # Transpose (C << A.T)
                    delayed = MatrixExpression(
                        "transpose",
                        "GrB_transpose",
                        [delayed],
                        expr_repr="{0}",
                        dtype=delayed.dtype,
                        nrows=delayed._nrows,
                        ncols=delayed._ncols,
                    )
                else:
                    from .scalar import Scalar

                    if type(delayed) is Scalar:
                        scalar = delayed
                    else:
                        try:
                            scalar = Scalar.from_value(delayed, name=repr(delayed))
                        except TypeError:
                            raise TypeError(
                                "Assignment value must be a valid expression type, not "
                                f"{type(delayed)}.\n\nValid expression types include "
                                f"{type(self).__name__}, {type(self).__name__}Expression, "
                                "AmbiguousAssignOrExtract, and scalars."
                            )
                    updater = self(mask=mask, accum=accum, replace=replace, input_mask=input_mask)
                    if type(self) is Matrix:
                        if mask is None:
                            raise TypeError(
                                "Warning: updating a Matrix with a scalar without a mask will "
                                "make the Matrix dense.  This may use a lot of memory and probably "
                                "isn't what you want.  Perhaps you meant:"
                                "\n\n    M(M.S) << s\n\n"
                                "If you do wish to make a dense matrix, then please be explicit:"
                                "\n\n    M[:, :] = s"
                            )
                        updater[:, :] = scalar
                    else:  # Vector
                        updater[:] = scalar
                    return

        if input_mask is not None:
            raise TypeError("`input_mask` argument may only be used for extract")
        # Normalize mask and separate out complement and structural flags
        if mask is None:
            complement = False
            structure = False
        else:
            _check_mask(mask, self)
            complement = mask.complement
            structure = mask.structure

        # Normalize accumulator
        if accum is not None:
            accum = get_typed_op(accum, self.dtype)
            if accum.opclass == "Monoid":
                accum = accum.binaryop
            else:
                self._expect_op(accum, "BinaryOp", within="update", keyword_name="accum")

        # Get descriptor based on flags
        desc = descriptor_lookup(
            transpose_first=delayed.at,
            transpose_second=delayed.bt,
            mask_complement=complement,
            mask_structure=structure,
            output_replace=replace,
        )
        if self._is_scalar:
            args = [_Pointer(self), accum]
            cfunc_name = delayed.cfunc_name.format(output_dtype=self.dtype)
        else:
            args = [self, mask, accum]
            cfunc_name = delayed.cfunc_name
        if delayed.op is not None:
            args.append(delayed.op)
        args.extend(delayed.args)
        args.append(desc)
        # Make the GraphBLAS call
        call(cfunc_name, args)
        if self._is_scalar:
            self._is_empty = False

    @property
    def _name_html(self):
        """Treat characters after _ as subscript"""
        split = self.name.split("_", 1)
        if len(split) == 1:
            return self.name
        return f"{split[0]}<sub>{split[1]}</sub>"

    _expect_type = _expect_type
    _expect_op = _expect_op


class BaseExpression:
    output_type = None

    def __init__(
        self,
        method_name,
        cfunc_name,
        args,
        *,
        at=False,
        bt=False,
        op=None,
        dtype=None,
        expr_repr=None,
    ):
        self.method_name = method_name
        self.cfunc_name = cfunc_name
        self.args = args
        self.at = at
        self.bt = bt
        self.op = op
        if expr_repr is None:
            if len(args) == 1:
                expr_repr = "{0.name}.{method_name}({op})"
            elif len(args) == 2:
                expr_repr = "{0.name}.{method_name}({1.name}, op={op})"
            else:  # pragma: no cover
                raise ValueError(f"No default expr_repr for len(args) == {len(args)}")
        self.expr_repr = expr_repr
        if dtype is None:
            self.dtype = op.return_type
        else:
            self.dtype = dtype

    def new(self, *, dtype=None, mask=None, name=None):
        output = self.construct_output(dtype=dtype, name=name)
        if mask is None:
            output.update(self)
        else:
            _check_mask(mask, output)
            output(mask=mask).update(self)
        return output

    def __eq__(self, other):
        raise TypeError(
            f"__eq__ not defined for objects of type {type(self)}.  "
            f"Use `.new()` to create a new {self.output_type.__name__}, then use `.isequal` method."
        )

    def __bool__(self):
        raise TypeError(f"__bool__ not defined for objects of type {type(self)}.")

    def _format_expr(self):
        return self.expr_repr.format(*self.args, method_name=self.method_name, op=self.op)

    def _format_expr_html(self):
        expr_repr = self.expr_repr.replace(".name", "._name_html")
        return expr_repr.format(*self.args, method_name=self.method_name, op=self.op)

1	from contextvars import ContextVar	6✔
2	from . import ffi	6✔
3	from .descriptor import lookup as descriptor_lookup	6✔
4	from .dtypes import lookup_dtype	6✔
5	from .exceptions import check_status	6✔
6	from .expr import AmbiguousAssignOrExtract, Updater	6✔
7	from .mask import Mask	6✔
8	from .operator import UNKNOWN_OPCLASS, find_opclass, get_typed_op	6✔
9	from .unary import identity	6✔
10	from .utils import libget, _Pointer	6✔
11
12	NULL = ffi.NULL	6✔
13	CData = ffi.CData	6✔
14	_recorder = ContextVar("recorder")	6✔
15	_prev_recorder = None	6✔
16
17
18	def call(cfunc_name, args):	6✔
19	call_args = [getattr(x, "_carg", x) if x is not None else NULL for x in args]	6✔
20	cfunc = libget(cfunc_name)	6✔
21	try:	6✔
22	err_code = cfunc(*call_args)	6✔
23	except TypeError as exc:	6✔
24	# We should strive to not encounter this during normal usage
25	from .recorder import gbstr	6✔
26
27	callstr = f'{cfunc.__name__}({", ".join(gbstr(x) for x in args)})'	6✔
28	lines = cfunc.__doc__.splitlines()	6✔
29	sig = lines[0] if lines else ""	6✔
30	raise TypeError(	6✔
31	f"Error calling {cfunc.__name__}:\n"
32	f" - Call objects: {callstr}\n"
33	f" - C signature: {sig}\n"
34	f" - Error: {exc}"
35	)
36	rv = check_status(err_code, args)	6✔
37	rec = _recorder.get(_prev_recorder)	6✔
38	if rec is not None:	6✔
39	rec.record(cfunc_name, args)	6✔
40	return rv	6✔
41
42
43	def _expect_type_message(	6✔
44	self, x, types, *, within, argname=None, keyword_name=None, extra_message=""
45	):
46	if type(types) is tuple:	6✔
47	if type(x) in types:	6✔
48	return	6✔
49	elif type(x) is types:	6✔
50	return	6✔
51	if argname:	6✔
52	argmsg = f"for argument `{argname}` "	6✔
53	elif keyword_name:	6✔
54	argmsg = f"for keyword argument `{keyword_name}=` "	6✔
55	else:
56	argmsg = ""	6✔
57	if type(types) is tuple:	6✔
58	expected = ", ".join(typ.__name__ for typ in types)	6✔
59	else:
60	expected = types.__name__	6✔
61	if extra_message:	6✔
62	extra_message = f"\n{extra_message}"	6✔
63	return (	6✔
64	f"Bad type {argmsg}in {type(self).__name__}.{within}(...).\n"
65	f" - Expected type: {expected}.\n"
66	f" - Got: {type(x)}."
67	f"{extra_message}"
68	)
69
70
71	def _expect_type(self, x, types, **kwargs):	6✔
72	message = _expect_type_message(self, x, types, **kwargs)	6✔
73	if message is not None:	6✔
74	raise TypeError(message) from None	6✔
75
76
77	def _expect_op_message(	6✔
78	self, op, values, *, within, argname=None, keyword_name=None, extra_message=""
79	):
80	if type(values) is tuple:	6✔
81	if op.opclass in values:	6✔
82	return	6✔
83	elif op.opclass == values:	6✔
84	return	6✔
85	if argname:	6!
86	argmsg = f"for argument `{argname}` "	6✔
UNCOV 87	elif keyword_name:	×
UNCOV 88	argmsg = f"for keyword argument `{keyword_name}=` "	×
89	else: # pragma: no cover
90	argmsg = ""
91	if type(values) is tuple:	6✔
92	expected = ", ".join(values)	6✔
93	else:
94	expected = values	6✔
95	if extra_message:	6✔
96	extra_message = f"\n{extra_message}"	6✔
97	return (	6✔
98	f"Bad type {argmsg}in {type(self).__name__}.{within}(...).\n"
99	f" - Expected type: {expected}.\n"
100	f" - Got: {op.opclass}."
101	f"{extra_message}"
102	)
103
104
105	def _expect_op(self, op, values, **kwargs):	6✔
106	message = _expect_op_message(self, op, values, **kwargs)	6✔
107	if message is not None:	6✔
108	raise TypeError(message) from None	6✔
109
110
111	def _check_mask(mask, output=None):	6✔
112	if isinstance(mask, BaseType) or type(mask) is Mask:	6✔
113	raise TypeError("Mask must indicate values (M.V) or structure (M.S)")	6✔
114	if not isinstance(mask, Mask):	6✔
115	raise TypeError(f"Invalid mask: {type(mask)}")	6✔
116	if output is not None:	6✔
117	from .vector import Vector	6✔
118
119	if type(output) is Vector and type(mask.mask) is not Vector:	6✔
120	raise TypeError(f"Mask object must be type Vector; got {type(mask.mask)}")	6✔
121
122
123	class BaseType:	6✔
124	# Flag for operations which depend on scalar vs vector/matrix
125	_is_scalar = False	6✔
126
127	def __init__(self, gb_obj, dtype, name):	6✔
128	if not isinstance(gb_obj, CData):	6✔
129	raise TypeError("Object passed to __init__ must be CData type")	6✔
130	self.gb_obj = gb_obj	6✔
131	self.dtype = lookup_dtype(dtype)	6✔
132	self.name = name	6✔
133
134	def __call__(	6✔
135	self, *optional_mask_and_accum, mask=None, accum=None, replace=False, input_mask=None
136	):
137	# Pick out mask and accum from positional arguments
138	mask_arg = None	6✔
139	accum_arg = None	6✔
140	for arg in optional_mask_and_accum:	6✔
141	if isinstance(arg, (BaseType, Mask)):	6✔
142	if self._is_scalar:	6✔
143	raise TypeError("Mask not allowed for Scalars")	6✔
144	if mask_arg is not None:	6✔
145	raise TypeError("Got multiple values for argument 'mask'")	6✔
146	mask_arg = arg	6✔
147	else:
148	if accum_arg is not None:	6✔
149	raise TypeError("Got multiple values for argument 'accum'")	6✔
150	accum_arg, opclass = find_opclass(arg)	6✔
151	if opclass == UNKNOWN_OPCLASS:	6✔
152	raise TypeError(f"Invalid item found in output params: {type(arg)}")	6✔
153	elif opclass == "Monoid":	6✔
154	accum_arg = accum_arg.binaryop	6✔
155	elif opclass != "BinaryOp":	6✔
156	raise TypeError(f"accum must be a BinaryOp, not {opclass}")	6✔
157	# Merge positional and keyword arguments
158	if mask_arg is not None and mask is not None:	6✔
159	raise TypeError("Got multiple values for argument 'mask'")	6✔
160	if mask_arg is not None:	6✔
161	mask = mask_arg	6✔
162	if mask is None:	6✔
163	if input_mask is None:	6✔
164	pass	6✔
165	elif self._is_scalar:	6✔
166	raise TypeError("input_mask not allowed for Scalars")	6✔
167	else:
168	_check_mask(input_mask)	6✔
169	elif self._is_scalar:	6✔
170	raise TypeError("Mask not allowed for Scalars")	6✔
171	elif input_mask is not None:	6✔
172	raise TypeError("mask and input_mask arguments cannot both be given")	6✔
173	else:
174	_check_mask(mask)	6✔
175	if accum_arg is not None and accum is not None:	6✔
176	raise TypeError("Got multiple values for argument 'accum'")	6✔
177	if accum_arg is not None:	6✔
178	accum = accum_arg	6✔
179	return Updater(self, mask=mask, accum=accum, replace=replace, input_mask=input_mask)	6✔
180
181	def __eq__(self, other):	6✔
182	raise TypeError(	6✔
183	f"__eq__ not defined for objects of type {type(self)}. Use `.isequal` method instead."
184	)
185
186	def __bool__(self):	6✔
187	raise TypeError(	6✔
188	f"__bool__ not defined for objects of type {type(self)}. "
189	"Perhaps use .nvals attribute instead."
190	)
191
192	def __lshift__(self, delayed):	6✔
193	return self._update(delayed)	6✔
194
195	def update(self, delayed):	6✔
196	"""
197	Convenience function when no output arguments (mask, accum, replace) are used
198	"""
199	return self._update(delayed)	6✔
200
201	def _update(self, delayed, mask=None, accum=None, replace=False, input_mask=None):	6✔
202	# TODO: check expected output type (now included in Expression object)
203	if not isinstance(delayed, BaseExpression):	6✔
204	if type(delayed) is AmbiguousAssignOrExtract:	6✔
205	if delayed.resolved_indexes.is_single_element and self._is_scalar:	6✔
206	# Extract element (s << v[1])
207	if accum is not None:	6✔
208	raise TypeError(	6✔
209	"Scalar accumulation with extract element"
210	"--such as `s(accum=accum) << v[0]`--is not supported"
211	)
212	self.value = delayed.new(dtype=self.dtype, name="s_extract").value	6✔
213	return	6✔
214
215	# Extract (C << A[rows, cols])
216	if input_mask is not None:	6✔
217	if mask is not None:	6!
218	raise TypeError("mask and input_mask arguments cannot both be given")	×
219	_check_mask(input_mask, output=delayed.parent)	6✔
220	mask = delayed._input_mask_to_mask(input_mask)	6✔
221	input_mask = None	6✔
222	delayed = delayed._extract_delayed()	6✔
223	elif type(delayed) is type(self):	6✔
224	# Simple assignment (w << v)
225	if self._is_scalar:	6✔
226	if accum is not None:	6✔
227	raise TypeError(	6✔
228	"Scalar update with accumulation--such as `s(accum=accum) << t`"
229	"--is not supported"
230	)
231	self.value = delayed.value	6✔
232	return	6✔
233
234	delayed = delayed.apply(identity)	6✔
235	elif self._is_scalar:	6✔
236	if accum is not None:	6✔
237	raise TypeError(	6✔
238	"Scalar update with accumulation--such as `s(accum=accum) << t`"
239	"--is not supported"
240	)
241	self.value = delayed	6✔
242	return	6✔
243
244	else:
245	from .matrix import Matrix, TransposedMatrix, MatrixExpression	6✔
246
247	if type(delayed) is TransposedMatrix and type(self) is Matrix:	6✔
248	# Transpose (C << A.T)
249	delayed = MatrixExpression(	6✔
250	"transpose",
251	"GrB_transpose",
252	[delayed],
253	expr_repr="{0}",
254	dtype=delayed.dtype,
255	nrows=delayed._nrows,
256	ncols=delayed._ncols,
257	)
258	else:
259	from .scalar import Scalar	6✔
260
261	if type(delayed) is Scalar:	6✔
262	scalar = delayed	6✔
263	else:
264	try:	6✔
265	scalar = Scalar.from_value(delayed, name=repr(delayed))	6✔
266	except TypeError:	6✔
267	raise TypeError(	6✔
268	"Assignment value must be a valid expression type, not "
269	f"{type(delayed)}.\n\nValid expression types include "
270	f"{type(self).__name__}, {type(self).__name__}Expression, "
271	"AmbiguousAssignOrExtract, and scalars."
272	)
273	updater = self(mask=mask, accum=accum, replace=replace, input_mask=input_mask)	6✔
274	if type(self) is Matrix:	6✔
275	if mask is None:	6✔
276	raise TypeError(	6✔
277	"Warning: updating a Matrix with a scalar without a mask will "
278	"make the Matrix dense. This may use a lot of memory and probably "
279	"isn't what you want. Perhaps you meant:"
280	"\n\n M(M.S) << s\n\n"
281	"If you do wish to make a dense matrix, then please be explicit:"
282	"\n\n M[:, :] = s"
283	)
284	updater[:, :] = scalar	6✔
285	else: # Vector
286	updater[:] = scalar	6✔
287	return	6✔
288
289	if input_mask is not None:	6✔
290	raise TypeError("`input_mask` argument may only be used for extract")	6✔
291	# Normalize mask and separate out complement and structural flags
292	if mask is None:	6✔
293	complement = False	6✔
294	structure = False	6✔
295	else:
296	_check_mask(mask, self)	6✔
297	complement = mask.complement	6✔
298	structure = mask.structure	6✔
299
300	# Normalize accumulator
301	if accum is not None:	6✔
302	accum = get_typed_op(accum, self.dtype)	6✔
303	if accum.opclass == "Monoid":	6✔
304	accum = accum.binaryop	6✔
305	else:
306	self._expect_op(accum, "BinaryOp", within="update", keyword_name="accum")	6✔
307
308	# Get descriptor based on flags
309	desc = descriptor_lookup(	6✔
310	transpose_first=delayed.at,
311	transpose_second=delayed.bt,
312	mask_complement=complement,
313	mask_structure=structure,
314	output_replace=replace,
315	)
316	if self._is_scalar:	6✔
317	args = [_Pointer(self), accum]	6✔
318	cfunc_name = delayed.cfunc_name.format(output_dtype=self.dtype)	6✔
319	else:
320	args = [self, mask, accum]	6✔
321	cfunc_name = delayed.cfunc_name	6✔
322	if delayed.op is not None:	6✔
323	args.append(delayed.op)	6✔
324	args.extend(delayed.args)	6✔
325	args.append(desc)	6✔
326	# Make the GraphBLAS call
327	call(cfunc_name, args)	6✔
328	if self._is_scalar:	6✔
329	self._is_empty = False	6✔
330
331	@property	6✔
332	def _name_html(self):	3✔
333	"""Treat characters after _ as subscript"""
334	split = self.name.split("_", 1)	6✔
335	if len(split) == 1:	6✔
336	return self.name	6✔
337	return f"{split[0]}<sub>{split[1]}</sub>"	6✔
338
339	_expect_type = _expect_type	6✔
340	_expect_op = _expect_op	6✔
341
342
343	class BaseExpression:	6✔
344	output_type = None	6✔
345
346	def __init__(	6✔
347	self,
348	method_name,
349	cfunc_name,
350	args,
351	*,
352	at=False,
353	bt=False,
354	op=None,
355	dtype=None,
356	expr_repr=None,
357	):
358	self.method_name = method_name	6✔
359	self.cfunc_name = cfunc_name	6✔
360	self.args = args	6✔
361	self.at = at	6✔
362	self.bt = bt	6✔
363	self.op = op	6✔
364	if expr_repr is None:	6✔
365	if len(args) == 1:	6✔
366	expr_repr = "{0.name}.{method_name}({op})"	6✔
367	elif len(args) == 2:	6✔
368	expr_repr = "{0.name}.{method_name}({1.name}, op={op})"	6✔
369	else: # pragma: no cover
370	raise ValueError(f"No default expr_repr for len(args) == {len(args)}")
371	self.expr_repr = expr_repr	6✔
372	if dtype is None:	6✔
373	self.dtype = op.return_type	6✔
374	else:
375	self.dtype = dtype	6✔
376
377	def new(self, *, dtype=None, mask=None, name=None):	6✔
378	output = self.construct_output(dtype=dtype, name=name)	6✔
379	if mask is None:	6✔
380	output.update(self)	6✔
381	else:
382	_check_mask(mask, output)	6✔
383	output(mask=mask).update(self)	6✔
384	return output	6✔
385
386	def __eq__(self, other):	6✔
387	raise TypeError(	6✔
388	f"__eq__ not defined for objects of type {type(self)}. "
389	f"Use `.new()` to create a new {self.output_type.__name__}, then use `.isequal` method."
390	)
391
392	def __bool__(self):	6✔
393	raise TypeError(f"__bool__ not defined for objects of type {type(self)}.")	6✔
394
395	def _format_expr(self):	6✔
396	return self.expr_repr.format(*self.args, method_name=self.method_name, op=self.op)	6✔
397
398	def _format_expr_html(self):	6✔
399	expr_repr = self.expr_repr.replace(".name", "._name_html")	6✔
400	return expr_repr.format(*self.args, method_name=self.method_name, op=self.op)	6✔

metagraph-dev / grblas / 585234361

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous