pydrepr/drepr/executors/cf_convention_map/cf_convention_map.py
def execute(cls, ds_model: DRepr, resource_file: str):
        """
        Execution steps:

        1. create resource reader (only support one resource)
        2. execute preprocessing function (which is map function and only mutate the value)
        3. gather attributes
        4. Create table:
            a. If there are duplicated predicates (duplicated column_id), then the resulted array will have an extra dimension at
               the end, which size is equal to the number of duplicated columns of `column_id`. (see the ndarray documentation)
        """
        # 1st: create resource reader
        resource = ds_model.resources[0]
        if resource.type == ResourceType.NetCDF4:
            reader = NetCDF4Reader.from_file(resource_file)
        elif resource.type == ResourceType.GeoTIFF:
            reader = GeoTIFFReader.from_file(resource_file)
        elif resource.type == ResourceType.NPDict:
            reader = NPDictReader.from_file(resource_file)
        elif resource.type == ResourceType.Shapefile:
            reader = ShapefileReader.from_file(resource_file)
        elif resource.type == ResourceType.Container:
            reader = ReaderContainer.get_instance().get(resource_file)
        else:
            raise NotImplementedError()

        # 2nd: execute preprocessing function
        context = Context(reader)
        for preprocess_fn in ds_model.preprocessing:
            if preprocess_fn.type == PreprocessingType.pmap:
                steps = preprocess_fn.value.path.steps
                n_range_exprs = sum(int(not isinstance(step, IndexExpr)) for step in steps)
                fn = PyExec.compile(preprocess_fn.value.code)

                if n_range_exprs == 0:
                    index = [step.val for step in steps]
                    value = fn.exec(reader.get_value(index), index, context)
                    reader.set_value(index, value)
                else:
                    # TODO: improve it
                    range_indices = [i for i in range(len(steps)) if isinstance(steps[i], RangeExpr)]
                    index = [step.start if isinstance(step, RangeExpr) else step.val for step in steps]
                    if n_range_exprs == 1:
                        range_idx = range_indices[0]
                        if steps[range_idx].end is None:
                            if range_idx == 0:
                                end = reader.len_range()
                            else:
                                end = len(reader.get_value(index[:range_idx]))
                        else:
                            end = steps[range_idx].end
                        for i in range(steps[range_idx].start, end, steps[range_idx].step):
                            index[range_idx] = i
                            value = fn.exec(reader.get_value(index), index, context)
                            reader.set_value(index, value)
                    else:
                        raise NotImplementedError()
            else:
                raise Exception("You found a bug")

        # 3rd: gather attributes
        attrs = {}
        new_attrs = {}
        for attr in ds_model.attrs:
            attrs[attr.id] = reader.select(attr.path.steps)
            is_array_attr = any(not isinstance(step, IndexExpr) for step in attr.path.steps)
            if is_array_attr:
                step2dim = []
                path = attr.path
                if len(attr.missing_values) == 0:
                    nodata = None
                elif len(attr.missing_values) == 1:
                    # we need to convert this into numpy because comparing between numpy number
                    # and python number leads to error
                    if isinstance(attrs[attr.id], np.ndarray):
                        nodata = NoData(np.array(attr.missing_values, dtype=attrs[attr.id].dtype)[0])
                    else:
                        nodata = NoData(attr.missing_values[0])
                else:
                    # need to convert other values back to just one value and use it!
                    raise NotImplementedError()

                count = 0
                for step in path.steps:
                    if not isinstance(step, IndexExpr):
                        # in this function all the shape of ndarray will be determined by range only,
                        # and they will follow the order. Will we have other cases?
                        assert isinstance(step, RangeExpr)
                        step2dim.append(count)
                        count += 1
                    else:
                        step2dim.append(None)

                new_attrs[f"dnode:{attr.id}"] = ArrayAttr(f"dnode:{attr.id}", attrs[attr.id], path, step2dim, nodata)
            else:
                new_attrs[f"dnode:{attr.id}"] = ScalarAttr(f"dnode:{attr.id}", attrs[attr.id])

        # 4th: create tables from the semantic model
        sm = ds_model.sm
        tables = {}
        alignments = {}

        for nid, node in sm.nodes.items():
            if isinstance(node, ClassNode):
                outgoing_edges = list(sm.iter_outgoing_edges(nid))
                tables[nid] = {}
                for e in outgoing_edges:
                    c = sm.nodes[e.target_id]
                    if isinstance(c, DataNode):
                        if isinstance(attrs[c.attr_id], np.ndarray):
                            step2dim = []
                            _attr = ds_model.get_attr_by_id(c.attr_id)
                            path = _attr.path
                            if len(_attr.missing_values) == 0:
                                nodata = None
                            elif len(_attr.missing_values) == 1:
                                nodata = NoData(_attr.missing_values[0])
                            else:
                                # need to convert other values back to just one value and use it!
                                raise NotImplementedError()

                            count = 0
                            for step in path.steps:
                                if not isinstance(step, IndexExpr):
                                    # in this function all the shape of ndarray will be determined by range only,
                                    # and they will follow the order. Will we have other cases?
                                    assert isinstance(step, RangeExpr)
                                    step2dim.append(count)
                                    count += 1
                                else:
                                    step2dim.append(None)
                            # tables[nid][e.target_id] = ColArray(e.target_id, attrs[c.attr_id], path, step2dim, nodata)
                        else:
                            pass
                            # tables[nid][e.target_id] = ColSingle(attrs[c.attr_id])
                    elif isinstance(c, LiteralNode):
                        pass
                        # tables[nid][e.target_id] = ColSingle(c.value)
                    elif not isinstance(c, ClassNode):
                        raise NotImplementedError()
            elif isinstance(node, LiteralNode):
                new_attrs[nid] = ScalarAttr(nid, node.value)

        # inferring alignments
        for align in ds_model.aligns:
            source = f"dnode:" + align.source
            target = f"dnode:" + align.target
            if source not in alignments:
                alignments[source] = {}
            alignments[source][target] = align
        return None, new_attrs
        # return ndarray.NDArrayGraph(sm, tables, alignments), new_attrs

        Execution steps:

        1. create resource reader (only support one resource)
        2. execute preprocessing function (which is map function and only mutate the value)
        3. gather attributes
        4. Create table:
            a. If there are duplicated predicates (duplicated column_id), then the resulted array will have an extra dimension at
               the end, which size is equal to the number of duplicated columns of `column_id`. (see the ndarray documentation)
        
Execution steps:
1. create resource reader (only support one resource)
2. execute preprocessing function (which is map function and only mutate the value)
3. gather attributes
4. Create table:
a. If there are duplicated predicates (duplicated column_id), then the resulted array will have an extra dimension at
the end, which size is equal to the number of duplicated columns of `column_id`.
Execution steps : 1 . create resource reader ( only support one resource ) 2 . execute preprocessing function ( which is map function and only mutate the value ) 3 . gather attributes 4 . Create table : a . If there are duplicated predicates ( duplicated column_id ) then the resulted array will have an extra dimension at the end which size is equal to the number of duplicated columns of ` column_id ` .
def execute(cls, ds_model: DRepr, resource_file: str):
        resource = ds_model.resources[0]
        if resource.type == ResourceType.NetCDF4:
            reader = NetCDF4Reader.from_file(resource_file)
        elif resource.type == ResourceType.GeoTIFF:
            reader = GeoTIFFReader.from_file(resource_file)
        elif resource.type == ResourceType.NPDict:
            reader = NPDictReader.from_file(resource_file)
        elif resource.type == ResourceType.Shapefile:
            reader = ShapefileReader.from_file(resource_file)
        elif resource.type == ResourceType.Container:
            reader = ReaderContainer.get_instance().get(resource_file)
        else:
            raise NotImplementedError()
        context = Context(reader)
        for preprocess_fn in ds_model.preprocessing:
            if preprocess_fn.type == PreprocessingType.pmap:
                steps = preprocess_fn.value.path.steps
                n_range_exprs = sum(int(not isinstance(step, IndexExpr)) for step in steps)
                fn = PyExec.compile(preprocess_fn.value.code)
                if n_range_exprs == 0:
                    index = [step.val for step in steps]
                    value = fn.exec(reader.get_value(index), index, context)
                    reader.set_value(index, value)
                else:
                    range_indices = [i for i in range(len(steps)) if isinstance(steps[i], RangeExpr)]
                    index = [step.start if isinstance(step, RangeExpr) else step.val for step in steps]
                    if n_range_exprs == 1:
                        range_idx = range_indices[0]
                        if steps[range_idx].end is None:
                            if range_idx == 0:
                                end = reader.len_range()
                            else:
                                end = len(reader.get_value(index[:range_idx]))
                        else:
                            end = steps[range_idx].end
                        for i in range(steps[range_idx].start, end, steps[range_idx].step):
                            index[range_idx] = i
                            value = fn.exec(reader.get_value(index), index, context)
                            reader.set_value(index, value)
                    else:
                        raise NotImplementedError()
            else:
                raise Exception("You found a bug")
        attrs = {}
        new_attrs = {}
        for attr in ds_model.attrs:
            attrs[attr.id] = reader.select(attr.path.steps)
            is_array_attr = any(not isinstance(step, IndexExpr) for step in attr.path.steps)
            if is_array_attr:
                step2dim = []
                path = attr.path
                if len(attr.missing_values) == 0:
                    nodata = None
                elif len(attr.missing_values) == 1:
                    if isinstance(attrs[attr.id], np.ndarray):
                        nodata = NoData(np.array(attr.missing_values, dtype=attrs[attr.id].dtype)[0])
                    else:
                        nodata = NoData(attr.missing_values[0])
                else:
                    raise NotImplementedError()
                count = 0
                for step in path.steps:
                    if not isinstance(step, IndexExpr):
                        assert isinstance(step, RangeExpr)
                        step2dim.append(count)
                        count += 1
                    else:
                        step2dim.append(None)
                new_attrs[f"dnode:{attr.id}"] = ArrayAttr(f"dnode:{attr.id}", attrs[attr.id], path, step2dim, nodata)
            else:
                new_attrs[f"dnode:{attr.id}"] = ScalarAttr(f"dnode:{attr.id}", attrs[attr.id])
        sm = ds_model.sm
        tables = {}
        alignments = {}
        for nid, node in sm.nodes.items():
            if isinstance(node, ClassNode):
                outgoing_edges = list(sm.iter_outgoing_edges(nid))
                tables[nid] = {}
                for e in outgoing_edges:
                    c = sm.nodes[e.target_id]
                    if isinstance(c, DataNode):
                        if isinstance(attrs[c.attr_id], np.ndarray):
                            step2dim = []
                            _attr = ds_model.get_attr_by_id(c.attr_id)
                            path = _attr.path
                            if len(_attr.missing_values) == 0:
                                nodata = None
                            elif len(_attr.missing_values) == 1:
                                nodata = NoData(_attr.missing_values[0])
                            else:
                                raise NotImplementedError()
                            count = 0
                            for step in path.steps:
                                if not isinstance(step, IndexExpr):
                                    assert isinstance(step, RangeExpr)
                                    step2dim.append(count)
                                    count += 1
                                else:
                                    step2dim.append(None)
                        else:
                            pass
                    elif isinstance(c, LiteralNode):
                        pass
                    elif not isinstance(c, ClassNode):
                        raise NotImplementedError()
            elif isinstance(node, LiteralNode):
                new_attrs[nid] = ScalarAttr(nid, node.value)
        for align in ds_model.aligns:
            source = f"dnode:" + align.source
            target = f"dnode:" + align.target
            if source not in alignments:
                alignments[source] = {}
            alignments[source][target] = align
        return None, new_attrs
def execute ( cls , ds_model : DRepr , resource_file : str ) : resource = ds_model . resources [ 0 ] if resource . type == ResourceType . NetCDF4 : reader = NetCDF4Reader . from_file ( resource_file ) elif resource . type == ResourceType . GeoTIFF : reader = GeoTIFFReader . from_file ( resource_file ) elif resource . type == ResourceType . NPDict : reader = NPDictReader . from_file ( resource_file ) elif resource . type == ResourceType . Shapefile : reader = ShapefileReader . from_file ( resource_file ) elif resource . type == ResourceType . Container : reader = ReaderContainer . get_instance ( ) . get ( resource_file ) else : raise NotImplementedError ( ) context = Context ( reader ) for preprocess_fn in ds_model . preprocessing : if preprocess_fn . type == PreprocessingType . pmap : steps = preprocess_fn . value . path . steps n_range_exprs = sum ( int ( not isinstance ( step , IndexExpr ) ) for step in steps ) fn = PyExec . compile ( preprocess_fn . value . code ) if n_range_exprs == 0 : index = [ step . val for step in steps ] value = fn . exec ( reader . get_value ( index ) , index , context ) reader . set_value ( index , value ) else : range_indices = [ i for i in range ( len ( steps ) ) if isinstance ( steps [ i ] , RangeExpr ) ] index = [ step . start if isinstance ( step , RangeExpr ) else step . val for step in steps ] if n_range_exprs == 1 : range_idx = range_indices [ 0 ] if steps [ range_idx ] . end is None : if range_idx == 0 : end = reader . len_range ( ) else : end = len ( reader . get_value ( index [ : range_idx ] ) ) else : end = steps [ range_idx ] . end for i in range ( steps [ range_idx ] . start , end , steps [ range_idx ] . step ) : index [ range_idx ] = i value = fn . exec ( reader . get_value ( index ) , index , context ) reader . set_value ( index , value ) else : raise NotImplementedError ( ) else : raise Exception ( "You found a bug" ) attrs = { } new_attrs = { } for attr in ds_model . attrs : attrs [ attr . id ] = reader . select ( attr . path . steps ) is_array_attr = any ( not isinstance ( step , IndexExpr ) for step in attr . path . steps ) if is_array_attr : step2dim = [ ] path = attr . path if len ( attr . missing_values ) == 0 : nodata = None elif len ( attr . missing_values ) == 1 : if isinstance ( attrs [ attr . id ] , np . ndarray ) : nodata = NoData ( np . array ( attr . missing_values , dtype = attrs [ attr . id ] . dtype ) [ 0 ] ) else : nodata = NoData ( attr . missing_values [ 0 ] ) else : raise NotImplementedError ( ) count = 0 for step in path . steps : if not isinstance ( step , IndexExpr ) : assert isinstance ( step , RangeExpr ) step2dim . append ( count ) count += 1 else : step2dim . append ( None ) new_attrs [ f"dnode:{attr.id}" ] = ArrayAttr ( f"dnode:{attr.id}" , attrs [ attr . id ] , path , step2dim , nodata ) else : new_attrs [ f"dnode:{attr.id}" ] = ScalarAttr ( f"dnode:{attr.id}" , attrs [ attr . id ] ) sm = ds_model . sm tables = { } alignments = { } for nid , node in sm . nodes . items ( ) : if isinstance ( node , ClassNode ) : outgoing_edges = list ( sm . iter_outgoing_edges ( nid ) ) tables [ nid ] = { } for e in outgoing_edges : c = sm . nodes [ e . target_id ] if isinstance ( c , DataNode ) : if isinstance ( attrs [ c . attr_id ] , np . ndarray ) : step2dim = [ ] _attr = ds_model . get_attr_by_id ( c . attr_id ) path = _attr . path if len ( _attr . missing_values ) == 0 : nodata = None elif len ( _attr . missing_values ) == 1 : nodata = NoData ( _attr . missing_values [ 0 ] ) else : raise NotImplementedError ( ) count = 0 for step in path . steps : if not isinstance ( step , IndexExpr ) : assert isinstance ( step , RangeExpr ) step2dim . append ( count ) count += 1 else : step2dim . append ( None ) else : pass elif isinstance ( c , LiteralNode ) : pass elif not isinstance ( c , ClassNode ) : raise NotImplementedError ( ) elif isinstance ( node , LiteralNode ) : new_attrs [ nid ] = ScalarAttr ( nid , node . value ) for align in ds_model . aligns : source = f"dnode:" + align . source target = f"dnode:" + align . target if source not in alignments : alignments [ source ] = { } alignments [ source ] [ target ] = align return None , new_attrs
Execution steps:
1. create resource reader (only support one resource)
2. execute preprocessing function (which is map function and only mutate the value)
3. gather attributes
4.
Execution steps : 1 . create resource reader ( only support one resource ) 2 . execute preprocessing function ( which is map function and only mutate the value ) 3 . gather attributes 4 .
"""
        Execution steps:

        1. create resource reader (only support one resource)
        2. execute preprocessing function (which is map function and only mutate the value)
        3. gather attributes
        4. Create table:
            a. If there are duplicated predicates (duplicated column_id), then the resulted array will have an extra dimension at
               the end, which size is equal to the number of duplicated columns of `column_id`. (see the ndarray documentation)
        """ # 1st: create resource reader # 2nd: execute preprocessing function # TODO: improve it # 3rd: gather attributes # we need to convert this into numpy because comparing between numpy number # and python number leads to error # need to convert other values back to just one value and use it! # in this function all the shape of ndarray will be determined by range only, # and they will follow the order. Will we have other cases? # 4th: create tables from the semantic model # need to convert other values back to just one value and use it! # in this function all the shape of ndarray will be determined by range only, # and they will follow the order. Will we have other cases? # tables[nid][e.target_id] = ColArray(e.target_id, attrs[c.attr_id], path, step2dim, nodata) # tables[nid][e.target_id] = ColSingle(attrs[c.attr_id]) # tables[nid][e.target_id] = ColSingle(c.value) # inferring alignments # return ndarray.NDArrayGraph(sm, tables, alignments), new_attrs
cls
ds_model
DRepr
resource_file
str