metagraph-dev / grblas / 2110321838

import numpy as np

from . import config, monoid, unary

from .dtypes import BOOL

from .exceptions import OutOfMemory

from .matrix import Matrix, TransposedMatrix

from .vector import Vector

try:

    import pandas as pd

    has_pandas = True

CSS_STYLE = """

def _update_matrix_dataframe(df, matrix, rows, row_offset, columns, column_offset, *, mask=None):

    if rows is None and columns is None:

        if mask is None:

            submatrix = matrix

            submatrix = Matrix.new("UINT8", matrix._nrows, matrix._ncols, name="")

            if mask.structure:

                submatrix(matrix.S)[...] = 0 if mask.complement else 1

                submatrix(matrix.S)[...] = 1 if mask.complement else 0

                submatrix(matrix.V)[...] = 0 if mask.complement else 1

        if rows is None:

            rows = slice(None)

        if columns is None:

            columns = slice(None)

        if type(matrix) is TransposedMatrix:

            parent = matrix._matrix

            submatrix = Matrix.new(parent.dtype, parent._nrows, parent._ncols, name="")

                val = parent.reduce_scalar(monoid.any, allow_empty=False).new(name="")

                submatrix(parent.S)[columns, rows] = val

                submatrix(submatrix.S)[...] = parent

            if row_offset is not None or column_offset is not None:

                submatrix = submatrix[column_offset:, row_offset:].new(name="")

            submatrix = submatrix.T

            if mask is None:

                submatrix = Matrix.new(matrix.dtype, matrix._nrows, matrix._ncols, name="")

                if matrix._nvals > 0:

                    val = matrix.reduce_scalar(monoid.any).new(name="")

                    submatrix(matrix.S)[rows, columns] = val

                    submatrix(submatrix.S)[...] = matrix

                submatrix = Matrix.new("UINT8", matrix._nrows, matrix._ncols, name="")

                if mask.structure:

                    submatrix(matrix.S)[rows, columns] = 0 if mask.complement else 1

                    submatrix(matrix.S)[rows, columns] = 1 if mask.complement else 0

                    submatrix(matrix.V)[rows, columns] = 0 if mask.complement else 1

            if row_offset is not None or column_offset is not None:

                submatrix = submatrix[row_offset:, column_offset:].new(name="")

    rows, cols, vals = submatrix.to_values()

    np_type = submatrix.dtype.np_type

    df.values[rows, cols] = vals

    if np.issubdtype(np_type, np.inexact):

        nulls = np.isnan(vals)

        df.values[rows[nulls], cols[nulls]] = "nan"

def _update_vector_dataframe(df, vector, columns, column_offset, *, mask=None):

    if columns is None:

        if mask is None:

            subvector = vector

            subvector = Vector.new("UINT8", vector._size, name="")

            if mask.structure:

                subvector(vector.S)[...] = 0 if mask.complement else 1

                subvector(vector.S)[...] = 1 if mask.complement else 0

                subvector(vector.V)[...] = 0 if mask.complement else 1

        if mask is None:

            subvector = Vector.new(vector.dtype, vector._size, name="")

            if vector._nvals > 0:

                val = vector.reduce(monoid.any).new(name="")

                subvector(vector.S)[columns] = val

                subvector(subvector.S)[...] = vector

            subvector = Vector.new("UINT8", vector._size, name="")

            if mask.structure:

                subvector(vector.S)[columns] = 0 if mask.complement else 1

                subvector(vector.S)[columns] = 1 if mask.complement else 0

                subvector(vector.V)[columns] = 0 if mask.complement else 1

        if column_offset is not None:

            subvector = subvector[column_offset:].new(name="")

    cols, vals = subvector.to_values()

    np_type = subvector.dtype.np_type

    df.values[0, cols] = vals

    if np.issubdtype(np_type, np.inexact):

        df.values[0, cols[np.isnan(vals)]] = "nan"

def _get_max_columns():

    max_columns = pd.options.display.max_columns

    if max_columns == 0:

        max_columns = 150

    return max_columns

def _get_chunk(length, min_length, max_length):

    if length <= max_length:

        chunk = list(range(length))

        chunk_groups = [(None, None)]

        half = min_length // 2

        first_chunk = list(range(half))

        second_chunk = list(range(length - half, length))

        chunk = list(range(max_length + 1))

        chunk[-half:] = second_chunk

        chunk_groups = [(first_chunk, None), (second_chunk, length - len(chunk))]

    return chunk, chunk_groups

def _get_matrix_dataframe(matrix, max_rows, min_rows, max_columns, *, mask=None):

    if not has_pandas:

        return

    if max_rows is None:  # pragma: no branch

        max_rows = pd.options.display.max_rows

    if min_rows is None:  # pragma: no branch

        min_rows = pd.options.display.min_rows

    if max_columns is None:  # pragma: no branch

        max_columns = _get_max_columns()

    rows, row_groups = _get_chunk(matrix._nrows, min_rows, max_rows)

    columns, column_groups = _get_chunk(matrix._ncols, max_columns, max_columns)

    df = pd.DataFrame(columns=columns, index=rows)

    for row_group, row_offset in row_groups:

        for column_group, column_offset in column_groups:

            _update_matrix_dataframe(

    if (

        num_rows = matrix._nvals if matrix._nvals <= max_rows else min_rows

        if matrix._is_transposed:

            cols, rows, vals = matrix._matrix.ss.head(num_rows, sort=True)

            rows, cols, vals = matrix.ss.head(num_rows, sort=True)

        if mask is not None:

            if mask.complement:

                vals = np.zeros(vals.size, dtype=np.uint8)

                vals = np.ones(vals.size, dtype=np.uint8)

        df = pd.DataFrame({"row": rows, "col": cols, "val": vals})

        if num_rows < matrix._nvals:

            df.loc["..."] = ["..."] * 3

        return df

    if mask is not None and not mask.structure and df.shape != matrix.shape:

        nonzero = matrix.apply(unary.one["UINT8"]).new(mask=matrix.V, name="")

        num_rows = matrix._nvals if matrix._nvals <= max_rows else min_rows

        if min(nonzero._nvals, num_rows) > 2 * df.count().sum():

            rows, cols, vals = nonzero.ss.head(num_rows, sort=True)

            if mask.complement:

                    vals = vals.copy()

                vals[:] = 0

            df = pd.DataFrame({"row": rows, "col": cols, "val": vals})

            if num_rows < nonzero._nvals:

                df.loc["..."] = ["..."] * 3

            return df

    return df.where(pd.notnull(df), "")

def _get_vector_dataframe(vector, max_rows, min_rows, max_columns, *, mask=None):

    if not has_pandas:

        return

    if max_rows is None:  # pragma: no branch

        max_rows = pd.options.display.max_rows

    if min_rows is None:  # pragma: no branch

        min_rows = pd.options.display.min_rows

    if max_columns is None:  # pragma: no branch

        max_columns = _get_max_columns()

    columns, column_groups = _get_chunk(vector._size, max_columns, max_columns)

    df = pd.DataFrame(columns=columns, index=[""])

    for column_group, column_offset in column_groups:

        _update_vector_dataframe(df, vector, column_group, column_offset, mask=mask)

    if (

        num_rows = vector._nvals if vector._nvals <= max_rows else min_rows

        indices, vals = vector.ss.head(num_rows, sort=True)

        if mask is not None:

            if mask.complement:

                vals = np.zeros(vals.size, dtype=np.uint8)

                vals = np.ones(vals.size, dtype=np.uint8)

        df = pd.DataFrame({"index": indices, "val": vals})

        if num_rows < vector._nvals:

            df.loc["..."] = ["..."] * 2

        return df

    if mask is not None and not mask.structure and df.size != vector._size:

        nonzero = vector.apply(unary.one["UINT8"]).new(mask=vector.V, name="")

        num_rows = vector._nvals if vector._nvals <= max_rows else min_rows

        if min(nonzero._nvals, num_rows) > 2 * df.count().sum():

            indices, vals = nonzero.ss.head(num_rows, sort=True)

            if mask.complement:

                    vals = vals.copy()

                vals[:] = 0

            df = pd.DataFrame({"index": indices, "val": vals})

            if num_rows < nonzero._nvals:

                df.loc["..."] = ["..."] * 2

            return df

    return df.where(pd.notnull(df), "")

def get_format(x, is_transposed=False):

    fmt = x.ss.format

    if is_transposed:

        fmt = fmt[:-1] + ("c" if fmt[-1] == "r" else "r")

    if x.ss.is_iso:

        return f"{fmt} (iso)"

    return fmt

def matrix_info(matrix, *, mask=None, for_html=True):

    if mask is not None:

        if for_html:

            name = f"{type(mask).__name__}\nof\ngrblas.{type(matrix).__name__}"

            name = [f"{type(mask).__name__}", f"of grblas.{type(matrix).__name__}"]

        name = f"grblas.{type(matrix).__name__}"

    keys = ["nvals", "nrows", "ncols", "dtype", "format"]

    vals = [matrix._nvals, matrix._nrows, matrix._ncols, matrix.dtype.name]

    if type(matrix) is Matrix:

        vals.append(get_format(matrix))

        vals.append(get_format(matrix._matrix, is_transposed=True))

    return name, keys, vals

def vector_info(vector, *, mask=None, for_html=True):

    if mask is not None:

        if for_html:

            name = f"{type(mask).__name__}\nof\ngrblas.{type(vector).__name__}"

            name = [f"{type(mask).__name__}", f"of grblas.{type(vector).__name__}"]

        name = f"grblas.{type(vector).__name__}"

    keys = ["nvals", "size", "dtype", "format"]

    vals = [vector._nvals, vector._size, vector.dtype.name, get_format(vector)]

    return name, keys, vals

def create_header_html(name, keys, vals):

    text = [

    text.extend(f"    <td><pre>{key}</pre></td>\n" for key in keys)

    text.append("  </tr>\n  <tr>\n")

    text.extend(f"    <td>{val}</td>\n" for val in vals)

    text.append("  </tr>\n</table>\n</div>\n")

    return "".join(text)

def matrix_header_html(matrix, *, mask=None):

    name, keys, vals = matrix_info(matrix, mask=mask, for_html=True)

    return create_header_html(name, keys, vals)

def vector_header_html(vector, *, mask=None):

    name, keys, vals = vector_info(vector, mask=mask, for_html=True)

    return create_header_html(name, keys, vals)

def _format_html(name, header, df, collapse):

    if has_pandas:

        state = "" if collapse else " open"

        with pd.option_context("display.show_dimensions", False, "display.large_repr", "truncate"):

            details = df._repr_html_()

        state = ""

        details = "<em>(Install</em> <tt>pandas</tt> <em>to see a preview of the data)</em>"

    return (

def format_matrix_html(

    header = matrix_header_html(matrix, mask=mask)

    df = _get_matrix_dataframe(matrix, max_rows, min_rows, max_columns, mask=mask)

    if mask is None:

        name = matrix._name_html

        name = mask._name_html

    return _format_html(name, header, df, collapse)

def format_vector_html(

    header = vector_header_html(vector, mask=mask)

    df = _get_vector_dataframe(vector, max_rows, min_rows, max_columns, mask=mask)

    if mask is None:

        name = vector._name_html

        name = mask._name_html

    return _format_html(name, header, df, collapse)

def format_scalar_html(scalar):

    header = create_header_html("grblas.Scalar", ["value", "dtype"], [scalar.value, scalar.dtype])

    return f'<div class="gb-scalar"><tt>{scalar._name_html}</tt>{header}</div>'

def format_scalar(scalar):

    return create_header(

def get_expr_result(expr, html=False):

    try:

        val = expr._get_value()

    except OutOfMemory:

        arg_string = "Result is too large to compute!"

        if html:

            arg_string = f"<hr><b>{arg_string}</b>"

        name = val.name

        val.name = "Result"

        if html:

            arg_string = f"<hr>{val._repr_html_()}"

            arg_string = f"{val}"

        val.name = name

    return arg_string

def _format_expression(expr, header):

    pos_to_arg = {}

    for i, arg in enumerate(expr.args):

        pos = expr.expr_repr.find("{%s" % i)  # -1 if not found

        if pos >= 0:  # pragma: no branch

            pos_to_arg[pos] = arg

    args = [pos_to_arg[pos] for pos in sorted(pos_to_arg)]

    arg_string = "".join(x._repr_html_(collapse=True) for x in args if hasattr(x, "_repr_html_"))

    if config.get("autocompute"):

        arg_string += get_expr_result(expr, html=True)

    return (

def format_matrix_expression_html(expr):

    expr_html = expr._format_expr_html()

    header = create_header_html(

    return _format_expression(expr, header)

def get_result_string(expr):

    if config.get("autocompute"):

        arg_string = get_expr_result(expr)

        arg_string += "\n\n"

        arg_string = ""

    return arg_string

def format_matrix_expression(expr):

    expr_repr = expr._format_expr()

    name = f"grblas.{type(expr).__name__}"

    header = create_header(

    arg_string = get_result_string(expr)

    return (

def format_vector_expression_html(expr):

    expr_html = expr._format_expr_html()

    header = create_header_html(expr_html, ["size", "dtype"], [expr._size, expr.dtype])

    return _format_expression(expr, header)

def format_vector_expression(expr):

    expr_repr = expr._format_expr()

    name = f"grblas.{type(expr).__name__}"

    header = create_header(

    arg_string = get_result_string(expr)

    return (

def format_scalar_expression_html(expr):

    expr_html = expr._format_expr_html()

    header = create_header_html(expr_html, ["dtype"], [expr.dtype])

    return _format_expression(expr, header)

def format_scalar_expression(expr):

    expr_repr = expr._format_expr()

    name = f"grblas.{type(expr).__name__}"

    header = create_header(expr_repr, ["dtype"], [expr.dtype], name=name, quote=False)

    arg_string = get_result_string(expr)

    return (

def create_header(type_name, keys, vals, *, lower_border=False, name="", quote=True):

    vals = [str(x) for x in vals]

    if name and quote:

        name = f'"{name}"'

    key_text = []

    val_text = []

    for key, val in zip(keys, vals):

        width = max(len(key), len(val)) + 2

        key_text.append(key.rjust(width))

        val_text.append(val.rjust(width))

    if isinstance(type_name, str):

        name_width = max(len(type_name), len(name))

        lines = [

        name_width = max(map(len, type_name))

        name_width = max(name_width, len(name))

        lines = [f"{name.ljust(name_width)}{''.join(key_text)}"]

        lines.extend(line.ljust(name_width) for line in type_name)

        lines[-1] += "".join(val_text)

    if lower_border:

        lines.append("-" * len(lines[0]))

    return "\n".join(lines)

def format_matrix(matrix, *, max_rows=None, min_rows=None, max_columns=None, mask=None):

    name, keys, vals = matrix_info(matrix, mask=mask, for_html=False)

    header = create_header(

    if has_pandas:

        df = _get_matrix_dataframe(matrix, max_rows, min_rows, max_columns, mask=mask)

        with pd.option_context("display.show_dimensions", False, "display.large_repr", "truncate"):

            df_repr = df.__repr__()

        return f"{header}\n{df_repr}"

    return header

def format_vector(vector, *, max_rows=None, min_rows=None, max_columns=None, mask=None):

    name, keys, vals = vector_info(vector, mask=mask, for_html=False)

    header = create_header(

    if has_pandas:

        df = _get_vector_dataframe(vector, max_rows, min_rows, max_columns, mask=mask)

        with pd.option_context("display.show_dimensions", False, "display.large_repr", "truncate"):

            df_repr = df.__repr__()

        return f"{header}\n{df_repr}"

    return header

def _format_infix_expression(expr, header, expr_name):

    arg_string = f"{expr.left._repr_html_(collapse=True)}{expr.right._repr_html_(collapse=True)}"

    if config.get("autocompute") and (

        arg_string += get_expr_result(expr, html=True)

    return (

def format_scalar_infix_expression(expr):

    expr_repr = expr._format_expr()

    name = f"grblas.{type(expr).__name__}"

    header = create_header(

    arg_string = get_result_string(expr)

    return (

def format_scalar_infix_expression_html(expr):

    expr_html = expr._format_expr_html()

    header = create_header_html(

    return _format_infix_expression(expr, header, expr_html)

def get_infix_result_string(expr):

    if (

        arg_string = get_result_string(expr)

        arg_string = ""

    return arg_string

def format_vector_infix_expression(expr):

    expr_repr = expr._format_expr()

    name = f"grblas.{type(expr).__name__}"

    header = create_header(

    arg_string = get_infix_result_string(expr)

    return (

def format_vector_infix_expression_html(expr):

    expr_html = expr._format_expr_html()

    header = create_header_html(

    return _format_infix_expression(expr, header, expr_html)

def format_matrix_infix_expression(expr):

    expr_repr = expr._format_expr()

    name = f"grblas.{type(expr).__name__}"

    header = create_header(

    arg_string = get_infix_result_string(expr)

    return (

def format_matrix_infix_expression_html(expr):

    expr_html = expr._format_expr_html()

    header = create_header_html(

    return _format_infix_expression(expr, header, expr_html)

def format_index_expression(expr):

    name = f"grblas.{type(expr).__name__}"

    indices = ", ".join(index._expr_name for index in expr.resolved_indexes.indices)

    expr_repr = f"{expr.parent.name}[{indices}]"

    keys = []

    values = []

    if expr.output_type is Vector:

        keys.append("size")

        values.append(expr._size)

    elif expr.output_type is Matrix:

        keys.extend(["nrows", "ncols"])

        values.extend([expr._nrows, expr._ncols])

    keys.append("dtype")

    values.append(expr.dtype)

    header = create_header(

    arg_string = get_result_string(expr)

    c = expr.output_type.__name__[0]

    return (

def format_index_expression_html(expr):

    indices = ", ".join(index._expr_name for index in expr.resolved_indexes.indices)

    expr_repr = f"{expr.parent._name_html}[{indices}]"

    keys = []

    values = []

    if expr.output_type is Vector:

        keys.append("size")

        values.append(expr._size)

    elif expr.output_type is Matrix:

        keys.extend(["nrows", "ncols"])

        values.extend([expr._nrows, expr._ncols])

    keys.append("dtype")

    values.append(expr.dtype)

    header = create_header_html(

    arg_string = expr.parent._repr_html_(collapse=True)

    if config.get("autocompute"):

        arg_string += get_expr_result(expr, html=True)

    c = expr.output_type.__name__[0]

    c = "M" if c == "M" else c.lower()