10798992192

Committed 10 Sep 2024 07:03PM CUT coverage: 98.807% (-0.09%) from 98.892%

Build # 10798992192

Build Type

Pull #1044

github

Committed by

web-flow

Commit Message

Merge bc24d2bbb into ae617c7b3

Pull Request Pull Request #1044: test runner for basilisp.test #980

Run Details

1286 of 1294 branches covered (99.38%)

Branch coverage included in aggregate %.

8570 of 8681 relevant lines covered (98.72%)

0.99 hits per line

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

96.35

/src/basilisp/lang/vector.py

from functools import total_ordering
from typing import (
    TYPE_CHECKING,
    Iterable,
    Optional,
    Sequence,
    TypeVar,
    Union,
    cast,
    overload,
)

from pyrsistent import PVector, pvector  # noqa # pylint: disable=unused-import
from pyrsistent.typing import PVectorEvolver
from typing_extensions import Unpack

from basilisp.lang.interfaces import (
    IEvolveableCollection,
    ILispObject,
    IMapEntry,
    IPersistentMap,
    IPersistentVector,
    IReduce,
    IReduceKV,
    ISeq,
    ITransientVector,
    IWithMeta,
    ReduceFunction,
    ReduceKVFunction,
    seq_equals,
)
from basilisp.lang.obj import PrintSettings
from basilisp.lang.obj import seq_lrepr as _seq_lrepr
from basilisp.lang.reduced import Reduced
from basilisp.lang.seq import sequence
from basilisp.util import partition

if TYPE_CHECKING:
    from typing import Tuple

T = TypeVar("T")

T_reduce = TypeVar("T_reduce")
V_contra = TypeVar("V_contra", contravariant=True)


class TransientVector(ITransientVector[T]):
    __slots__ = ("_inner",)

    def __init__(self, wrapped: "PVectorEvolver[T]") -> None:
        self._inner = wrapped

    def __bool__(self):
        return True

    def __contains__(self, item):
        return item in self._inner

    def __eq__(self, other):
        return self is other

    def __len__(self):
        return len(self._inner)

    def cons_transient(self, *elems: T) -> "TransientVector[T]":  # type: ignore[override]
        for elem in elems:
            self._inner.append(elem)
        return self

    def assoc_transient(self, *kvs: T) -> "TransientVector[T]":
        for i, v in cast("Sequence[Tuple[int, T]]", partition(kvs, 2)):
            self._inner.set(i, v)
        return self

    def contains_transient(self, k: int) -> bool:
        return 0 <= k < len(self._inner)

    def entry_transient(self, k: int) -> Optional[IMapEntry[int, T]]:
        try:
            return MapEntry.of(k, self._inner[k])
        except IndexError:
            return None

    def val_at(self, k: int, default=None):
        try:
            return self._inner[k]
        except IndexError:
            return default

    def pop_transient(self) -> "TransientVector[T]":
        if len(self) == 0:
            raise IndexError("Cannot pop an empty vector")
        del self._inner[-1]
        return self

    def to_persistent(self) -> "PersistentVector[T]":
        return PersistentVector(self._inner.persistent())


@total_ordering
class PersistentVector(
    IReduce,
    IReduceKV,
    IPersistentVector[T],
    IEvolveableCollection[TransientVector],
    ILispObject,
    IWithMeta,
):
    """Basilisp Vector. Delegates internally to a pyrsistent.PVector object.
    Do not instantiate directly. Instead use the v() and vec() factory
    methods below."""

    __slots__ = ("_inner", "_meta")

    def __init__(
        self, wrapped: "PVector[T]", meta: Optional[IPersistentMap] = None
    ) -> None:
        self._inner = wrapped
        self._meta = meta

    def __bool__(self):
        return True

    def __contains__(self, item):
        return item in self._inner

    def __eq__(self, other):
        if self is other:
            return True
        if hasattr(other, "__len__") and len(self) != len(other):
            return False
        return seq_equals(self, other)

    def __getitem__(self, item):
        if isinstance(item, slice):
            return PersistentVector(self._inner[item])
        return self._inner[item]

    def __hash__(self):
        return hash(self._inner)

    def __iter__(self):
        yield from self._inner

    def __len__(self):
        return len(self._inner)

    def __call__(self, k: int, default: Optional[T] = None) -> Optional[T]:
        return self.val_at(k, default=default)

    def __lt__(self, other):
        """Return true if the `self` vector is shorter than the
        `other` vector, or the first unequal element in `self` when
        iterating from left to right is less than the corresponding
        `other` element.

        This is to support the comparing and sorting operations of
        vectors in Clojure."""

        if other is None:  # pragma: no cover
            return False
        if not isinstance(other, PersistentVector):
            return NotImplemented
        if len(self) != len(other):
            return len(self) < len(other)

        for x, y in zip(self, other):
            if x < y:
                return True
            elif y < x:
                return False
        return False

    def _lrepr(self, **kwargs: Unpack[PrintSettings]) -> str:
        return _seq_lrepr(self._inner, "[", "]", meta=self._meta, **kwargs)

    @property
    def meta(self) -> Optional[IPersistentMap]:
        return self._meta

    def with_meta(self, meta: Optional[IPersistentMap]) -> "PersistentVector[T]":
        return vector(self._inner, meta=meta)

    def cons(self, *elems: T) -> "PersistentVector[T]":  # type: ignore[override]
        e = self._inner.evolver()
        for elem in elems:
            e.append(elem)
        return PersistentVector(e.persistent(), meta=self.meta)

    def assoc(self, *kvs: T) -> "PersistentVector[T]":
        return PersistentVector(self._inner.mset(*kvs))  # type: ignore[arg-type]

    def contains(self, k: int) -> bool:
        return 0 <= k < len(self._inner)

    def entry(self, k: int) -> Optional[IMapEntry[int, T]]:
        try:
            return MapEntry.of(k, self._inner[k])
        except IndexError:
            return None

    def val_at(self, k: int, default: Optional[T] = None) -> Optional[T]:
        try:
            return self._inner[k]
        except IndexError:
            return default

    @staticmethod
    def empty() -> "PersistentVector[T]":
        return EMPTY

    def seq(self) -> Optional[ISeq[T]]:  # type: ignore[override]
        if len(self._inner) == 0:
            return None
        return sequence(self)

    def peek(self) -> Optional[T]:
        if len(self) == 0:
            return None
        return self[-1]

    def pop(self) -> "PersistentVector[T]":
        if len(self) == 0:
            raise IndexError("Cannot pop an empty vector")
        return self[:-1]

    def rseq(self) -> ISeq[T]:
        return sequence(reversed(self))

    def to_transient(self) -> TransientVector:
        return TransientVector(self._inner.evolver())

    @overload
    def reduce(self, f: ReduceFunction[T_reduce, V_contra]) -> T_reduce: ...

    @overload
    def reduce(  # pylint: disable=arguments-differ
        self, f: ReduceFunction[T_reduce, V_contra], init: T_reduce
    ) -> T_reduce: ...

    def reduce(self, f, init=IReduce.REDUCE_SENTINEL):
        if init is IReduce.REDUCE_SENTINEL:
            if len(self) == 0:
                return f()
            else:
                init = self._inner[0]
                for item in self._inner[1:]:
                    init = f(init, item)
                    if isinstance(init, Reduced):
                        return init.deref()
        else:
            for item in self._inner:
                init = f(init, item)
                if isinstance(init, Reduced):
                    return init.deref()
        return init

    def reduce_kv(self, f: ReduceKVFunction, init: T_reduce) -> T_reduce:
        for idx, item in enumerate(self._inner):
            init = f(init, idx, item)
            if isinstance(init, Reduced):
                return init.deref()
        return init


K = TypeVar("K")
V = TypeVar("V")


class MapEntry(IMapEntry[K, V], PersistentVector[Union[K, V]]):
    __slots__ = ()

    def __init__(self, wrapped: "PVector[Union[K, V]]") -> None:
        try:
            if not len(wrapped) == 2:
                raise ValueError("Vector arg to map conj must be a pair")
        except TypeError as e:
            raise TypeError(f"Cannot make map entry from {type(wrapped)}") from e

        super().__init__(wrapped)

    @property
    def key(self) -> K:
        return self[0]

    @property
    def value(self) -> V:
        return self[1]

    @staticmethod
    def of(k: K, v: V) -> "MapEntry[K, V]":
        return MapEntry(pvector([k, v]))

    @staticmethod
    def from_vec(v: Sequence[Union[K, V]]) -> "MapEntry[K, V]":
        return MapEntry(pvector(v))


EMPTY: PersistentVector = PersistentVector(pvector(()))


def vector(
    members: Iterable[T], meta: Optional[IPersistentMap] = None
) -> PersistentVector[T]:
    """Creates a new vector."""
    return PersistentVector(pvector(members), meta=meta)


def v(*members: T, meta: Optional[IPersistentMap] = None) -> PersistentVector[T]:
    """Creates a new vector from members."""
    return PersistentVector(pvector(members), meta=meta)

1	from functools import total_ordering	1✔
2	from typing import (	1✔
3	TYPE_CHECKING,
4	Iterable,
5	Optional,
6	Sequence,
7	TypeVar,
8	Union,
9	cast,
10	overload,
11	)
12
13	from pyrsistent import PVector, pvector # noqa # pylint: disable=unused-import	1✔
14	from pyrsistent.typing import PVectorEvolver	1✔
15	from typing_extensions import Unpack	1✔
16
17	from basilisp.lang.interfaces import (	1✔
18	IEvolveableCollection,
19	ILispObject,
20	IMapEntry,
21	IPersistentMap,
22	IPersistentVector,
23	IReduce,
24	IReduceKV,
25	ISeq,
26	ITransientVector,
27	IWithMeta,
28	ReduceFunction,
29	ReduceKVFunction,
30	seq_equals,
31	)
32	from basilisp.lang.obj import PrintSettings	1✔
33	from basilisp.lang.obj import seq_lrepr as _seq_lrepr	1✔
34	from basilisp.lang.reduced import Reduced	1✔
35	from basilisp.lang.seq import sequence	1✔
36	from basilisp.util import partition	1✔
37
38	if TYPE_CHECKING:	1✔
39	from typing import Tuple	×
40
41	T = TypeVar("T")	1✔
42
43	T_reduce = TypeVar("T_reduce")	1✔
44	V_contra = TypeVar("V_contra", contravariant=True)	1✔
45
46
47	class TransientVector(ITransientVector[T]):	1✔
48	__slots__ = ("_inner",)	1✔
49
50	def __init__(self, wrapped: "PVectorEvolver[T]") -> None:	1✔
51	self._inner = wrapped	1✔
52
53	def __bool__(self):	1✔
54	return True	×
55
56	def __contains__(self, item):	1✔
57	return item in self._inner	×
58
59	def __eq__(self, other):	1✔
60	return self is other	1✔
61
62	def __len__(self):	1✔
63	return len(self._inner)	1✔
64
65	def cons_transient(self, *elems: T) -> "TransientVector[T]": # type: ignore[override]	1✔
66	for elem in elems:	1✔
67	self._inner.append(elem)	1✔
68	return self	1✔
69
70	def assoc_transient(self, *kvs: T) -> "TransientVector[T]":	1✔
71	for i, v in cast("Sequence[Tuple[int, T]]", partition(kvs, 2)):	1✔
72	self._inner.set(i, v)	1✔
73	return self	1✔
74
75	def contains_transient(self, k: int) -> bool:	1✔
76	return 0 <= k < len(self._inner)	1✔
77
78	def entry_transient(self, k: int) -> Optional[IMapEntry[int, T]]:	1✔
79	try:	×
80	return MapEntry.of(k, self._inner[k])	×
81	except IndexError:	×
82	return None	×
83
84	def val_at(self, k: int, default=None):	1✔
85	try:	1✔
86	return self._inner[k]	1✔
87	except IndexError:	1✔
88	return default	1✔
89
90	def pop_transient(self) -> "TransientVector[T]":	1✔
91	if len(self) == 0:	1✔
92	raise IndexError("Cannot pop an empty vector")	1✔
93	del self._inner[-1]	1✔
94	return self	1✔
95
96	def to_persistent(self) -> "PersistentVector[T]":	1✔
97	return PersistentVector(self._inner.persistent())	1✔
98
99
100	@total_ordering	1✔
101	class PersistentVector(	1✔
102	IReduce,
103	IReduceKV,
104	IPersistentVector[T],
105	IEvolveableCollection[TransientVector],
106	ILispObject,
107	IWithMeta,
108	):
109	"""Basilisp Vector. Delegates internally to a pyrsistent.PVector object.
110	Do not instantiate directly. Instead use the v() and vec() factory
111	methods below."""
112
113	__slots__ = ("_inner", "_meta")	1✔
114
115	def __init__(	1✔
116	self, wrapped: "PVector[T]", meta: Optional[IPersistentMap] = None
117	) -> None:
118	self._inner = wrapped	1✔
119	self._meta = meta	1✔
120
121	def __bool__(self):	1✔
122	return True	1✔
123
124	def __contains__(self, item):	1✔
125	return item in self._inner	1✔
126
127	def __eq__(self, other):	1✔
128	if self is other:	1✔
129	return True	1✔
130	if hasattr(other, "__len__") and len(self) != len(other):	1✔
131	return False	1✔
132	return seq_equals(self, other)	1✔
133
134	def __getitem__(self, item):	1✔
135	if isinstance(item, slice):	1✔
136	return PersistentVector(self._inner[item])	1✔
137	return self._inner[item]	1✔
138
139	def __hash__(self):	1✔
140	return hash(self._inner)	1✔
141
142	def __iter__(self):	1✔
143	yield from self._inner	1✔
144
145	def __len__(self):	1✔
146	return len(self._inner)	1✔
147
148	def __call__(self, k: int, default: Optional[T] = None) -> Optional[T]:	1✔
149	return self.val_at(k, default=default)	1✔
150
151	def __lt__(self, other):	1✔
152	"""Return true if the `self` vector is shorter than the
153	`other` vector, or the first unequal element in `self` when
154	iterating from left to right is less than the corresponding
155	`other` element.
156
157	This is to support the comparing and sorting operations of
158	vectors in Clojure."""
159
160	if other is None: # pragma: no cover
161	return False
162	if not isinstance(other, PersistentVector):	1✔
163	return NotImplemented	1✔
164	if len(self) != len(other):	1✔
165	return len(self) < len(other)	1✔
166
167	for x, y in zip(self, other):	1✔
168	if x < y:	1✔
169	return True	1✔
170	elif y < x:	1✔
171	return False	1✔
172	return False	1✔
173
174	def _lrepr(self, **kwargs: Unpack[PrintSettings]) -> str:	1✔
175	return _seq_lrepr(self._inner, "[", "]", meta=self._meta, **kwargs)	1✔
176
177	@property	1✔
178	def meta(self) -> Optional[IPersistentMap]:	1✔
179	return self._meta	1✔
180
181	def with_meta(self, meta: Optional[IPersistentMap]) -> "PersistentVector[T]":	1✔
182	return vector(self._inner, meta=meta)	1✔
183
184	def cons(self, *elems: T) -> "PersistentVector[T]": # type: ignore[override]	1✔
185	e = self._inner.evolver()	1✔
186	for elem in elems:	1✔
187	e.append(elem)	1✔
188	return PersistentVector(e.persistent(), meta=self.meta)	1✔
189
190	def assoc(self, *kvs: T) -> "PersistentVector[T]":	1✔
191	return PersistentVector(self._inner.mset(*kvs)) # type: ignore[arg-type]	1✔
192
193	def contains(self, k: int) -> bool:	1✔
194	return 0 <= k < len(self._inner)	1✔
195
196	def entry(self, k: int) -> Optional[IMapEntry[int, T]]:	1✔
197	try:	1✔
198	return MapEntry.of(k, self._inner[k])	1✔
199	except IndexError:	1✔
200	return None	1✔
201
202	def val_at(self, k: int, default: Optional[T] = None) -> Optional[T]:	1✔
203	try:	1✔
204	return self._inner[k]	1✔
205	except IndexError:	1✔
206	return default	1✔
207
208	@staticmethod	1✔
209	def empty() -> "PersistentVector[T]":	1✔
210	return EMPTY	1✔
211
212	def seq(self) -> Optional[ISeq[T]]: # type: ignore[override]	1✔
213	if len(self._inner) == 0:	1✔
214	return None	1✔
215	return sequence(self)	1✔
216
217	def peek(self) -> Optional[T]:	1✔
218	if len(self) == 0:	1✔
219	return None	1✔
220	return self[-1]	1✔
221
222	def pop(self) -> "PersistentVector[T]":	1✔
223	if len(self) == 0:	1✔
224	raise IndexError("Cannot pop an empty vector")	1✔
225	return self[:-1]	1✔
226
227	def rseq(self) -> ISeq[T]:	1✔
228	return sequence(reversed(self))	1✔
229
230	def to_transient(self) -> TransientVector:	1✔
231	return TransientVector(self._inner.evolver())	1✔
232
233	@overload	1✔
234	def reduce(self, f: ReduceFunction[T_reduce, V_contra]) -> T_reduce: ...	1✔
235
236	@overload	1✔
237	def reduce( # pylint: disable=arguments-differ	1✔
238	self, f: ReduceFunction[T_reduce, V_contra], init: T_reduce
239	) -> T_reduce: ...
240
241	def reduce(self, f, init=IReduce.REDUCE_SENTINEL):	1✔
242	if init is IReduce.REDUCE_SENTINEL:	1✔
243	if len(self) == 0:	1✔
244	return f()	1✔
245	else:
246	init = self._inner[0]	1✔
247	for item in self._inner[1:]:	1✔
248	init = f(init, item)	1✔
249	if isinstance(init, Reduced):	1✔
250	return init.deref()	1✔
251	else:
252	for item in self._inner:	1✔
253	init = f(init, item)	1✔
254	if isinstance(init, Reduced):	1✔
255	return init.deref()	1✔
256	return init	1✔
257
258	def reduce_kv(self, f: ReduceKVFunction, init: T_reduce) -> T_reduce:	1✔
259	for idx, item in enumerate(self._inner):	1✔
260	init = f(init, idx, item)	1✔
261	if isinstance(init, Reduced):	1✔
262	return init.deref()	1✔
263	return init	1✔
264
265
266	K = TypeVar("K")	1✔
267	V = TypeVar("V")	1✔
268
269
270	class MapEntry(IMapEntry[K, V], PersistentVector[Union[K, V]]):	1✔
271	__slots__ = ()	1✔
272
273	def __init__(self, wrapped: "PVector[Union[K, V]]") -> None:	1✔
274	try:	1✔
275	if not len(wrapped) == 2:	1✔
276	raise ValueError("Vector arg to map conj must be a pair")	1✔
277	except TypeError as e:	1✔
278	raise TypeError(f"Cannot make map entry from {type(wrapped)}") from e	×
279
280	super().__init__(wrapped)	1✔
281
282	@property	1✔
283	def key(self) -> K:	1✔
284	return self[0]	1✔
285
286	@property	1✔
287	def value(self) -> V:	1✔
288	return self[1]	1✔
289
290	@staticmethod	1✔
291	def of(k: K, v: V) -> "MapEntry[K, V]":	1✔
292	return MapEntry(pvector([k, v]))	1✔
293
294	@staticmethod	1✔
295	def from_vec(v: Sequence[Union[K, V]]) -> "MapEntry[K, V]":	1✔
296	return MapEntry(pvector(v))	1✔
297
298
299	EMPTY: PersistentVector = PersistentVector(pvector(()))	1✔
300
301
302	def vector(	1✔
303	members: Iterable[T], meta: Optional[IPersistentMap] = None
304	) -> PersistentVector[T]:
305	"""Creates a new vector."""
306	return PersistentVector(pvector(members), meta=meta)	1✔
307
308
309	def v(*members: T, meta: Optional[IPersistentMap] = None) -> PersistentVector[T]:	1✔
310	"""Creates a new vector from members."""
311	return PersistentVector(pvector(members), meta=meta)	1✔

basilisp-lang / basilisp / 10798992192

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