<jupyter_start><jupyter_text>Data block API from config class  Core helper functions<jupyter_code># export
def noop(x): return x
# test
assert noop(1) == 1
# export
def listify(o):
    "Make `o` a list."
    if o is None: return []
    if isinstance(o, list): return o
    if isinstance(o, str): return [o]
    if isinstance(o, Iterable): return list(o)
    return [o]
# test
assert listify(None) == []
assert listify([1,2,3]) == [1,2,3]
assert listify('abc') == ['abc']
assert listify(range(0,3)) == [0,1,2]
# export
def compose(x, funcs, *args, order_key='_order', **kwargs):
    "Apply all `funcs` to `x` in order, pass along `args` and `kwargs`."
    key = lambda o: getattr(o, order_key, 0)
    for f in sorted(listify(funcs), key=key): x = f(x, *args, **kwargs)
    return x
# test
# basic behavior
def _test_f1(x, a=2): return x**a
def _test_f2(x, a=2): return a*x
assert compose(2, [_test_f1, _test_f2]) == 8
# order
_test_f1._order = 1
assert compose(2, [_test_f1, _test_f2]) == 16
#args
assert compose(2, [_test_f1, _test_f2], 3) == 216
#kwargs
assert compose(2, [_test_f1, _test_f2], a=3) == 216
# export
def uniqueify(x, sort=False):
    "Return the unqiue elements in `x`, optionally `sort`-ed."
    res = list(OrderedDict.fromkeys(x).keys())
    if sort: res.sort()
    return res
# test
assert set(uniqueify([1,1,0,5,0,3])) == {0,1,3,5}
assert uniqueify([1,1,0,5,0,3], sort=True) == [0,1,3,5]
# export
def setify(o): return o if isinstance(o,set) else set(listify(o))
# test
assert setify(None) == set()
assert setify('abc') == {'abc'}
assert setify([1,2,2]) == {1,2}
assert setify(range(0,3)) == {0,1,2}
assert setify({1,2}) == {1,2}
# export
def _get_files(p, fs, extensions=None):
    p = Path(p)
    res = [p/f for f in fs if not f.startswith('.')
           and ((not extensions) or f'.{f.split(".")[-1].lower()}' in extensions)]
    return res

def get_files(path, extensions=None, recurse=False, include=None):
    "Get all the files in `path` with optional `extensions`."
    path = Path(path)
    extensions = setify(extensions)
    extensions = {e.lower() for e in extensions}
    if recurse:
        res = []
        for i,(p,d,f) in enumerate(os.walk(path)): # returns (dirpath, dirnames, filenames)
            if include is not None and i==0: d[:] = [o for o in d if o in include]
            else:                            d[:] = [o for o in d if not o.startswith('.')]
            res += _get_files(p, f, extensions)
        return res
    else:
        f = [o.name for o in os.scandir(path) if o.is_file()]
        return _get_files(path, f, extensions)
# test
path = untar_data(URLs.MNIST_TINY)
assert len(get_files(path/'train'/'3')) == 346
assert len(get_files(path/'train'/'3', extensions='.png')) == 346
assert len(get_files(path/'train'/'3', extensions='.jpg')) == 0
assert len(get_files(path/'train', extensions='.png')) == 0
assert len(get_files(path/'train', extensions='.png', recurse=True)) == 709
assert len(get_files(path, extensions='.png', recurse=True, include=['train'])) == 709
assert len(get_files(path, extensions='.png', recurse=True, include=['train', 'test'])) == 729
# export
def grab_idx(batch, i):
    "Return the `i`-th sample in `batch`"
    return [grab_idx(b,i) for b in batch] if isinstance(batch, list) else batch[i]
# test
assert grab_idx(tensor([1,2]), 1) == 2
assert grab_idx([tensor([1,2]), tensor([3,4])], 1) == [2,4]<jupyter_output><empty_output><jupyter_text>Processor and ItemGetter<jupyter_code># export
class Processor():
    "A basic class to preprocess some data."
    def __call__(self, items): return items
    def process1(self, item):  return item
    def deproc1(self, item):   return item
    def deproc(self, items):   return items
# export
TfmY = Enum('TfmY', 'No Mask Image Point Bbox')
# export
class ItemGetter():
    "A basic class for representing some data type."
    #Subclass for default processor associated to this data type (example: CategoryGetter has CategoryProcessor)
    default_proc = None
    #Subclass for default `tfm_y` value associated to this data type (example: SegmentMaskGetter has TfmY.Mask)
    default_tfm = TfmY.No
    
    def __init__(self, procs=None): 
        "Initialize with `default_proc` if no `procs` are passed."
        self.procs = [p() for p in listify(self.default_proc)] if procs is None else procs
    def __call__(self, items): 
        "Process data when called on it."
        return compose(items, self.procs)
    
    def get(self, o): 
        "How to get the actual item from `o` (example: fn -> Image, idxs -> 1hot encoded tensor)."
        return o  
    def raw(self, o): 
        "Undoes get when needed for representation (1hot encoded tensor -> idxs)."
        return o  
    
    def show(self, x, ax):
        "How to show one element `x` on `ax`."
        raise NotImplementedError
    def show_xys(self, xs, ys, y_get, **kwargs): 
        "How to organize the show of multiple `xs` and `ys`"
        raise NotImplementedError
# test
class FakeProc1(Processor): 
    def __call__(self, items):  return [a ** 2 for a in items]
class FakeProc2(Processor): 
    def __call__(self, items):  return [a + 1  for a in items]
    def deproc1(self, item):    return item-1
    
class FakeItemGetter(ItemGetter):
    default_proc = FakeProc1
    
    def get(self, o): return o*2
    def raw(self, o): return o/2

#Basics
items = [0,1,2,3,4]
get1 = FakeItemGetter()
assert len(get1.procs) == 1
assert isinstance(get1.procs[0], FakeProc1)
assert get1(items) == [0,1,4,9,16]
assert get1.get(items[1]) == 2
assert get1.raw(4) == 2

#Passing a procs override the default.
get2 = FakeItemGetter(procs=FakeProc2())
assert isinstance(get2.procs, FakeProc2)
assert get2(items) == [1,2,3,4,5]<jupyter_output><empty_output><jupyter_text>Data block API core One ItemList to contain them all<jupyter_code># export
class ItemList():
    def __init__(self, items, item_get=None):
        self.item_get = ItemGetter() if item_get is None else item_get
        self.items = self.item_get(listify(items))
    def _get(self, i): return self.item_get.get(i)
    def __getitem__(self, idx):
        try: return self._get(self.items[idx])
        except TypeError:
            if isinstance(idx[0],bool):
                assert len(idx)==len(self) # bool mask
                return [self._get(o) for m,o in zip(idx,self.items) if m]
            return [self._get(self.items[i]) for i in idx]
    def __len__(self): return len(self.items)
    def __iter__(self): return iter(self.items)
    def __setitem__(self, i, o): self.items[i] = o
    def __delitem__(self, i): del(self.items[i])
    def __repr__(self):
        res = f'{self.__class__.__name__} ({len(self)} items)\n{self.items[:10]}'
        if len(self)>10: res = res[:-1]+ '...]'
        return res
    
    def deproc(self, item):
        "Calls `raw` on item then `deproc1` for all processors."
        item = self.item_get.raw(item)
        for proc in reversed(listify(self.item_get.procs)):
            item = proc.deproc1(item)
        return item
    
    def obj(self, idx):
        isint = isinstance(idx, int) or (isinstance(idx,torch.LongTensor) and not idx.ndim)
        item = self[idx]
        return self.deproc(item) if isint else [self.deproc(o) for o in item]
# test
il = ItemList([0,1,2,3,4], get1)
#Items are processed
assert il.items == [0,1,4,9,16]
#Test get and indexing
assert il[2] == 8
assert il[[1,2]] == [2,8]
assert il[[True,False,False,True,False]] == [0,18]
#Test obj
assert il.obj(2) == 4
assert il.obj([1,2]) == [1,4]
assert il.obj([True,False,False,True,False]) == [0,9]
#Test obj with a processor that implements deproc
il = ItemList([0,1,2,3,4], get2)
assert il.items == [1,2,3,4,5]
assert il.obj(2) == 2
assert il.obj([1,2]) == [1,2]
assert il.obj([True,False,False,True,False]) == [0,3]<jupyter_output><empty_output><jupyter_text>Sometimes, the get function for our `ys` need to know the `x`: for instance, points need to be rescaled during the `get` method so they need to know the size of the image. Same for bounding boxes. The following context manager adds that information to `ItemList.item_get`.<jupyter_code># export
class AddXContext():
    "Context manager that adds `x` to `il.item_get`."
    def __init__(self, il, x): self.il,self.x = il,x 
    def __enter__(self, *args):
        self.il.item_get._x = self.x
        return self.il
    def __exit__(self, *args): self.il.item_get._x = None
# test
class FakeYItemGetter(ItemGetter):
    default_proc = FakeProc2
    
    def get(self, o): return o + self._x
    def raw(self, o): return o - self._x
    
get2 = FakeYItemGetter()
ily = ItemList([0,1,2,3,4], get2)
with AddXContext(ily, 5):
    assert ily[1] == 7
    assert ily.obj(1) == 1<jupyter_output><empty_output><jupyter_text>We use `AddXContext` when requesting the `ys` in `getitem` and `deproc`.<jupyter_code># export
class LabeledData():
    def __init__(self, x, y, tfms=None, tfm_y=TfmY.No):  
        self.x,self.y,self.tfms,self.tfm_y = x,y,tfms,tfm_y
    def __repr__(self):        return f'{self.__class__.__name__}\nx: {self.x}\ny: {self.y}\n'
    def __getitem__(self,idx):
        #TODO? Make below works if idx is a mask/array/list
        x = self.x[idx]
        with AddXContext(self.y, x) as yil: y = yil[idx] 
        return compose((x,y), self.tfms, tfm_y=self.tfm_y)
    def __len__(self):         return len(self.x)
    
    def deproc(self, o):
        "Calls deproc on the x and y of `o`."
        x = self.x.deproc(o[0])
        with AddXContext(self.y, x) as yil: y = yil.deproc(o[1])
        return (x,y)
# test
getx = FakeItemGetter(procs=FakeProc2())
ilx = ItemList([0,1,2,3,4], getx)
ll = LabeledData(ilx, ily)

assert ll[2] == ((2+1)*2, (2+1) + (2+1)*2)
assert ll.deproc((10,10)) == (10/2 - 1, 10-1 - (10/2-1))<jupyter_output><empty_output><jupyter_text>Dataloader and DataBunch<jupyter_code># export
from torch.utils.data.dataloader import DataLoader
def get_dl(ds, bs, shuffle=False, drop_last=False, **kwargs):
    "Basic function to get a `DataLoader`"
    return DataLoader(ds, batch_size=bs, shuffle=shuffle, drop_last=drop_last, **kwargs)
# export
class DataBunch():
    "Basic wrapper around several `DataLoader`."
    def __init__(self, train_dl, valid_dl):
        self.train_dl,self.valid_dl = train_dl,valid_dl
        
    @property
    def train_ds(self): return self.train_dl.dataset
    @property
    def valid_ds(self): return self.valid_dl.dataset
    
    def show_batch(self, is_valid=False, items=9, **kwargs):
        "Show `items` element of a batch, depending on `is_valid` for the dl it draw from."
        xb,yb = next(iter(self.valid_dl if is_valid else self.train_dl))
        xs,ys = [],[]
        for i in range(items):
            x,y = self.train_ds.deproc((grab_idx(xb, i), grab_idx(yb, i)))
            xs.append(x); ys.append(y)
        self.train_ds.x.item_get.show_xys(xs, ys, self.train_ds.y.item_get, **kwargs)<jupyter_output><empty_output><jupyter_text>Main class to represent any kind of data. User provides the `get_x_cls` and `get_y_cls` then the four functions. At init everything is constructed with optionals transforms or custom instances of `get_x`/`get_y`.<jupyter_code># export
class DataBlock():
    "Main class to represent a dataset. Subclass the 2 properties and 4 methods below to your need."
    get_x_cls = ItemGetter #Type of input
    get_y_cls = ItemGetter #Type of targer
    def get_source(self):         
        "Return the source of your data (path, dataframe...), optionally download it."
        raise NotImplementedError
    def get_items(self, source):  
        "Use `source` to return the list of all items."
        raise NotImplementedError
    def split(self, items):       
        "Explain how so split the `items`. Return two disjoint lists of indices or two boolean masks."
        raise NotImplementedError
    def label(self, items):       
        "Explain how to label your `items`. Return a list of labels."
        raise NotImplementedError
        
    def __init__(self, tfms=None, tfm_y=None, get_x=None, get_y=None):
        self.source = self.get_source()
        items = ItemList(self.get_items(self.source)) #Just for fancy indexing
        split_idx = self.split(items)
        labels = ItemList(self.label(items))          #Just for fancy indexing
        if get_x is None: get_x = self.get_x_cls()
        if get_y is None: get_y = self.get_y_cls()
        x_train,x_valid = map(lambda o: ItemList(items[o],  item_get=get_x), split_idx)
        y_train,y_valid = map(lambda o: ItemList(labels[o], item_get=get_y), split_idx)
        if tfm_y is None: tfm_y = get_y.default_tfm
        self.train = LabeledData(x_train, y_train, tfms=tfms, tfm_y=tfm_y)
        self.valid = LabeledData(x_valid, y_valid, tfms=tfms, tfm_y=tfm_y)
    
    def databunch(self, bs=64, **kwargs):
        "How to convert to a `DataBunch`. Subclass if needed."
        dls = [get_dl(ds, bs, shuffle=s, drop_last=s, **kwargs) for (ds, s) in zip([self.train, self.valid], [True,False])]
        return DataBunch(*dls)<jupyter_output><empty_output><jupyter_text>First ItemGetters How to get images<jupyter_code># export
class ImageGetter(ItemGetter):
    "An `ItemGetter` for image types."
    default_tfm = TfmY.Image
    def __init__(self, procs=None, cmap=None, alpha=1.): 
        super().__init__(procs)
        self.cmap,self.alpha = cmap,alpha
    def get(self, fn): return PIL.Image.open(fn)
    def show(self, x, ax):
        ax.imshow(x[0] if x.shape[0] == 1 else x.permute(1,2,0), cmap=self.cmap, alpha=self.alpha)
        ax.axis('off')
    def show_xys(self, xs, ys, y_get, cols=3, figsize=None):
        rows = (len(xs) // cols if len(xs)%cols == 0 else len(xs)//cols+1) 
        if figsize is None: figsize = (cols*3, rows*3)
        fig,axs = plt.subplots(rows, cols, figsize=figsize)
        for x,y,ax in zip(xs, ys, axs.flatten()):
            self.show(x, ax)
            y_get.show(y, ax)<jupyter_output><empty_output><jupyter_text>How to get categories<jupyter_code># export
class CategoryProcessor(Processor):
    "A `Processor` for categories."
    def __init__(self): self.vocab=None
    
    def __call__(self, items):
        #The vocab is defined on the first use.
        if self.vocab is None: self.create_vocab(items)
        return [self.proc1(o) for o in items]
    def proc1(self, item):  return self.otoi[item]
    
    def deprocess(self, idxs):
        assert self.vocab is not None
        return [self.deproc1(idx) for idx in idxs]
    def deproc1(self, idx): return self.vocab[idx]
    
    def create_vocab(self, items):
        self.vocab = uniqueify(items, sort=True)
        self.otoi  = {v:k for k,v in enumerate(self.vocab)}
# test
proc = CategoryProcessor()
#Even if 'dog' is the first class, vocab is sorted for reproducibility
assert proc(['dog', 'cat', 'cat', 'dog', 'cat', 'dog']) == [1,0,0,1,0,1]
assert proc(['cat', 'cat', 'dog']) == [0,0,1]
assert proc.vocab == ['cat', 'dog']
assert proc.deprocess([1,0,1]) == ['dog', 'cat', 'dog']
assert proc.proc1('cat') == 0
assert proc.deproc1(1) == 'dog'
# export
class CategoryGetter(ItemGetter):
    "An `ItemGetter` suitable for single-label classification targets"
    default_proc = CategoryProcessor
    def show(self, x, ax): ax.set_title(x)<jupyter_output><empty_output><jupyter_text>Helper functions To get image files<jupyter_code># export
image_extensions = set(k for k,v in mimetypes.types_map.items() if v.startswith('image/'))
# export
def get_image_files(path, include=None):
    return get_files(path, extensions=image_extensions, recurse=True, include=include)
# test
path = untar_data(URLs.MNIST_TINY)
assert len(get_image_files(path)) == 1428
assert len(get_image_files(path/'train')) == 709
assert len(get_image_files(path, include='train')) == 709
assert len(get_image_files(path, include=['train','valid'])) == 1408<jupyter_output><empty_output><jupyter_text>To split<jupyter_code># export
def random_splitter(items, valid_pct=0.2, seed=None): 
    if seed is not None: torch.manual_seed(seed)
    rand_idx = torch.randperm(len(items))
    cut = int(valid_pct * len(items))
    return rand_idx[cut:],rand_idx[:cut]
#test
trn,val = random_splitter([0,1,2,3,4,5], seed=42)
assert (trn == tensor([3, 2, 4, 1, 5])).all()
assert (val == tensor([0])).all()
# export
def _grandparent_mask(items, name):
    return [(o.parent.parent.name if isinstance(o, Path) else o.split(os.path.sep)[-2]) == name for o in items]

def grandparent_splitter(items, train_name='train', valid_name='valid'):
    return _grandparent_mask(items, train_name),_grandparent_mask(items, valid_name)
path = untar_data(URLs.MNIST_TINY)
#test
#With string filenames
path = untar_data(URLs.MNIST_TINY)
items = [path/'train'/'3'/'9932.png', path/'valid'/'7'/'7189.png', 
         path/'valid'/'7'/'7320.png', path/'train'/'7'/'9833.png',  
         path/'train'/'3'/'7666.png', path/'valid'/'3'/'925.png',
         path/'train'/'7'/'724.png', path/'valid'/'3'/'93055.png']
trn,val = grandparent_splitter(items)
assert trn == [True,False,False,True,True,False,True,False]
assert val == [False,True,True,False,False,True,False,True]<jupyter_output><empty_output><jupyter_text>To label<jupyter_code># export
def parent_labeller(items):
    return [o.parent.name if isinstance(o, Path) else o.split(os.path.sep)[-1] for o in items]
# test
assert parent_labeller(items) == ['3','7','7','7','3','3','7','3']
# export
def func_labeller(items, func):
    return [func(o) for o in items]
# test
assert func_labeller(items, lambda x: int(x.parent.name)+1) == [4,8,8,8,4,4,8,4]
# export
def re_labeller(items, pat):
    pat = re.compile(pat)
    def _inner(o):
        res = pat.search(str(o))
        assert res,f'Failed to find "{pat}" in "{o}"'
        return res.group(1)
    return func_labeller(items, _inner)
# test
pat = re.compile(r'/([^/]+)/\d+.png$')
assert re_labeller(items, pat) == ['3','7','7','7','3','3','7','3']<jupyter_output><empty_output><jupyter_text>First data:<jupyter_code>class PetsData(DataBlock):
    get_x_cls = ImageGetter
    get_y_cls = CategoryGetter
    
    def get_source(self):        return untar_data(URLs.PETS)
    def get_items(self, source): return get_image_files(source/"images")
    def split(self, items):      return random_splitter(items)
    def label(self, items):      return re_labeller(items, pat = r'/([^/]+)_\d+.jpg$')
data = PetsData()
img,cls = data.train[0]
img
img.size
cls
data.train.y.obj(0)<jupyter_output><empty_output><jupyter_text>Transforms<jupyter_code># export
class Transform(): 
    _order=0
    _tfm_y_func={TfmY.Image: 'apply_img',   TfmY.Mask: 'apply_mask', TfmY.No: 'noop',
                 TfmY.Point: 'apply_point', TfmY.Bbox: 'apply_bbox'}
    
    def apply(self, x):       return x
    def apply_img(self, y):   return self.apply(y)
    def apply_mask(self, y):  return self.apply_img(y)
    def apply_point(self, y): return y
    def apply_bbox(self, y):  return self.apply_point(y)
    
    def randomize(self): pass
    
    def __call__(self, o, tfm_y=TfmY.No):
        (x,y) = o
        self.randomize() #Ensures we have the same state for x and y
        self.x = x #Saves the x in case it's needed in the apply for y (x.size for apply_point for instance)
        return self.apply(x),getattr(self, self._tfm_y_func[tfm_y], noop)(y)
class DecodeImg(Transform):
    def __init__(self, mode_x='RGB', mode_y=None):
        self.mode_x,self.mode_y = mode_x,mode_y
        
    def apply(self, x):       return x.convert(self.mode_x)
    def apply_image(self, y): return y.convert(self.mode_x if self.mode_y is None else self.mode_y)
    def apply_mask(self, y):  return y.convert('L' if self.mode_y is None else self.mode_y)
class ResizeFixed(Transform):
    _order=10
    def __init__(self, size, mode_x=PIL.Image.BILINEAR, mode_y=None):
        if isinstance(size,int): size=(size,size)
        size = (size[1],size[0]) #PIL takes size in the otherway round
        self.size,self.mode_x,self.mode_y = size,mode_x,mode_y
        
    def apply(self, x):       return x.resize(self.size, self.mode_x)
    def apply_image(self, y): return y.resize(self.size, self.mode_x if self.mode_y is None else self.mode_y)
    def apply_mask(self, y):  return y.resize(self.size, PIL.Image.NEAREST if self.mode_y is None else self.mode_y)
class ToByteTensor(Transform):
    _order=20
    
    def apply(self, x):
        res = torch.ByteTensor(torch.ByteStorage.from_buffer(x.tobytes()))
        w,h = x.size
        return res.view(h,w,-1).permute(2,0,1)
class ToFloatTensor(Transform):
    _order=20
    def __init__(self, div_x=255., div_y=None):
        self.div_x,self.div_y = div_x,div_y
    def apply(self, x):      return x.float().div_(self.div_x)
    def apply_mask(self, x): 
        return x.long() if self.div_y is None else x.long().div_(self.div_y)
tfms = [DecodeImg(), ResizeFixed(128), ToByteTensor(), ToFloatTensor()]
data = PetsData(tfms=tfms)
datab = data.databunch()
datab.show_batch()
x,y = next(iter(datab.train_dl))
x.shape<jupyter_output><empty_output><jupyter_text>Try different data  MNIST<jupyter_code>class MnistData(DataBlock):
    get_x_cls = ImageGetter
    get_y_cls = CategoryGetter
    
    def get_source(self):        return untar_data(URLs.MNIST)
    def get_items(self, source): return get_image_files(source)
    def split(self, items):      return grandparent_splitter(items, train_name='training', valid_name='testing')
    def label(self, items):      return parent_labeller(items)
data = MnistData(tfms=[ToByteTensor(), ToFloatTensor()]).databunch()
data.show_batch()
data.train_ds.x.item_get.cmap='gray'
data.show_batch()<jupyter_output><empty_output><jupyter_text>Planet<jupyter_code>path = untar_data(URLs.PLANET_SAMPLE)
path.ls()
df = pd.read_csv(path/'labels.csv')
df.head()
# export
def onehot(x, c):
    res = torch.zeros(c)
    res[x] = 1.
    return res
# export
class MultiCategoryProcessor(CategoryProcessor):
    def __call__(self, items):
        #The vocab is defined on the first use.
        if self.vocab is None:
            vocab = set()
            for c in items: vocab = vocab.union(set(c))
            self.vocab = list(vocab)
            self.vocab.sort()
            self.otoi  = {v:k for k,v in enumerate(self.vocab)}
        return [self.proc1(o) for o in items]
    def proc1(self, item):  return [self.otoi[o] for o in item if o in self.otoi]
    
    def deproc1(self, idx): return [self.vocab[i] for i in idx]

class MultiCategoryGetter(ItemGetter):
    default_proc = MultiCategoryProcessor
    
    def get(self, o): return onehot(o, len(self.procs[0].vocab))
    def raw(self, o): return [i for i,x in enumerate(o) if x == 1]
    def show(self, x, ax): ax.set_title(';'.join(x))
# export
def read_column(df, col_name, prefix='', suffix='', delim=None):
    values = df[col_name].values.astype(str)
    values = np.char.add(np.char.add(prefix, values), suffix)
    if delim is not None:
        values = np.array(list(csv.reader(values, delimiter=delim)))
    return values
class PlanetData(DataBlock):
    get_x_cls = ImageGetter
    get_y_cls = MultiCategoryGetter
    
    def get_source(self):        
        self.path = untar_data(URLs.PLANET_SAMPLE)
        return pd.read_csv(path/'labels.csv')
    def get_items(self, source): return read_column(source, 'image_name', prefix=f'{self.path}/train/', suffix='.jpg')
    def split(self, items):      return random_splitter(items)
    def label(self, items):      return read_column(self.source, 'tags', delim=' ')
data = PlanetData(tfms=tfms).databunch()
data.show_batch()<jupyter_output><empty_output><jupyter_text>Camvid<jupyter_code># export
class SegmentMaskGetter(ImageGetter):
    default_tfm = TfmY.Mask
    def __init__(self, procs=None, cmap='tab20', alpha=0.5): 
        super().__init__(procs, cmap=cmap, alpha=alpha)
class CamvidData(DataBlock):
    get_x_cls = ImageGetter
    get_y_cls = SegmentMaskGetter
    
    def get_source(self):        return untar_data(URLs.CAMVID_TINY)      
    def get_items(self, source): return get_image_files(source/'images')
    def split(self, items):      return random_splitter(items)
    def label(self, items):      
        path_lbl = self.source/'labels'
        codes = np.loadtxt(self.source/'codes.txt', dtype=str)
        return func_labeller(items, lambda x: path_lbl/f'{x.stem}_P{x.suffix}')
data = CamvidData(tfms=tfms).databunch(bs=16)
data.show_batch()<jupyter_output><empty_output><jupyter_text>Biwii<jupyter_code>import pickle
# export
class PointsGetter(ItemGetter):
    default_tfm = TfmY.Point
    def __init__(self, procs=None, do_scale=True, y_first=False): 
        super().__init__(procs)
        self.do_scale,self.y_first = do_scale,y_first
    
    def get(self, o):
        if not isinstance(o, torch.Tensor): o = tensor(o)
        o = o.view(-1, 2).float()
        if not self.y_first: o = o.flip(1)
        if self.do_scale and hasattr(self, '_x') and self._x is not None: 
            sz = tensor(list(self._x.size)).float()
            o = o * 2/sz - 1
        return o
    
    def raw(self, o):
        o = o.flip(1)
        if hasattr(self, '_x') and self._x is not None: 
            sz = tensor([self._x.shape[1:]]).float()
            o = (o + 1) * sz/2
        return o
    
    def show(self, x, ax):
        params = {'s': 10, 'marker': '.', 'c': 'r'}
        ax.scatter(x[:, 1], x[:, 0], **params)
class BiwiData(DataBlock):
    get_x_cls = ImageGetter
    get_y_cls = PointsGetter
    
    def get_source(self):        return untar_data(URLs.BIWI_SAMPLE)      
    def get_items(self, source): return get_image_files(source/'images')
    def split(self, items):      return random_splitter(items)
    def label(self, items):      
        fn2ctr = pickle.load(open(self.source/'centers.pkl', 'rb'))
        return func_labeller(items, lambda o:fn2ctr[o.name])
data = BiwiData(tfms=tfms).databunch(bs=16)
data.show_batch()<jupyter_output><empty_output><jupyter_text>Coco<jupyter_code>from fastai.vision.data import get_annotations
#export
class BBoxProcessor(MultiCategoryProcessor):
    def __call__(self, items): 
        if self.vocab is None:
            vocab = set()
            for c in items: vocab = vocab.union(set(c[1]))
            self.vocab = ['background'] + list(vocab)
            self.vocab.sort()
            self.otoi  = {v:k for k,v in enumerate(self.vocab)}
        return [self.proc1(o) for o in items]
    def proc1(self, item):  return item[0],super().proc1(item[1])
    def deproc1(self, idx): return idx[0],super().deproc1(idx[1])
#export 
from matplotlib import patches, patheffects

def _draw_outline(o, lw):
    o.set_path_effects([patheffects.Stroke(linewidth=lw, foreground='black'), patheffects.Normal()])

def _draw_rect(ax, b, color='white', text=None, text_size=14):
    patch = ax.add_patch(patches.Rectangle(b[:2], *b[-2:], fill=False, edgecolor=color, lw=2))
    _draw_outline(patch, 4)
    if text is not None:
        patch = ax.text(*b[:2], text, verticalalignment='top', color=color, fontsize=text_size, weight='bold')
        _draw_outline(patch,1)
# export
class BBoxGetter(PointsGetter):
    default_proc = BBoxProcessor
    default_tfm = TfmY.Bbox
     
    def get(self, o): return super().get(o[0]).view(-1,4),o[1]
    def raw(self, o): return super().raw(o[0].view(-1,2)).view(-1,4),o[1]
    
    def show(self, x, ax):
        bbox,label = x
        for b,l in zip(bbox, label): 
            if l != 'background': _draw_rect(ax, [b[1],b[0],b[3]-b[1],b[2]-b[0]], text=l)
def bb_pad_collate(samples, pad_idx=0):
    max_len = max([len(s[1][1]) for s in samples])
    bboxes = torch.zeros(len(samples), max_len, 4)
    labels = torch.zeros(len(samples), max_len).long() + pad_idx
    imgs = []
    for i,s in enumerate(samples):
        imgs.append(s[0][None])
        bbs, lbls = s[1]
        if not (bbs.nelement() == 0):
            bboxes[i,-len(lbls):] = bbs
            labels[i,-len(lbls):] = tensor(lbls)
    return torch.cat(imgs,0), (bboxes,labels)
class CocoData(DataBlock):
    get_x_cls = ImageGetter
    get_y_cls = BBoxGetter
    
    def get_source(self):        return untar_data(URLs.COCO_TINY)      
    def get_items(self, source): return get_image_files(source/'train')
    def split(self, items):      return random_splitter(items)
    def label(self, items):      
        images, lbl_bbox = get_annotations(self.source/'train.json')
        img2bbox = dict(zip(images, lbl_bbox))
        return func_labeller(items, lambda o:img2bbox[o.name])
    
    def databunch(self, bs=64, **kwargs):
        kwargs['collate_fn'] = bb_pad_collate
        return super().databunch(bs=bs, **kwargs)
data = CocoData(tfms=tfms).databunch(bs=16)
data.show_batch()<jupyter_output><empty_output><jupyter_text>Export<jupyter_code>! python notebook2script.py "200_datablock_config.ipynb"<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14568825139451894, 'fraction_numerical': 0.017115337975955377, 'mean_word_length': 3.6027746770849944, 'pattern_counts': {'":': 0, '<': 81, '<?xml version=': 0, '>': 85, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '20', 'dir': 'jupyter-structured-clean-dedup', 'id': '76460', 'license': 'Apache-2.0', 'n_tokens_mistral': 10664, 'n_tokens_neox': 10481, 'n_words': 2541, 'path': 'dev_nb/200_datablock_config.ipynb', 'repo_name': 'hfoffani/fastai-p2', 'size': '28863'}