just new types (#2358)

This commit is contained in:
Friedrich Carl Eichenroth
2023-11-18 23:29:46 -08:00
committed by GitHub
parent f203d37258
commit b3a21eee7d
3 changed files with 19 additions and 19 deletions
+1 -1
View File
@@ -194,7 +194,7 @@ class Kernel:
assert len(colors) == self.shape_len, "colors size mismatch"
return colors
def colored_shape(self, pad=None, dense=False) -> str:
def colored_shape(self, pad:Optional[int]=None, dense=False) -> str:
ret = ' '.join(colored(s, color) for s,color in zip([f"{s:4d}" if isinstance(s, int) and not dense else s for s in self.full_shape], self.colors()))
if pad: ret += ' '*(pad-ansilen(ret))
return ret
+8 -8
View File
@@ -1,7 +1,7 @@
from __future__ import annotations
import os, functools, platform, time, re, contextlib, operator, hashlib, pickle, sqlite3, cProfile, pstats
import numpy as np
from typing import Dict, Tuple, Union, List, NamedTuple, Final, Iterator, ClassVar, Optional, Iterable, Any, TypeVar, TYPE_CHECKING, Callable
from typing import Dict, Tuple, Union, List, NamedTuple, Final, ClassVar, Optional, Iterable, Any, TypeVar, TYPE_CHECKING, Callable
if TYPE_CHECKING: # TODO: remove this and import TypeGuard from typing once minimum python supported version is 3.10
from typing_extensions import TypeGuard
@@ -14,18 +14,18 @@ def prod(x:Iterable[T]) -> Union[T,int]: return functools.reduce(operator.__mul_
OSX = platform.system() == "Darwin"
CI = os.getenv("CI", "") != ""
def dedup(x): return list(dict.fromkeys(x)) # retains list order
def dedup(x:Iterable[T]): return list(dict.fromkeys(x)) # retains list order
def argfix(*x): return tuple(x[0]) if x and x[0].__class__ in (tuple, list) else x
def argsort(x): return type(x)(sorted(range(len(x)), key=x.__getitem__)) # https://stackoverflow.com/questions/3382352/equivalent-of-numpy-argsort-in-basic-python
def all_same(items): return all(x == items[0] for x in items)
def all_same(items:List[T]): return all(x == items[0] for x in items)
def all_int(t: Tuple[Any, ...]) -> TypeGuard[Tuple[int, ...]]: return all(isinstance(s, int) for s in t)
def colored(st, color, background=False): return f"\u001b[{10*background+60*(color.upper() == color)+30+['black', 'red', 'green', 'yellow', 'blue', 'magenta', 'cyan', 'white'].index(color.lower())}m{st}\u001b[0m" if color is not None else st # replace the termcolor library with one line
def ansistrip(s): return re.sub('\x1b\\[(K|.*?m)', '', s)
def ansilen(s): return len(ansistrip(s))
def colored(st, color:Optional[str], background=False): return f"\u001b[{10*background+60*(color.upper() == color)+30+['black', 'red', 'green', 'yellow', 'blue', 'magenta', 'cyan', 'white'].index(color.lower())}m{st}\u001b[0m" if color is not None else st # replace the termcolor library with one line
def ansistrip(s:str): return re.sub('\x1b\\[(K|.*?m)', '', s)
def ansilen(s:str): return len(ansistrip(s))
def make_pair(x:Union[int, Tuple[int, ...]], cnt=2) -> Tuple[int, ...]: return (x,)*cnt if isinstance(x, int) else x
def flatten(l:Union[List, Iterator]): return [item for sublist in l for item in sublist]
def flatten(l:Iterable[Iterable[T]]): return [item for sublist in l for item in sublist]
def fromimport(mod, frm): return getattr(__import__(mod, fromlist=[frm]), frm)
def strip_parens(fst): return fst[1:-1] if fst[0] == '(' and fst[-1] == ')' and fst[1:-1].find('(') <= fst[1:-1].find(')') else fst
def strip_parens(fst:str): return fst[1:-1] if fst[0] == '(' and fst[-1] == ')' and fst[1:-1].find('(') <= fst[1:-1].find(')') else fst
def round_up(num, amt): return num if num%amt == 0 else num+(amt-(num%amt))
def merge_dicts(ds:Iterable[Dict[T,U]]) -> Dict[T,U]:
assert len(kvs:=set([(k,v) for d in ds for k,v in d.items()])) == len(set(kv[0] for kv in kvs)), f"cannot merge, {kvs} contains different values for the same key"
+10 -10
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
import functools, operator
from dataclasses import dataclass
from typing import Tuple, List, Optional, Dict, Set, cast
from typing import Tuple, List, Optional, Dict, Set, cast, Union, Iterable
from tinygrad.ops import MovementOps
from tinygrad.helpers import prod, DEBUG, merge_dicts
from tinygrad.shape.symbolic import Variable, MulNode, Node, SumNode, NumNode, sint
@@ -21,7 +21,7 @@ def to_shape_strides(shape:Tuple[int, ...], strides:Tuple[int, ...]) -> Tuple[Tu
ret.append((shape[i], strides[i]))
return tuple(ret)
def expr_node_mask(view:View, idx, valid=None) -> Node:
def expr_node_mask(view:View, idx:Node, valid:Optional[Node]=None) -> Node:
expr = [valid] if valid is not None else []
if view.mask is not None:
acc = 1
@@ -33,7 +33,7 @@ def expr_node_mask(view:View, idx, valid=None) -> Node:
return Variable.ands(expr)
# generate an expression if you have a single idx variable
def expr_node(view:View, idx=None) -> Node:
def expr_node(view:View, idx:Optional[Node]=None) -> Node:
if idx is None: idx = Variable('idx', 0, prod(view.shape)-1)
ret: List[Node] = [NumNode(view.offset) if isinstance(view.offset, int) else view.offset] if view.offset else []
acc = 1
@@ -43,7 +43,7 @@ def expr_node(view:View, idx=None) -> Node:
return Variable.sum(ret)
# generate an expression if you have a variable or expression for each index
def expr_idxs(view:View, idxs) -> Node:
def expr_idxs(view:View, idxs:Tuple[Node, ...]) -> Node:
assert len(idxs) == len(view.shape), f"need an idx for all dimensions {idxs} vs {view.shape}"
return Variable.sum([NumNode(view.offset) if isinstance(view.offset, int) else view.offset] + [idx*st for idx,sh,st in zip(idxs, view.shape, view.strides) if sh != 1 and st != 0])
@@ -54,7 +54,7 @@ def merge_views(vm2:View, vm1:View) -> Optional[View]:
return View.create(vm1.shape, cast(Tuple[sint, ...], strides), vm2.offset, vm1.mask)
@functools.lru_cache(maxsize=None)
def idxs_to_idx(shape:Tuple[int, ...], idxs) -> Node:
def idxs_to_idx(shape:Tuple[int, ...], idxs:Tuple[Node, ...]) -> Node:
assert len(idxs) == len(shape), "need an idx for all dimensions"
acc = 1
ret = []
@@ -127,7 +127,7 @@ class ShapeTracker:
def unit_stride_axes(self, ignore_valid=False) -> List[int]: return [i for i,st in enumerate(self.real_strides(ignore_valid)) if st == 1]
def _expr_idx(self, idx, valid) -> Tuple[Node, Node]:
def _expr_idx(self, idx:Node, valid:Node) -> Tuple[Node, Node]:
for v in reversed(self.views[0:-1]):
if valid.max == 0: return NumNode(-1), valid
valid = expr_node_mask(v, idx, valid)
@@ -141,17 +141,17 @@ class ShapeTracker:
return ShapeTracker(self.views[:-2] + (new_view,)).simplify()
return self
def expr_idxs(self, idxs=None):
def expr_idxs(self, idxs:Optional[Iterable[Node]]=None):
if idxs is None: idxs = [Variable(f"idx{i}", 0, s-1) for i,s in enumerate(self.shape)]
idx = expr_idxs(self.views[-1], tuple(idxs))
valid = expr_node_mask(self.views[-1], idxs_to_idx(self.views[-1].shape, tuple(idxs)))
return self._expr_idx(idx, valid)
def expr_node(self, idx='idx'):
if idx.__class__ is str: idx = Variable(idx, 0, prod(self.shape)-1)
def expr_node(self, idx:Union[Node,str]='idx'):
if isinstance(idx, str): idx = Variable(idx, 0, prod(self.shape)-1)
return self._expr_idx(expr_node(self.views[-1], idx), expr_node_mask(self.views[-1], idx))
def axis_is_masked(self, axis) -> bool:
def axis_is_masked(self, axis:int) -> bool:
_, valid = self.expr_idxs()
return f'idx{axis}' in [v.expr for v in valid.vars()]