16431585733

Committed 16 Jul 2025 07:02AM UTC coverage: 90.819% (+0.004%) from 90.815%

Build # 16431585733

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #2403 from GavinHuttley/develop

DEV: bump version to 2025.7.10a3

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

498 existing lines in 32 files now uncovered.

30123 of 33168 relevant lines covered (90.82%)

5.45 hits per line

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

94.02

/src/cogent3/util/dict_array.py

"""Wrapper for numpy arrays so that they can be indexed by name

>>> a = numpy.identity(3, int)
>>> b = DictArrayTemplate("abc", "ABC").wrap(a)
>>> b[0]
===========
A    B    C
-----------
1    0    0
-----------
>>> b["a"]
===========
A    B    C
-----------
1    0    0
-----------
>>> b.keys()
['a', 'b', 'c']
>>> b["a"].keys()
['A', 'B', 'C']
"""

import json
import os
import typing
from collections import defaultdict
from itertools import combinations, product

import numpy

from cogent3._version import __version__
from cogent3.util import warning as c3warn
from cogent3.util.deserialise import get_class, register_deserialiser
from cogent3.util.io import atomic_write
from cogent3.util.misc import get_object_provenance

PySeq = typing.Sequence
PySeqStr = PySeq[str]


def convert_1D_dict(data, row_order=None):
    """returns a 1D list and header as dict keys

    Parameters
    ----------
    data : dict
        a 1D dict
    row_order
        series with column headings. If not provided, the sorted top level dict
        keys are used.
    """
    if row_order is None:
        row_order = sorted(data)

    rows = [data[c] for c in row_order]
    return rows, row_order


def convert2Ddistance(dists, header=None, row_order=None):
    """returns a 2 dimensional list, header and row order

    Parameters
    ----------
    dists : dict
        a 1Ddict with {(a, b): dist, ..}
    header
        series with column headings. If not provided, the sorted top level dict
        keys are used.
    row_order
        a specified order to generate the rows

    Returns
    -------
    2D list, header and row_order. If a dist not present, it's set to 0, or
    the symmetric value e.g. (a, b) -> (b, a).
    """
    if header is None:
        names = set()
        for pair in dists:
            names.update(set(pair))
        header = sorted(names)

    rows = []
    for i in range(len(header)):
        n1 = header[i]
        row = []
        for j in range(len(header)):
            n2 = header[j]
            dist = dists.get((n1, n2), dists.get((n2, n1), 0))
            row.append(dist)
        rows.append(row)

    row_order = header[:]
    return rows, row_order, header


def convert2DDict(twoDdict, header=None, row_order=None, make_symmetric=False):
    """returns a 2 dimensional list, header and row order

    Parameters
    ----------
    twoDdict : dict
        a 2 dimensional dict with top level keys corresponding to column
        headings, lower level keys correspond to row headings
    header
        series with column headings. If not provided, the sorted top level dict
        keys are used.
    row_order
        a specified order to generate the rows
    make_symmetric : bool
        if True, twoDdict[a][b] == twoDdict[b][a]
    """
    if not row_order:
        row_order = list(twoDdict.keys())
        row_order.sort()

    if not header:  # we assume columns consistent across dict
        header = list(twoDdict[row_order[0]].keys())
        header.sort()

    if make_symmetric:
        combined = sorted(set(header) | set(row_order))
        header = row_order = combined
        data = defaultdict(dict)

        for k1, k2 in combinations(combined, 2):
            if k1 in twoDdict:
                val = twoDdict[k1].get(k2, 0)
            elif k2 in twoDdict:
                val = twoDdict[k2].get(k1, 0)
            else:
                val = 0
            data[k1][k2] = data[k2][k1] = val
        for k in data:
            data[k][k] = 0
        twoDdict = data

    # make list of lists
    rows = []
    for row in row_order:
        elements = []
        for column in header:
            elements.append(twoDdict[row][column])
        rows.append(elements)

    return rows, row_order, header


def convert_dict(data, header=None, row_order=None):
    """returns a list, DictArrayTemplate args

    Parameters
    ----------
    data : dict
        a 1D or 2D dict
    header
        series with column headings. If not provided, the sorted top level dict
        keys are used.
    row_order
        a specified order to generate the rows
    """
    first_key = next(iter(data))
    if type(first_key) == tuple and len(first_key) == 2:
        rows, row_order, header = convert2Ddistance(data, header, row_order)
    elif hasattr(data[first_key], "keys"):
        rows, row_order, header = convert2DDict(data, header, row_order)
    else:
        rows, row_order = convert_1D_dict(data, header)
    return rows, row_order, header


def convert_series(data, row_order=None, header=None):
    """returns a list, header and row order

    Parameters
    ----------
    data : dict
        a 1D or 2D dict
    header
        series with column headings. If not provided, the sorted top level dict
        keys are used.
    row_order
        a specified order to generate the rows
    """
    first_element = data[0]
    nrows = len(data)
    try:
        ncols = len(first_element)
    except TypeError:
        ncols = 1

    if header is not None:
        dim_h = header if isinstance(header, int) else len(header)
    else:
        dim_h = None

    if row_order is not None:
        dim_r = row_order if isinstance(row_order, int) else len(row_order)
    else:
        dim_r = None

    if nrows == 1 and ncols > 1:
        if dim_h is not None and dim_h != ncols:
            msg = (
                f"mismatch between number columns={dim_h} "
                f"and number of elements in data={ncols}"
            )
            raise ValueError(
                msg,
            )
        if dim_r is not None and dim_r != 1:
            msg = (
                f"mismatch between number rows={dim_r} "
                f"and number of rows in data={ncols}"
            )
            raise ValueError(
                msg,
            )

    if not header:
        header = None if ncols == 1 else ncols
    row_order = row_order if row_order else nrows

    return data, row_order, header


def convert_for_dictarray(data, header=None, row_order=None):
    """returns a list, header and row order from data

    Parameters
    ----------
    data : iterable
        data series, dictarray, dict, etc..
    header
        series with column headings. If not provided, the sorted top level dict
        keys are used.
    row_order
        a specified order to generate the rows
    """
    if isinstance(data, DictArray):
        header = data.template.names[0]
        row_order = data.template.names[1]
        data = data.array.copy()
    elif hasattr(data, "keys"):  # dictlike, it could be defaultdict
        data, row_order, header = convert_dict(data, header, row_order)
    else:
        data, row_order, header = convert_series(data, header, row_order)

    return data, row_order, header


class NumericKey(int):
    """a distinct numerical type for use as a DictArray key"""

    def __new__(cls, val):
        return int.__new__(cls, val)


class DictArrayTemplate:
    def __init__(self, *dimensions) -> None:
        self.names = []
        self.ordinals = []
        for names in dimensions:
            if names is None:
                continue
            if isinstance(names, int):
                names = list(range(names))
            else:
                names = [NumericKey(v) if type(v) == int else v for v in names]

            self.names.append(names)
            self.ordinals.append({c: i for (i, c) in enumerate(names)})
        self._shape = tuple(len(keys) for keys in self.names)

    def __eq__(self, other):
        return self is other or (
            isinstance(other, DictArrayTemplate) and self.names == other.names
        )

    def _dict2list(self, value, depth=0):
        # Unpack (possibly nested) dictionary into correct order of elements
        if depth < len(self._shape):
            return [self._dict2list(value[key], depth + 1) for key in self.names[depth]]
        return value

    def unwrap(self, value):
        """Convert to a simple numpy array"""
        if isinstance(value, DictArray):
            if value.template == self:
                value = value.array
            else:
                raise ValueError  # used to return None, which can't be right
        elif isinstance(value, dict):
            value = self._dict2list(value)
        value = numpy.asarray(value)
        assert value.shape == self._shape, (value.shape, self._shape)
        return value

    def wrap(
        self,
        array: numpy.ndarray,
        dtype: numpy.dtype | None = None,
    ) -> "DictArray":
        if hasattr(array, "keys"):
            if len(self._shape) == 2:
                r, h = self.names[:2]
            else:
                r, h = self.names[0], None
            array, _, _ = convert_for_dictarray(array, h, r)
        array = numpy.asarray(array, dtype=dtype)
        for dim, categories in enumerate(self.names):
            assert len(categories) == numpy.shape(array)[dim], (
                f"cats={categories}; dim={dim}"
            )
        return DictArray(array, self)

    def interpret_index(self, names):
        if isinstance(names, numpy.ndarray) and "int" in names.dtype.name:
            # the numpy item() method casts to the nearest Python type
            names = tuple(v.item() for v in names)

        if not isinstance(names, tuple):
            names = (names,)

        index = []
        remaining = []
        for ordinals, allnames, name in zip(
            self.ordinals,
            self.names,
            names,
            strict=False,
        ):
            if type(name) not in (int, slice, list, numpy.ndarray):
                name = ordinals[name]
            elif isinstance(name, slice):
                start = name.start
                stop = name.stop
                try:
                    start = allnames.index(start)
                except ValueError:
                    # either None, or it's an int index
                    pass
                try:
                    stop = allnames.index(stop)
                except ValueError:
                    # as above
                    pass
                name = slice(start, stop, name.step)
                remaining.append(allnames.__getitem__(name))
            elif type(name) in (list, numpy.ndarray):
                name = [n if type(n) == int else ordinals[n] for n in name]
                remaining.append([allnames[i] for i in name])

            index.append(name)
        remaining.extend(self.names[len(index) :])
        klass = type(self)(*remaining) if remaining else None
        return (tuple(index), klass)


class DictArray:
    """Wraps a numpy array so that it can be indexed with strings. Behaves
    like nested dictionaries (only ordered).

    Notes
    -----
    Used for things like substitution matrices and bin probabilities.

    Indexing can be done via conventional integer based operations, using
    keys, lists of int/keys.

    Behaviour differs from numpy array indexing when you provide lists of
    indices. Such indexing is applied sequentially, e.g. darr[[0, 2], [1, 2]]
    will return the intersection of rows [0, 2] with columns [1, 2]. In numpy,
    the result would instead be the elements at [0, 1], [2, 2].
    """

    def __init__(self, *args, **kwargs) -> None:
        """allow alternate ways of creating for time being"""
        if len(args) == 1:
            vals, row_keys, col_keys = convert_for_dictarray(args[0])
            dtype = kwargs.get("dtype")
            self.array = numpy.asarray(vals, dtype=dtype)
            self.template = DictArrayTemplate(row_keys, col_keys)
        elif len(args) == 2:
            if not isinstance(args[1], DictArrayTemplate):
                raise NotImplementedError
            self.array = args[0]
            self.template = args[1]
        else:
            if "dtype" in kwargs or "typecode" in kwargs:
                dtype = kwargs["dtype"]
                kwargs.pop("dtype", None)
                kwargs.pop("typecode", None)
            else:
                dtype = None
            create_new = DictArrayTemplate(*args[1:]).wrap(args[0], dtype=dtype)
            self.__dict__ = create_new.__dict__
        self.shape = self.array.shape

    @classmethod
    def from_array_names(cls, array: numpy.ndarray, *names: tuple[PySeqStr, ...]):
        """creates instance directly from a numpy array and series of names

        Parameters
        ----------
        array
            any data type
        names
            must match the array dimensions
        """

        if len(names) != array.ndim or any(
            len(labels) != array.shape[dim] for dim, labels in enumerate(names)
        ):
            msg = "names must match array dimensions"
            raise ValueError(msg)

        template = DictArrayTemplate(*names)
        return DictArray(array, template)

    def to_array(self):
        return self.array

    def __add__(self, other):
        if not isinstance(other, type(self)):
            msg = f"Incompatible types: {type(self)} and {type(other)}"
            raise TypeError(msg)

        if other.template.names != self.template.names:
            msg = f"unequal dimension names {self.template.names} != {other.template.names}"
            raise ValueError(
                msg,
            )

        return self.template.wrap(self.array + other.array)

    def __array__(
        self,
        dtype: numpy.dtype | None = None,
        copy: bool = False,
    ) -> numpy.ndarray:
        array = self.array
        if dtype is not None:
            array = array.astype(dtype)
        return array

    def to_dict(self, flatten=False):
        """returns data as a dict

        Parameters
        ----------
        flatten : bool
            returns a 1D dictionary
        """
        names = self.template.names
        shape = self.shape
        result = {}
        if len(names) == 1:
            result = {
                names[0][i]: v.item() if hasattr(v, "item") else v
                for i, v in enumerate(self.array)
            }
        elif flatten:
            for indices in product(*[range(n) for n in shape]):
                value = self.array[indices]
                value = value.item() if hasattr(value, "item") else value
                coord = tuple(n[i] for n, i in zip(names, indices, strict=False))
                result[coord] = value
        else:
            for indices in product(*[range(n) for n in shape]):
                value = self.array[indices]
                value = value.item() if hasattr(value, "item") else value
                coord = tuple(n[i] for n, i in zip(names, indices, strict=False))
                current = result
                nested = coord[0]
                for nested in coord[:-1]:
                    current[nested] = current.get(nested, {})
                current[nested][coord[-1]] = value

        return result

    def to_rich_dict(self):
        data = self.array.tolist()
        return {
            "type": get_object_provenance(self.template),
            "array": data,
            "names": self.template.names,
            "version": __version__,
        }

    def to_json(self):
        return json.dumps(self.to_rich_dict())

    def __getitem__(self, names):
        (index, remaining) = self.template.interpret_index(names)
        if list in {type(v) for v in index}:
            result = self.array
            for dim, indices in enumerate(index):
                if isinstance(indices, slice):
                    indices = (
                        (indices,)
                        if dim == 0
                        else (slice(None, None),) * dim + (indices,)
                    )
                    result = result[tuple(indices)]
                    continue

                if isinstance(indices, int):
                    indices = [indices]

                result = result.take(indices, axis=dim)

        else:
            result = self.array[index]

        if remaining is not None:
            result = self.__class__(result.reshape(remaining._shape), remaining)
        return result

    def __iter__(self):
        (index, remaining) = self.template.interpret_index(0)
        for elt in self.array:
            if remaining is None:
                yield elt
            else:
                yield remaining.wrap(elt)

    def __len__(self) -> int:
        return len(self.template.names[0])

    def keys(self):
        return self.template.names[0][:]

    def items(self):
        return [(n, self[n]) for n in list(self.keys())]

    def __repr__(self) -> str:
        if self.array.ndim > 2:
            return f"{self.array.ndim} dimensional {type(self).__name__}"

        t = self.to_table()
        t.set_repr_policy(show_shape=False)
        return str(t)

    def __ne__(self, other):
        return not self.__eq__(other)

    def __eq__(self, other):
        if self is other:
            return True
        if isinstance(other, DictArray):
            return self.template == other.template and numpy.all(
                self.array == other.array,
            )
        if isinstance(other, type(self.array)):
            return self.array == other
        if isinstance(other, dict):
            return self.to_dict() == other
        return False

    def to_normalized(self, by_row=False, by_column=False):
        """returns a DictArray as frequencies

        Parameters
        ----------
        by_row
            rows sum to 1
        by_col
            columns sum to 1
        """
        assert not (by_row and by_column)
        # TODO need to check there are two dimension!
        if by_row:
            axis = 1
        elif by_column:
            axis = 0
        else:
            axis = None

        result = self.array / self.array.sum(axis=axis, keepdims=True)
        return self.template.wrap(result)

    def row_sum(self):
        """returns DictArray summed across rows"""
        axis = 1 if len(self.shape) == 2 else 0
        result = self.array.sum(axis=axis)
        template = DictArrayTemplate(self.template.names[0])
        return template.wrap(result)

    def col_sum(self):
        """returns DictArray summed across columns"""
        result = self.array.sum(axis=0)
        template = DictArrayTemplate(self.template.names[1])
        return template.wrap(result)

    def _repr_html_(self):
        if self.array.ndim > 2:
            return f"{self.array.ndim} dimensional {type(self).__name__}"

        t = self.to_table()
        t.set_repr_policy(show_shape=False)
        return t._repr_html_()

    @c3warn.deprecated_args(
        "2025.9", "don't use built in name", old_new=[("format", "format_name")]
    )
    def to_string(self, format_name: str = "tsv", sep: str | None = None) -> str:
        """Return the data as a formatted string.

        Parameters
        ----------
        format_name
            possible formats are 'csv', or 'tsv' (default).
        sep
            A string separator for delineating columns, e.g. ',' or
            '\t'. Overrides format.
        """
        if format_name.lower() not in ("tsv", "csv"):
            msg = f"'{format_name}' not supported"
            raise ValueError(msg)

        sep = sep or {"tsv": "\t", "csv": ","}[format_name.lower()]

        data = self.to_dict(flatten=True)
        rows = [[f"dim-{i + 1}" for i in range(self.array.ndim)] + ["value"]] + [
            [str(x) for x in row] for row in [[*list(k), v] for k, v in data.items()]
        ]
        return "\n".join([sep.join(row) for row in rows])

    def to_table(self):
        """return Table instance

        Notes
        -----
        Raises ValueError if number of dimensions > 2
        """
        ndim = self.array.ndim
        if ndim > 2:
            msg = f"cannot make 2D table from {ndim}D array"
            raise ValueError(msg)

        from cogent3.core.table import Table

        header = self.template.names[0] if ndim == 1 else self.template.names[1]
        index = "" if ndim == 2 else None
        if ndim == 1:
            data = {c: [v] for c, v in zip(header, self.array, strict=False)}
        else:
            data = {c: self.array[:, i].tolist() for i, c in enumerate(header)}
            data[""] = self.template.names[0]

        return Table(header=header, data=data, index_name=index)

    @c3warn.deprecated_args(
        "2025.9", "don't use built in name", old_new=[("format", "format_name")]
    )
    def write(
        self, path: str | os.PathLike, format_name: str = "tsv", sep: str = "\t"
    ) -> None:
        """writes a flattened version to path

        Parameters
        ----------
        path
        format_name
            possible formats are 'rest'/'rst', 'markdown'/'md',
            'latex', 'html', 'phylip', 'bedgraph', 'csv', 'tsv', or 'simple'
            (default).
        sep
            used to split fields, will be inferred from path suffix if not
            provided
        """
        data = self.to_string(format_name=format_name, sep=sep)
        with atomic_write(path, mode="wt") as outfile:
            outfile.write(data)


@register_deserialiser(
    get_object_provenance(DictArrayTemplate),
)
def deserialise_dict_array(data: dict) -> DictArray:
    """deserialising DictArray, Table instances"""
    data.pop("version", None)
    type_ = data.pop("type")
    klass = get_class(type_)
    named_dims = data.pop("names")
    array = data.pop("array")
    template = klass(*named_dims)
    return template.wrap(array)

1	"""Wrapper for numpy arrays so that they can be indexed by name
2
3	>>> a = numpy.identity(3, int)
4	>>> b = DictArrayTemplate("abc", "ABC").wrap(a)
5	>>> b[0]
6	===========
7	A B C
8	-----------
9	1 0 0
10	-----------
11	>>> b["a"]
12	===========
13	A B C
14	-----------
15	1 0 0
16	-----------
17	>>> b.keys()
18	['a', 'b', 'c']
19	>>> b["a"].keys()
20	['A', 'B', 'C']
21	"""
22
23	import json	6✔
24	import os	6✔
25	import typing	6✔
26	from collections import defaultdict	6✔
27	from itertools import combinations, product	6✔
28
29	import numpy	6✔
30
31	from cogent3._version import __version__	6✔
32	from cogent3.util import warning as c3warn	6✔
33	from cogent3.util.deserialise import get_class, register_deserialiser	6✔
34	from cogent3.util.io import atomic_write	6✔
35	from cogent3.util.misc import get_object_provenance	6✔
36
37	PySeq = typing.Sequence	6✔
38	PySeqStr = PySeq[str]	6✔
39
40
41	def convert_1D_dict(data, row_order=None):	6✔
42	"""returns a 1D list and header as dict keys
43
44	Parameters
45	----------
46	data : dict
47	a 1D dict
48	row_order
49	series with column headings. If not provided, the sorted top level dict
50	keys are used.
51	"""
52	if row_order is None:	6✔
53	row_order = sorted(data)	6✔
54
55	rows = [data[c] for c in row_order]	6✔
56	return rows, row_order	6✔
57
58
59	def convert2Ddistance(dists, header=None, row_order=None):	6✔
60	"""returns a 2 dimensional list, header and row order
61
62	Parameters
63	----------
64	dists : dict
65	a 1Ddict with {(a, b): dist, ..}
66	header
67	series with column headings. If not provided, the sorted top level dict
68	keys are used.
69	row_order
70	a specified order to generate the rows
71
72	Returns
73	-------
74	2D list, header and row_order. If a dist not present, it's set to 0, or
75	the symmetric value e.g. (a, b) -> (b, a).
76	"""
77	if header is None:	6✔
78	names = set()	6✔
79	for pair in dists:	6✔
80	names.update(set(pair))	6✔
81	header = sorted(names)	6✔
82
83	rows = []	6✔
84	for i in range(len(header)):	6✔
85	n1 = header[i]	6✔
86	row = []	6✔
87	for j in range(len(header)):	6✔
88	n2 = header[j]	6✔
89	dist = dists.get((n1, n2), dists.get((n2, n1), 0))	6✔
90	row.append(dist)	6✔
91	rows.append(row)	6✔
92
93	row_order = header[:]	6✔
94	return rows, row_order, header	6✔
95
96
97	def convert2DDict(twoDdict, header=None, row_order=None, make_symmetric=False):	6✔
98	"""returns a 2 dimensional list, header and row order
99
100	Parameters
101	----------
102	twoDdict : dict
103	a 2 dimensional dict with top level keys corresponding to column
104	headings, lower level keys correspond to row headings
105	header
106	series with column headings. If not provided, the sorted top level dict
107	keys are used.
108	row_order
109	a specified order to generate the rows
110	make_symmetric : bool
111	if True, twoDdict[a][b] == twoDdict[b][a]
112	"""
113	if not row_order:	6✔
114	row_order = list(twoDdict.keys())	6✔
115	row_order.sort()	6✔
116
117	if not header: # we assume columns consistent across dict	6✔
118	header = list(twoDdict[row_order[0]].keys())	6✔
119	header.sort()	6✔
120
121	if make_symmetric:	6✔
122	combined = sorted(set(header) \| set(row_order))	6✔
123	header = row_order = combined	6✔
124	data = defaultdict(dict)	6✔
125
126	for k1, k2 in combinations(combined, 2):	6✔
127	if k1 in twoDdict:	6✔
128	val = twoDdict[k1].get(k2, 0)	6✔
129	elif k2 in twoDdict:	6✔
UNCOV 130	val = twoDdict[k2].get(k1, 0)	×
131	else:
132	val = 0	6✔
133	data[k1][k2] = data[k2][k1] = val	6✔
134	for k in data:	6✔
135	data[k][k] = 0	6✔
136	twoDdict = data	6✔
137
138	# make list of lists
139	rows = []	6✔
140	for row in row_order:	6✔
141	elements = []	6✔
142	for column in header:	6✔
143	elements.append(twoDdict[row][column])	6✔
144	rows.append(elements)	6✔
145
146	return rows, row_order, header	6✔
147
148
149	def convert_dict(data, header=None, row_order=None):	6✔
150	"""returns a list, DictArrayTemplate args
151
152	Parameters
153	----------
154	data : dict
155	a 1D or 2D dict
156	header
157	series with column headings. If not provided, the sorted top level dict
158	keys are used.
159	row_order
160	a specified order to generate the rows
161	"""
162	first_key = next(iter(data))	6✔
163	if type(first_key) == tuple and len(first_key) == 2:	6✔
164	rows, row_order, header = convert2Ddistance(data, header, row_order)	6✔
165	elif hasattr(data[first_key], "keys"):	6✔
166	rows, row_order, header = convert2DDict(data, header, row_order)	6✔
167	else:
168	rows, row_order = convert_1D_dict(data, header)	6✔
169	return rows, row_order, header	6✔
170
171
172	def convert_series(data, row_order=None, header=None):	6✔
173	"""returns a list, header and row order
174
175	Parameters
176	----------
177	data : dict
178	a 1D or 2D dict
179	header
180	series with column headings. If not provided, the sorted top level dict
181	keys are used.
182	row_order
183	a specified order to generate the rows
184	"""
185	first_element = data[0]	6✔
186	nrows = len(data)	6✔
187	try:	6✔
188	ncols = len(first_element)	6✔
189	except TypeError:	6✔
190	ncols = 1	6✔
191
192	if header is not None:	6✔
193	dim_h = header if isinstance(header, int) else len(header)	6✔
194	else:
195	dim_h = None	6✔
196
197	if row_order is not None:	6✔
198	dim_r = row_order if isinstance(row_order, int) else len(row_order)	6✔
199	else:
200	dim_r = None	6✔
201
202	if nrows == 1 and ncols > 1:	6✔
203	if dim_h is not None and dim_h != ncols:	6✔
UNCOV 204	msg = (	×
205	f"mismatch between number columns={dim_h} "
206	f"and number of elements in data={ncols}"
207	)
UNCOV 208	raise ValueError(	×
209	msg,
210	)
211	if dim_r is not None and dim_r != 1:	6✔
UNCOV 212	msg = (	×
213	f"mismatch between number rows={dim_r} "
214	f"and number of rows in data={ncols}"
215	)
UNCOV 216	raise ValueError(	×
217	msg,
218	)
219
220	if not header:	6✔
221	header = None if ncols == 1 else ncols	6✔
222	row_order = row_order if row_order else nrows	6✔
223
224	return data, row_order, header	6✔
225
226
227	def convert_for_dictarray(data, header=None, row_order=None):	6✔
228	"""returns a list, header and row order from data
229
230	Parameters
231	----------
232	data : iterable
233	data series, dictarray, dict, etc..
234	header
235	series with column headings. If not provided, the sorted top level dict
236	keys are used.
237	row_order
238	a specified order to generate the rows
239	"""
240	if isinstance(data, DictArray):	6✔
241	header = data.template.names[0]	6✔
242	row_order = data.template.names[1]	6✔
243	data = data.array.copy()	6✔
244	elif hasattr(data, "keys"): # dictlike, it could be defaultdict	6✔
245	data, row_order, header = convert_dict(data, header, row_order)	6✔
246	else:
247	data, row_order, header = convert_series(data, header, row_order)	6✔
248
249	return data, row_order, header	6✔
250
251
252	class NumericKey(int):	6✔
253	"""a distinct numerical type for use as a DictArray key"""
254
255	def __new__(cls, val):	6✔
256	return int.__new__(cls, val)	6✔
257
258
259	class DictArrayTemplate:	6✔
260	def __init__(self, *dimensions) -> None:	6✔
261	self.names = []	6✔
262	self.ordinals = []	6✔
263	for names in dimensions:	6✔
264	if names is None:	6✔
265	continue	6✔
266	if isinstance(names, int):	6✔
267	names = list(range(names))	6✔
268	else:
269	names = [NumericKey(v) if type(v) == int else v for v in names]	6✔
270
271	self.names.append(names)	6✔
272	self.ordinals.append({c: i for (i, c) in enumerate(names)})	6✔
273	self._shape = tuple(len(keys) for keys in self.names)	6✔
274
275	def __eq__(self, other):	6✔
276	return self is other or (	6✔
277	isinstance(other, DictArrayTemplate) and self.names == other.names
278	)
279
280	def _dict2list(self, value, depth=0):	6✔
281	# Unpack (possibly nested) dictionary into correct order of elements
282	if depth < len(self._shape):	6✔
283	return [self._dict2list(value[key], depth + 1) for key in self.names[depth]]	6✔
284	return value	6✔
285
286	def unwrap(self, value):	6✔
287	"""Convert to a simple numpy array"""
288	if isinstance(value, DictArray):	6✔
289	if value.template == self:	6✔
290	value = value.array	6✔
291	else:
UNCOV 292	raise ValueError # used to return None, which can't be right	×
293	elif isinstance(value, dict):	6✔
294	value = self._dict2list(value)	6✔
295	value = numpy.asarray(value)	6✔
296	assert value.shape == self._shape, (value.shape, self._shape)	6✔
297	return value	6✔
298
299	def wrap(	6✔
300	self,
301	array: numpy.ndarray,
302	dtype: numpy.dtype \| None = None,
303	) -> "DictArray":
304	if hasattr(array, "keys"):	6✔
305	if len(self._shape) == 2:	6✔
306	r, h = self.names[:2]	6✔
307	else:
308	r, h = self.names[0], None	6✔
309	array, _, _ = convert_for_dictarray(array, h, r)	6✔
310	array = numpy.asarray(array, dtype=dtype)	6✔
311	for dim, categories in enumerate(self.names):	6✔
312	assert len(categories) == numpy.shape(array)[dim], (	6✔
313	f"cats={categories}; dim={dim}"
314	)
315	return DictArray(array, self)	6✔
316
317	def interpret_index(self, names):	6✔
318	if isinstance(names, numpy.ndarray) and "int" in names.dtype.name:	6✔
319	# the numpy item() method casts to the nearest Python type
320	names = tuple(v.item() for v in names)	6✔
321
322	if not isinstance(names, tuple):	6✔
323	names = (names,)	6✔
324
325	index = []	6✔
326	remaining = []	6✔
327	for ordinals, allnames, name in zip(	6✔
328	self.ordinals,
329	self.names,
330	names,
331	strict=False,
332	):
333	if type(name) not in (int, slice, list, numpy.ndarray):	6✔
334	name = ordinals[name]	6✔
335	elif isinstance(name, slice):	6✔
336	start = name.start	6✔
337	stop = name.stop	6✔
338	try:	6✔
339	start = allnames.index(start)	6✔
340	except ValueError:	6✔
341	# either None, or it's an int index
342	pass	6✔
343	try:	6✔
344	stop = allnames.index(stop)	6✔
345	except ValueError:	6✔
346	# as above
347	pass	6✔
348	name = slice(start, stop, name.step)	6✔
349	remaining.append(allnames.__getitem__(name))	6✔
350	elif type(name) in (list, numpy.ndarray):	6✔
351	name = [n if type(n) == int else ordinals[n] for n in name]	6✔
352	remaining.append([allnames[i] for i in name])	6✔
353
354	index.append(name)	6✔
355	remaining.extend(self.names[len(index) :])	6✔
356	klass = type(self)(*remaining) if remaining else None	6✔
357	return (tuple(index), klass)	6✔
358
359
360	class DictArray:	6✔
361	"""Wraps a numpy array so that it can be indexed with strings. Behaves
362	like nested dictionaries (only ordered).
363
364	Notes
365	-----
366	Used for things like substitution matrices and bin probabilities.
367
368	Indexing can be done via conventional integer based operations, using
369	keys, lists of int/keys.
370
371	Behaviour differs from numpy array indexing when you provide lists of
372	indices. Such indexing is applied sequentially, e.g. darr[[0, 2], [1, 2]]
373	will return the intersection of rows [0, 2] with columns [1, 2]. In numpy,
374	the result would instead be the elements at [0, 1], [2, 2].
375	"""
376
377	def __init__(self, args, *kwargs) -> None:	6✔
378	"""allow alternate ways of creating for time being"""
379	if len(args) == 1:	6✔
380	vals, row_keys, col_keys = convert_for_dictarray(args[0])	6✔
381	dtype = kwargs.get("dtype")	6✔
382	self.array = numpy.asarray(vals, dtype=dtype)	6✔
383	self.template = DictArrayTemplate(row_keys, col_keys)	6✔
384	elif len(args) == 2:	6✔
385	if not isinstance(args[1], DictArrayTemplate):	6✔
UNCOV 386	raise NotImplementedError	×
387	self.array = args[0]	6✔
388	self.template = args[1]	6✔
389	else:
390	if "dtype" in kwargs or "typecode" in kwargs:	×
391	dtype = kwargs["dtype"]	×
UNCOV 392	kwargs.pop("dtype", None)	×
393	kwargs.pop("typecode", None)	×
394	else:
395	dtype = None	×
UNCOV 396	create_new = DictArrayTemplate(*args[1:]).wrap(args[0], dtype=dtype)	×
UNCOV 397	self.__dict__ = create_new.__dict__	×
398	self.shape = self.array.shape	6✔
399
400	@classmethod	6✔
401	def from_array_names(cls, array: numpy.ndarray, *names: tuple[PySeqStr, ...]):	6✔
402	"""creates instance directly from a numpy array and series of names
403
404	Parameters
405	----------
406	array
407	any data type
408	names
409	must match the array dimensions
410	"""
411
412	if len(names) != array.ndim or any(	6✔
413	len(labels) != array.shape[dim] for dim, labels in enumerate(names)
414	):
UNCOV 415	msg = "names must match array dimensions"	×
UNCOV 416	raise ValueError(msg)	×
417
418	template = DictArrayTemplate(*names)	6✔
419	return DictArray(array, template)	6✔
420
421	def to_array(self):	6✔
422	return self.array	6✔
423
424	def __add__(self, other):	6✔
425	if not isinstance(other, type(self)):	6✔
426	msg = f"Incompatible types: {type(self)} and {type(other)}"	6✔
427	raise TypeError(msg)	6✔
428
429	if other.template.names != self.template.names:	6✔
430	msg = f"unequal dimension names {self.template.names} != {other.template.names}"	6✔
431	raise ValueError(	6✔
432	msg,
433	)
434
435	return self.template.wrap(self.array + other.array)	6✔
436
437	def __array__(	6✔
438	self,
439	dtype: numpy.dtype \| None = None,
440	copy: bool = False,
441	) -> numpy.ndarray:
442	array = self.array	6✔
443	if dtype is not None:	6✔
444	array = array.astype(dtype)	6✔
445	return array	6✔
446
447	def to_dict(self, flatten=False):	6✔
448	"""returns data as a dict
449
450	Parameters
451	----------
452	flatten : bool
453	returns a 1D dictionary
454	"""
455	names = self.template.names	6✔
456	shape = self.shape	6✔
457	result = {}	6✔
458	if len(names) == 1:	6✔
459	result = {	6✔
460	names[0][i]: v.item() if hasattr(v, "item") else v
461	for i, v in enumerate(self.array)
462	}
463	elif flatten:	6✔
464	for indices in product(*[range(n) for n in shape]):	6✔
465	value = self.array[indices]	6✔
466	value = value.item() if hasattr(value, "item") else value	6✔
467	coord = tuple(n[i] for n, i in zip(names, indices, strict=False))	6✔
468	result[coord] = value	6✔
469	else:
470	for indices in product(*[range(n) for n in shape]):	6✔
471	value = self.array[indices]	6✔
472	value = value.item() if hasattr(value, "item") else value	6✔
473	coord = tuple(n[i] for n, i in zip(names, indices, strict=False))	6✔
474	current = result	6✔
475	nested = coord[0]	6✔
476	for nested in coord[:-1]:	6✔
477	current[nested] = current.get(nested, {})	6✔
478	current[nested][coord[-1]] = value	6✔
479
480	return result	6✔
481
482	def to_rich_dict(self):	6✔
483	data = self.array.tolist()	6✔
484	return {	6✔
485	"type": get_object_provenance(self.template),
486	"array": data,
487	"names": self.template.names,
488	"version": __version__,
489	}
490
491	def to_json(self):	6✔
492	return json.dumps(self.to_rich_dict())	6✔
493
494	def __getitem__(self, names):	6✔
495	(index, remaining) = self.template.interpret_index(names)	6✔
496	if list in {type(v) for v in index}:	6✔
497	result = self.array	6✔
498	for dim, indices in enumerate(index):	6✔
499	if isinstance(indices, slice):	6✔
500	indices = (	6✔
501	(indices,)
502	if dim == 0
503	else (slice(None, None),) * dim + (indices,)
504	)
505	result = result[tuple(indices)]	6✔
506	continue	6✔
507
508	if isinstance(indices, int):	6✔
509	indices = [indices]	6✔
510
511	result = result.take(indices, axis=dim)	6✔
512
513	else:
514	result = self.array[index]	6✔
515
516	if remaining is not None:	6✔
517	result = self.__class__(result.reshape(remaining._shape), remaining)	6✔
518	return result	6✔
519
520	def __iter__(self):	6✔
521	(index, remaining) = self.template.interpret_index(0)	6✔
522	for elt in self.array:	6✔
523	if remaining is None:	6✔
524	yield elt	6✔
525	else:
526	yield remaining.wrap(elt)	6✔
527
528	def __len__(self) -> int:	6✔
529	return len(self.template.names[0])	6✔
530
531	def keys(self):	6✔
532	return self.template.names[0][:]	6✔
533
534	def items(self):	6✔
535	return [(n, self[n]) for n in list(self.keys())]	6✔
536
537	def __repr__(self) -> str:	6✔
538	if self.array.ndim > 2:	6✔
UNCOV 539	return f"{self.array.ndim} dimensional {type(self).__name__}"	×
540
541	t = self.to_table()	6✔
542	t.set_repr_policy(show_shape=False)	6✔
543	return str(t)	6✔
544
545	def __ne__(self, other):	6✔
546	return not self.__eq__(other)	6✔
547
548	def __eq__(self, other):	6✔
549	if self is other:	6✔
UNCOV 550	return True	×
551	if isinstance(other, DictArray):	6✔
552	return self.template == other.template and numpy.all(	6✔
553	self.array == other.array,
554	)
555	if isinstance(other, type(self.array)):	6✔
UNCOV 556	return self.array == other	×
557	if isinstance(other, dict):	6✔
558	return self.to_dict() == other	6✔
UNCOV 559	return False	×
560
561	def to_normalized(self, by_row=False, by_column=False):	6✔
562	"""returns a DictArray as frequencies
563
564	Parameters
565	----------
566	by_row
567	rows sum to 1
568	by_col
569	columns sum to 1
570	"""
571	assert not (by_row and by_column)	6✔
572	# TODO need to check there are two dimension!
573	if by_row:	6✔
574	axis = 1	6✔
575	elif by_column:	6✔
576	axis = 0	6✔
577	else:
UNCOV 578	axis = None	×
579
580	result = self.array / self.array.sum(axis=axis, keepdims=True)	6✔
581	return self.template.wrap(result)	6✔
582
583	def row_sum(self):	6✔
584	"""returns DictArray summed across rows"""
585	axis = 1 if len(self.shape) == 2 else 0	6✔
586	result = self.array.sum(axis=axis)	6✔
587	template = DictArrayTemplate(self.template.names[0])	6✔
588	return template.wrap(result)	6✔
589
590	def col_sum(self):	6✔
591	"""returns DictArray summed across columns"""
592	result = self.array.sum(axis=0)	6✔
593	template = DictArrayTemplate(self.template.names[1])	6✔
594	return template.wrap(result)	6✔
595
596	def _repr_html_(self):	6✔
597	if self.array.ndim > 2:	6✔
598	return f"{self.array.ndim} dimensional {type(self).__name__}"	6✔
599
600	t = self.to_table()	6✔
601	t.set_repr_policy(show_shape=False)	6✔
602	return t._repr_html_()	6✔
603
604	@c3warn.deprecated_args(	6✔
605	"2025.9", "don't use built in name", old_new=[("format", "format_name")]
606	)
607	def to_string(self, format_name: str = "tsv", sep: str \| None = None) -> str:	6✔
608	"""Return the data as a formatted string.
609
610	Parameters
611	----------
612	format_name
613	possible formats are 'csv', or 'tsv' (default).
614	sep
615	A string separator for delineating columns, e.g. ',' or
616	'\t'. Overrides format.
617	"""
618	if format_name.lower() not in ("tsv", "csv"):	6✔
619	msg = f"'{format_name}' not supported"	6✔
620	raise ValueError(msg)	6✔
621
622	sep = sep or {"tsv": "\t", "csv": ","}[format_name.lower()]	6✔
623
624	data = self.to_dict(flatten=True)	6✔
625	rows = [[f"dim-{i + 1}" for i in range(self.array.ndim)] + ["value"]] + [	6✔
626	[str(x) for x in row] for row in [[*list(k), v] for k, v in data.items()]
627	]
628	return "\n".join([sep.join(row) for row in rows])	6✔
629
630	def to_table(self):	6✔
631	"""return Table instance
632
633	Notes
634	-----
635	Raises ValueError if number of dimensions > 2
636	"""
637	ndim = self.array.ndim	6✔
638	if ndim > 2:	6✔
639	msg = f"cannot make 2D table from {ndim}D array"	6✔
640	raise ValueError(msg)	6✔
641
642	from cogent3.core.table import Table	6✔
643
644	header = self.template.names[0] if ndim == 1 else self.template.names[1]	6✔
645	index = "" if ndim == 2 else None	6✔
646	if ndim == 1:	6✔
647	data = {c: [v] for c, v in zip(header, self.array, strict=False)}	6✔
648	else:
649	data = {c: self.array[:, i].tolist() for i, c in enumerate(header)}	6✔
650	data[""] = self.template.names[0]	6✔
651
652	return Table(header=header, data=data, index_name=index)	6✔
653
654	@c3warn.deprecated_args(	6✔
655	"2025.9", "don't use built in name", old_new=[("format", "format_name")]
656	)
657	def write(	6✔
658	self, path: str \| os.PathLike, format_name: str = "tsv", sep: str = "\t"
659	) -> None:
660	"""writes a flattened version to path
661
662	Parameters
663	----------
664	path
665	format_name
666	possible formats are 'rest'/'rst', 'markdown'/'md',
667	'latex', 'html', 'phylip', 'bedgraph', 'csv', 'tsv', or 'simple'
668	(default).
669	sep
670	used to split fields, will be inferred from path suffix if not
671	provided
672	"""
673	data = self.to_string(format_name=format_name, sep=sep)	6✔
674	with atomic_write(path, mode="wt") as outfile:	6✔
675	outfile.write(data)	6✔
676
677
678	@register_deserialiser(	6✔
679	get_object_provenance(DictArrayTemplate),
680	)
681	def deserialise_dict_array(data: dict) -> DictArray:	6✔
682	"""deserialising DictArray, Table instances"""
683	data.pop("version", None)	6✔
684	type_ = data.pop("type")	6✔
685	klass = get_class(type_)	6✔
686	named_dims = data.pop("names")	6✔
687	array = data.pop("array")	6✔
688	template = klass(*named_dims)	6✔
689	return template.wrap(array)	6✔

rmcar17 / cogent3 / 16431585733

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