<jupyter_start><jupyter_text>High Level Python APIUp until now we have been using the low level Python API that Bifrost has to show the inner workings of the framework and how to build a pipeline.  However, Bifrost also has a high level Python API that makes building blocks and pipelines easier with less code.  In this section we will look at this interface.The general idea of the high-level API is to allow a pipeline to be built out of processing blocks that are connected together like Lego. For example, here is a pipeline that reads from a sigproc file, performs a fast dispersion measure transform, and then writes out the data to disk:```pythonimport bifrost as bfimport sysfilenames = sys.argv[1:]print("Building pipeline")data = bf.blocks.read_sigproc(filenames, gulp_nframe=128)data = bf.blocks.copy(data, 'cuda')data = bf.blocks.transpose(data, ['pol', 'freq', 'time'])data = bf.blocks.fdmt(data, max_dm=100.)data = bf.blocks.copy(data, 'cuda_host')bf.blocks.write_sigproc(data)print("Running pipeline")bf.get_default_pipeline().run()print("All done")```Here we are continually passing the block's output, `data`, to the next block in the pipeline. At runtime, the blocks are connected together (as a directed graph) with ring buffers in between.  Creating a transform blockMany common processing operations have bifrost blocks, but it likely that you will need to make a custom block at some stage. To see how, let's start by revisiting the `CopyOp` block from the pipelines section:<jupyter_code>class CopyOp(object):
    def __init__(self, iring, oring, ntime_gulp=250, guarantee=True, core=-1):
        self.iring = iring
        self.oring = oring
        self.ntime_gulp = ntime_gulp
        self.guarantee = guarantee
        self.core = core
        
    def main(self):
        with self.oring.begin_writing() as oring:
            for iseq in self.iring.read(guarantee=self.guarantee):
                ihdr = json.loads(iseq.header.tostring())
                
                print("Copy: Start of new sequence:", str(ihdr))
                
                time_tag = ihdr['time_tag']
                navg     = ihdr['navg']
                nbeam    = ihdr['nbeam']
                chan0    = ihdr['chan0']
                nchan    = ihdr['nchan']
                chan_bw  = ihdr['bw'] / nchan
                npol     = ihdr['npol']
                pols     = ihdr['pols']
                pols     = pols.replace('CR', 'XY_real')
                pols     = pols.replace('CI', 'XY_imag')

                igulp_size = self.ntime_gulp*nbeam*nchan*npol*4        # float32
                ishape = (self.ntime_gulp,nbeam,nchan,npol)
                self.iring.resize(igulp_size, igulp_size*5)
                
                ogulp_size = igulp_size
                oshape = ishape
                self.oring.resize(ogulp_size)
                
                ohdr = ihdr.copy()
                ohdr_str = json.dumps(ohdr)
                
                iseq_spans = iseq.read(igulp_size)
                with oring.begin_sequence(time_tag=time_tag, header=ohdr_str) as oseq:
                    for ispan in iseq_spans:
                        if ispan.size < igulp_size:
                            continue # Ignore final gulp
                            
                        with oseq.reserve(ogulp_size) as ospan:
                            idata = ispan.data_view(numpy.float32)
                            odata = ospan.data_view(numpy.float32)    
                            odata[...] = idata<jupyter_output><empty_output><jupyter_text>There is a lot of setup in here and iteration control that is common across many of the blocks that we have looked at.  In the high level API much of this can be abstracted away using the classes defined in `bifrost.pipelines`:<jupyter_code>import copy

from bifrost import pipeline

class NewCopyOp(pipeline.TransformBlock):
    def __init__(self, iring, *args, **kwargs):
        super(NewCopyOp, self).__init__(iring, *args, **kwargs)
        
    def on_sequence(self, iseq):
        ihdr = iseq.header
        print("Copy: Start of new sequence:", str(ihdr))
        
        ohdr = copy.deepcopy(iseq.header)
        return ohdr

    def on_data(self, ispan, ospan):
        in_nframe  = ispan.nframe
        out_nframe = in_nframe

        idata = ispan.data
        odata = ospan.data

        odata[...] = idata
        return out_nframe<jupyter_output><empty_output><jupyter_text>That is much more compact.  The key things in this new class are: 1. the `on_sequence` method is called whenever a new sequence starts and is used to update the header for the output ring buffer and 2. the `on_data` method is called for each span/gulp that is processed.Put another way, the `on_sequence` happens when there is new *metadata*, and `on_data` happens whenever there is new *data* to process. For example, `on_sequence` may be called when reading a new file, or starting an observation.   Creating a source block Similarly, we can translate the `GeneratorOp` and `WriterOp` blocks as well using sub-classes of `bifrost.pipeline.SourceBlock` and `bifrost.pipeline.SinkBlock`, respectively:<jupyter_code>import os
import time
import numpy

class NewGeneratorOp(pipeline.SourceBlock):
    def __init__(self, ntime_gulp, *args, **kwargs):
        super(NewGeneratorOp, self).__init__(['generator',], 1,
                                             *args, **kwargs)
        
        self.ntime_gulp = ntime_gulp
        self.ngulp_done = 0
        self.ngulp_max = 10
        
        self.navg = 24
        tint = self.navg / 25e3
        self.tgulp = tint * self.ntime_gulp
        self.nbeam = 1
        self.chan0 = 1234
        self.nchan = 16*184
        self.npol = 4
        
    def create_reader(self, name):
        self.ngulp_done = 0
        
        class Random(object):
            def __init__(self, name):
                self.name = name
            def __enter__(self):
                return self
            def __exit__(self, type, value, tb):
                return True
            def read(self, *args):
                return numpy.random.randn(*args)
                
        return Random(name)
        
    def on_sequence(self, reader, name):
        ohdr = {'time_tag': int(int(time.time())*196e6),
                'seq0':     0, 
                'chan0':    self.chan0,
                'cfreq0':   self.chan0*25e3,
                'bw':       self.nchan*25e3,
                'navg':     self.navg,
                'nbeam':    self.nbeam,
                'nchan':    self.nchan,
                'npol':     self.npol,
                'pols':     'XX,YY,CR,CI',
               }
        ohdr['_tensor'] = {'dtype':  'f32',
                           'shape':  [-1,
                                      self.ntime_gulp,
                                      self.nbeam,
                                      self.nchan,
                                      self.npol]
                          }
        return [ohdr,]
        
    def on_data(self, reader, ospans):
        indata = reader.read(self.ntime_gulp, self.nbeam, self.nchan, self.npol)
        time.sleep(self.tgulp)

        if indata.shape[0] == self.ntime_gulp \
           and self.ngulp_done < self.ngulp_max:
            ospans[0].data[...] = indata
            self.ngulp_done += 1
            return [1]
        else:
            return [0]<jupyter_output><empty_output><jupyter_text>For the `bifrost.pipeline.SourceBlock` we need have slightly different requirements on `on_sequence` and `on_data`.  Plus, we also need to define a `create_reader` method that returns a context manager (a class with `__enter__` and `__exit__` methods).  For `on_sequence` we need to accept two arguments: a context manager created by `create_reader` and an identifying name (although it is not used here).  For `on_data` we also have two arguments now, the context manager and a list of output spans.  In here we need to grab the data from `reader` and put it into the appropriate part of the output spans. The all-important bifrost `_tensor` dictionaryWe also see in `on_sequence` here that the header dictionary has a new required `_tensor` key.  This key is the key to automatically chaining blocks together into a pipeline since it defines the data type and dimensionality for the spans/gulps. At a minimum, the `_tensor` must define the data dtype and shape:```python    '_tensor': {            'dtype':  self.dtype,            'shape':  [-1, self.gulp_size],            },```The first index of shape is -1, to indicate that this is a single gulp from the data stream. However, most block require three additional keywords: * `labels`, which give human-friendly names to each axis (e.g. 'time' or 'frequency').* `scales`, which defines the start value and step size for each axis (e.g. `[1420, 0.1]` sets start value 1420, step size 0.1).* `units`, which defines the units for each axis (e.g. 's' or 'MHz'). These are parsed using [Pint](https://pint.readthedocs.io/en/stable/), and can be used for consistency checking. For example, the `bf.views.merge_axes` view won't allow axes to merge if they have different units, say 'MHz' and 'Jy'. If you attempted to merge axes with 'MHz' and 'kHz' units, this would be allowed, and the corresponding `scales` are updated consistently.Here is another example `_tensor` with all keywords:```python    t0 = 1620192408.005579   unix timestamp from when writing this tutorial    dt = 0.1                 0.1 second step size    '_tensor': {            'dtype':  'cf32'            'shape':  [-1, 1024, 4],            'labels': ['time', 'frequency', 'pol'],            'units':  ['s', 'MHz', ''],            'scales': [[t0, dt], [1420, 0.1], [None, None]]    }```A transform block reads the tensor metadata, and must output a copy of the tensor with any required changes to shape/scale/units etc.  Creating a sink blockWe can also translate the original `WriterOp`:<jupyter_code>class NewWriterOp(pipeline.SinkBlock):
    def __init__(self, iring, *args, **kwargs):
        super(NewWriterOp, self).__init__(iring, *args, **kwargs)
        
        self.time_tag = None
        self.navg = 0
        self.nbeam = 0
        self.nchan = 0
        self.npol = 0
        
    def on_sequence(self, iseq):
        ihdr = iseq.header
        print("Writer: Start of new sequence:", str(ihdr))
        
        self.time_tag = iseq.time_tag
        self.navg = ihdr['navg']
        self.nbeam = ihdr['nbeam']
        self.nchan = ihdr['nchan']
        self.npol = ihdr['npol']

    def on_data(self, ispan):
        idata = ispan.data.view(numpy.float32)
        idata = idata.reshape(-1, self.nbeam, self.nchan, self.npol)
        
        with open(f"{self.time_tag}.dat", 'wb') as fh:
            fh.write(idata.tobytes())
            print('  ', fh.name, '@', os.path.getsize(fh.name))
        self.time_tag += self.navg * idata.shape[0] * (int(196e6) // int(25e3))<jupyter_output><empty_output><jupyter_text>Since this is a data sink we only have one argument for `on_data` which gives the block the current data span/gulp.We then can put these new blocks all together and launch them under Bifrost's default pipeline with:<jupyter_code>b_gen = NewGeneratorOp(250)
b_cpy = NewCopyOp(b_gen)
b_out = NewWriterOp(b_cpy)

p =  pipeline.get_default_pipeline()
p.run()
del p<jupyter_output>Copy: Start of new sequence: {'time_tag': 317553970916000000, 'seq0': 0, 'chan0': 1234, 'cfreq0': 30850000.0, 'bw': 73600000.0, 'navg': 24, 'nbeam': 1, 'nchan': 2944, 'npol': 4, 'pols': 'XX,YY,CR,CI', '_tensor': {'dtype': 'f32', 'shape': [-1, 250, 1, 2944, 4]}, 'name': 'unnamed-sequence-0', 'gulp_nframe': 1}
Writer: Start of new sequence: {'time_tag': 317553970916000000, 'seq0': 0, 'chan0': 1234, 'cfreq0': 30850000.0, 'bw': 73600000.0, 'navg': 24, 'nbeam': 1, 'nchan': 2944, 'npol': 4, 'pols': 'XX,YY,CR,CI', '_tensor': {'dtype': 'f32', 'shape': [-1, 250, 1, 2944, 4]}, 'name': 'unnamed-sequence-0', 'gulp_nframe': 1}
   317553970916000000.dat @ 11776000
   317553970963040000.dat @ 11776000
   317553971010080000.dat @ 11776000
   317553971057120000.dat @ 11776000
   317553971104160000.dat @ 11776000
   317553971151200000.dat @ 11776000
   317553971198240000.dat @ 11776000
   317553971245280000.dat @ 11776000
   317553971292320000.dat @ 11776000
   317553971339360000.dat @ 11776000
{'fraction_non_alphanumeric': 0.10348170288391631, 'fraction_numerical': 0.04273366184667582, 'mean_word_length': 2.3659597607395324, 'pattern_counts': {'":': 0, '<': 22, '<?xml version=': 0, '>': 20, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '47325', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 4170, 'n_tokens_neox': 3738, 'n_words': 1189, 'path': '07_high_level_api.ipynb', 'repo_name': 'telegraphic/bifrost_tutorial', 'size': '12379'}