pydrepr/drepr/models/drepr.py
def to_engine_format(self) -> EngineFormat:
        """
        Turn this D-REPR configuration into the format that the engine can read
        :return:
        """
        # map string id to incremental numbers (for resource id and attribute id)
        ridmap = OrderedDict()
        for resource in self.resources:
            ridmap[resource.id] = len(ridmap)
        for pref in self.preprocessing:
            if pref.value.output is not None:
                ridmap[pref.value.output] = len(ridmap)
        aidmap = OrderedDict()
        for attr in self.attrs:
            aidmap[attr.id] = len(aidmap)

        resources = []
        for res in self.resources:
            resources.append({"type": res.type.value})
            if isinstance(res.prop, CSVProp):
                resources[-1]['value'] = {
                    "resource_id": ridmap[res.id],
                    "delimiter": res.prop.delimiter
                }
            else:
                resources[-1]['value'] = ridmap[res.id]

        preprocessing = []
        for pref in self.preprocessing:
            prepro = {
                "type": pref.type.value,
                "resource_id": ridmap[pref.value.resource_id],
                "path": pref.value.path.to_engine_format(),
                "output": ridmap[pref.value.output] if pref.value.output is not None else None,
            }
            if isinstance(pref.value, PMap):
                prepro['code'] = pref.value.code
                prepro['change_structure'] = pref.value.change_structure
            elif isinstance(pref.value, (PFilter, PSplit)):
                prepro['code'] = pref.value.code
            elif isinstance(pref.value, RMap):
                prepro['func_id'] = {"t": pref.value.func_id.value}
            else:
                raise NotImplementedError()
            preprocessing.append(prepro)

        attributes = [{
            "id": aidmap[a.id],
            "resource_id": ridmap[a.resource_id],
            "path": a.path.to_engine_format(),
            "unique": a.unique,
            "sorted": a.sorted.value,
            "vtype": a.value_type.value,
            "missing_values": [self._serde_engine_value(v) for v in a.missing_values]
        } for a in self.attrs]

        alignments = []
        for align in self.aligns:
            if isinstance(align, RangeAlignment):
                alignments.append({
                    "type": AlignmentType.range.value,
                    "source": aidmap[align.source],
                    "target": aidmap[align.target],
                    "aligned_dims": [{
                        "source": ad.source_idx,
                        "target": ad.target_idx
                    } for ad in align.aligned_steps]
                })
            elif isinstance(align, ValueAlignment):
                alignments.append({
                    "type": AlignmentType.value.value,
                    "source": aidmap[align.source],
                    "target": aidmap[align.target]
                })
            else:
                raise NotImplementedError()

        engine_sm: Dict[str, Any] = {
            "nodes": [],
            "edges": [],
            "prefixes": [(prefix, uri) for prefix, uri in self.sm.prefixes.items()]
        }
        nodes = {"cnodes": [], "dnodes": [], "lnodes": []}
        for node_id, node in self.sm.nodes.items():
            if isinstance(node, ClassNode):
                nodes['cnodes'].append({
                    "type": "class_node",
                    "node_id": node_id,
                    "rel_label": node.label,
                    "abs_label": self.sm.get_abs_iri(node.label)
                })
            elif isinstance(node, DataNode):
                nodes['dnodes'].append({
                    "type": "data_node",
                    "node_id": node_id,
                    "attr_id": aidmap[node.attr_id],
                    "data_type": node.data_type.value if node.data_type is not None else None,
                })
            elif isinstance(node, LiteralNode):
                nodes["lnodes"].append({
                    "type": "literal_node",
                    "node_id": node_id,
                    "val": self._serde_engine_value(node.value),
                    "data_type": node.data_type.value if node.data_type is not None else None,
                })
            else:
                raise NotImplementedError()

        for k in ["cnodes", "dnodes", "lnodes"]:
            for n in nodes[k]:
                engine_sm['nodes'].append(n)
        nidmap = {}
        for i, n in enumerate(engine_sm["nodes"]):
            nidmap[n['node_id']] = i
            n['node_id'] = i

        for eid, edge in self.sm.edges.items():
            engine_sm['edges'].append({
                "edge_id": eid,
                "source": nidmap[edge.source_id],
                "target": nidmap[edge.target_id],
                "rel_label": edge.label,
                "abs_label": self.sm.get_abs_iri(edge.label),
                "is_subject": edge.is_subject
            })

        edges_optional = [
            not edge.is_required
            for edge in self.sm.edges.values()
        ]

        return EngineFormat({
            "resources": resources,
            "preprocessing": preprocessing,
            "attributes": attributes,
            "alignments": alignments,
            "semantic_model": engine_sm
        }, edges_optional, ridmap, aidmap, nidmap)

        Turn this D-REPR configuration into the format that the engine can read
        :return:
        
Turn this D-REPR configuration into the format that the engine can read
Turn this D - REPR configuration into the format that the engine can read
def to_engine_format(self) -> EngineFormat:
        ridmap = OrderedDict()
        for resource in self.resources:
            ridmap[resource.id] = len(ridmap)
        for pref in self.preprocessing:
            if pref.value.output is not None:
                ridmap[pref.value.output] = len(ridmap)
        aidmap = OrderedDict()
        for attr in self.attrs:
            aidmap[attr.id] = len(aidmap)
        resources = []
        for res in self.resources:
            resources.append({"type": res.type.value})
            if isinstance(res.prop, CSVProp):
                resources[-1]['value'] = {
                    "resource_id": ridmap[res.id],
                    "delimiter": res.prop.delimiter
                }
            else:
                resources[-1]['value'] = ridmap[res.id]
        preprocessing = []
        for pref in self.preprocessing:
            prepro = {
                "type": pref.type.value,
                "resource_id": ridmap[pref.value.resource_id],
                "path": pref.value.path.to_engine_format(),
                "output": ridmap[pref.value.output] if pref.value.output is not None else None,
            }
            if isinstance(pref.value, PMap):
                prepro['code'] = pref.value.code
                prepro['change_structure'] = pref.value.change_structure
            elif isinstance(pref.value, (PFilter, PSplit)):
                prepro['code'] = pref.value.code
            elif isinstance(pref.value, RMap):
                prepro['func_id'] = {"t": pref.value.func_id.value}
            else:
                raise NotImplementedError()
            preprocessing.append(prepro)
        attributes = [{
            "id": aidmap[a.id],
            "resource_id": ridmap[a.resource_id],
            "path": a.path.to_engine_format(),
            "unique": a.unique,
            "sorted": a.sorted.value,
            "vtype": a.value_type.value,
            "missing_values": [self._serde_engine_value(v) for v in a.missing_values]
        } for a in self.attrs]
        alignments = []
        for align in self.aligns:
            if isinstance(align, RangeAlignment):
                alignments.append({
                    "type": AlignmentType.range.value,
                    "source": aidmap[align.source],
                    "target": aidmap[align.target],
                    "aligned_dims": [{
                        "source": ad.source_idx,
                        "target": ad.target_idx
                    } for ad in align.aligned_steps]
                })
            elif isinstance(align, ValueAlignment):
                alignments.append({
                    "type": AlignmentType.value.value,
                    "source": aidmap[align.source],
                    "target": aidmap[align.target]
                })
            else:
                raise NotImplementedError()
        engine_sm: Dict[str, Any] = {
            "nodes": [],
            "edges": [],
            "prefixes": [(prefix, uri) for prefix, uri in self.sm.prefixes.items()]
        }
        nodes = {"cnodes": [], "dnodes": [], "lnodes": []}
        for node_id, node in self.sm.nodes.items():
            if isinstance(node, ClassNode):
                nodes['cnodes'].append({
                    "type": "class_node",
                    "node_id": node_id,
                    "rel_label": node.label,
                    "abs_label": self.sm.get_abs_iri(node.label)
                })
            elif isinstance(node, DataNode):
                nodes['dnodes'].append({
                    "type": "data_node",
                    "node_id": node_id,
                    "attr_id": aidmap[node.attr_id],
                    "data_type": node.data_type.value if node.data_type is not None else None,
                })
            elif isinstance(node, LiteralNode):
                nodes["lnodes"].append({
                    "type": "literal_node",
                    "node_id": node_id,
                    "val": self._serde_engine_value(node.value),
                    "data_type": node.data_type.value if node.data_type is not None else None,
                })
            else:
                raise NotImplementedError()
        for k in ["cnodes", "dnodes", "lnodes"]:
            for n in nodes[k]:
                engine_sm['nodes'].append(n)
        nidmap = {}
        for i, n in enumerate(engine_sm["nodes"]):
            nidmap[n['node_id']] = i
            n['node_id'] = i
        for eid, edge in self.sm.edges.items():
            engine_sm['edges'].append({
                "edge_id": eid,
                "source": nidmap[edge.source_id],
                "target": nidmap[edge.target_id],
                "rel_label": edge.label,
                "abs_label": self.sm.get_abs_iri(edge.label),
                "is_subject": edge.is_subject
            })
        edges_optional = [
            not edge.is_required
            for edge in self.sm.edges.values()
        ]
        return EngineFormat({
            "resources": resources,
            "preprocessing": preprocessing,
            "attributes": attributes,
            "alignments": alignments,
            "semantic_model": engine_sm
        }, edges_optional, ridmap, aidmap, nidmap)
def to_engine_format ( self ) -> EngineFormat : ridmap = OrderedDict ( ) for resource in self . resources : ridmap [ resource . id ] = len ( ridmap ) for pref in self . preprocessing : if pref . value . output is not None : ridmap [ pref . value . output ] = len ( ridmap ) aidmap = OrderedDict ( ) for attr in self . attrs : aidmap [ attr . id ] = len ( aidmap ) resources = [ ] for res in self . resources : resources . append ( { "type" : res . type . value } ) if isinstance ( res . prop , CSVProp ) : resources [ - 1 ] [ 'value' ] = { "resource_id" : ridmap [ res . id ] , "delimiter" : res . prop . delimiter } else : resources [ - 1 ] [ 'value' ] = ridmap [ res . id ] preprocessing = [ ] for pref in self . preprocessing : prepro = { "type" : pref . type . value , "resource_id" : ridmap [ pref . value . resource_id ] , "path" : pref . value . path . to_engine_format ( ) , "output" : ridmap [ pref . value . output ] if pref . value . output is not None else None , } if isinstance ( pref . value , PMap ) : prepro [ 'code' ] = pref . value . code prepro [ 'change_structure' ] = pref . value . change_structure elif isinstance ( pref . value , ( PFilter , PSplit ) ) : prepro [ 'code' ] = pref . value . code elif isinstance ( pref . value , RMap ) : prepro [ 'func_id' ] = { "t" : pref . value . func_id . value } else : raise NotImplementedError ( ) preprocessing . append ( prepro ) attributes = [ { "id" : aidmap [ a . id ] , "resource_id" : ridmap [ a . resource_id ] , "path" : a . path . to_engine_format ( ) , "unique" : a . unique , "sorted" : a . sorted . value , "vtype" : a . value_type . value , "missing_values" : [ self . _serde_engine_value ( v ) for v in a . missing_values ] } for a in self . attrs ] alignments = [ ] for align in self . aligns : if isinstance ( align , RangeAlignment ) : alignments . append ( { "type" : AlignmentType . range . value , "source" : aidmap [ align . source ] , "target" : aidmap [ align . target ] , "aligned_dims" : [ { "source" : ad . source_idx , "target" : ad . target_idx } for ad in align . aligned_steps ] } ) elif isinstance ( align , ValueAlignment ) : alignments . append ( { "type" : AlignmentType . value . value , "source" : aidmap [ align . source ] , "target" : aidmap [ align . target ] } ) else : raise NotImplementedError ( ) engine_sm : Dict [ str , Any ] = { "nodes" : [ ] , "edges" : [ ] , "prefixes" : [ ( prefix , uri ) for prefix , uri in self . sm . prefixes . items ( ) ] } nodes = { "cnodes" : [ ] , "dnodes" : [ ] , "lnodes" : [ ] } for node_id , node in self . sm . nodes . items ( ) : if isinstance ( node , ClassNode ) : nodes [ 'cnodes' ] . append ( { "type" : "class_node" , "node_id" : node_id , "rel_label" : node . label , "abs_label" : self . sm . get_abs_iri ( node . label ) } ) elif isinstance ( node , DataNode ) : nodes [ 'dnodes' ] . append ( { "type" : "data_node" , "node_id" : node_id , "attr_id" : aidmap [ node . attr_id ] , "data_type" : node . data_type . value if node . data_type is not None else None , } ) elif isinstance ( node , LiteralNode ) : nodes [ "lnodes" ] . append ( { "type" : "literal_node" , "node_id" : node_id , "val" : self . _serde_engine_value ( node . value ) , "data_type" : node . data_type . value if node . data_type is not None else None , } ) else : raise NotImplementedError ( ) for k in [ "cnodes" , "dnodes" , "lnodes" ] : for n in nodes [ k ] : engine_sm [ 'nodes' ] . append ( n ) nidmap = { } for i , n in enumerate ( engine_sm [ "nodes" ] ) : nidmap [ n [ 'node_id' ] ] = i n [ 'node_id' ] = i for eid , edge in self . sm . edges . items ( ) : engine_sm [ 'edges' ] . append ( { "edge_id" : eid , "source" : nidmap [ edge . source_id ] , "target" : nidmap [ edge . target_id ] , "rel_label" : edge . label , "abs_label" : self . sm . get_abs_iri ( edge . label ) , "is_subject" : edge . is_subject } ) edges_optional = [ not edge . is_required for edge in self . sm . edges . values ( ) ] return EngineFormat ( { "resources" : resources , "preprocessing" : preprocessing , "attributes" : attributes , "alignments" : alignments , "semantic_model" : engine_sm } , edges_optional , ridmap , aidmap , nidmap )
Turn this D-REPR configuration into the format that the engine can read
Turn this D - REPR configuration into the format that the engine can read
"""
        Turn this D-REPR configuration into the format that the engine can read
        :return:
        """ # map string id to incremental numbers (for resource id and attribute id)
None
self