Skip to content

CosmoTech_Acceleration_Library.Modelops.core.io.model_exporter

ModelExporter

Bases: ExportableGraphHandler

Model Exporter for cached data

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_exporter.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class ModelExporter(ExportableGraphHandler):
    """
    Model Exporter for cached data
    """

    def __init__(self, host: str, port: int, name: str, version: int, password: str = None, export_dir: str = "/"):
        super().__init__(host=host, port=port, name=name, version=version, password=password, export_dir=export_dir)
        self.mr = ModelReader(host=host, port=port, name=name, password=password, version=version)

    @do_if_graph_exist
    def export_all_twins(self):
        """
        Export all twins
        :return: Csv files containing all twin instances exported into {export_dir} folder named by twin type
        """
        logger.debug("Start exporting twins...")
        logger.debug("Get twin types...")
        get_types_start = time.time()
        twin_names = self.mr.get_twin_types()
        get_types_end = time.time() - get_types_start
        logger.debug(f"Get twin types took {get_types_end} s")

        for twin_name in twin_names:
            logger.debug(f"Get twin info for type {twin_name} ...")

            get_twin_info_start = time.time()
            twin_results = self.mr.get_twins_by_type(twin_name)
            get_twin_info_end = time.time() - get_twin_info_start

            logger.debug(f"Get twin info for type {twin_name} took {get_twin_info_end} s")
            logger.debug(f"Export twin info for type {twin_name} ...")

            export_twin_info_start = time.time()
            CsvWriter.write_twin_data(self.export_dir, twin_name, twin_results)
            export_twin_info_end = time.time() - export_twin_info_start

            logger.debug(f"Export twin info for type {twin_name} took {export_twin_info_end} s")
            logger.debug(f"Twins exported :{twin_name}")
        logger.debug("... End exporting twins")

    @do_if_graph_exist
    def export_all_relationships(self):
        """
        Export all relationships
        :return: Csv files containing all relationship instances exported into {export_dir}
        folder named by relationship type
        """
        logger.debug("Start exporting relationships...")
        logger.debug("Get relationship types...")
        get_relationship_types_start = time.time()
        relationship_names = self.mr.get_relationship_types()
        get_relationship_types_end = time.time() - get_relationship_types_start
        logger.debug(f"Get relationship types took {get_relationship_types_end} s")

        for relationship_name in relationship_names:
            logger.debug(f"Get relationship info for type {relationship_name} ...")

            get_relationship_info_start = time.time()
            headers = self.mr.get_relationship_properties_by_type(relationship_name)
            relationship_result = self.mr.get_relationships_by_type(relationship_name)

            get_relationship_info_end = time.time() - get_relationship_info_start
            logger.debug(f"Get relationship info for type {relationship_name} took {get_relationship_info_end} s")
            logger.debug(f"Export relationship info for type {relationship_name} ...")

            export_relationship_info_start = time.time()
            CsvWriter.write_relationship_data(self.export_dir, relationship_name, relationship_result)
            export_relationship_info_end = time.time() - export_relationship_info_start

            logger.debug(f"Export relationship info for type {relationship_name} took {export_relationship_info_end} s")
            logger.debug(f"Relationships exported :{relationship_name}")
        logger.debug("... End exporting relationships")

    @do_if_graph_exist
    def export_all_data(self):
        """
        Export all data
        :return: a bunch of csv files corresponding to graph data
        """
        self.export_all_twins()
        self.export_all_relationships()

export_all_data()

Export all data :return: a bunch of csv files corresponding to graph data

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_exporter.py 
87
88
89
90
91
92
93
94
@do_if_graph_exist
def export_all_data(self):
    """
    Export all data
    :return: a bunch of csv files corresponding to graph data
    """
    self.export_all_twins()
    self.export_all_relationships()

export_all_relationships()

Export all relationships :return: Csv files containing all relationship instances exported into {export_dir} folder named by relationship type

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_exporter.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@do_if_graph_exist
def export_all_relationships(self):
    """
    Export all relationships
    :return: Csv files containing all relationship instances exported into {export_dir}
    folder named by relationship type
    """
    logger.debug("Start exporting relationships...")
    logger.debug("Get relationship types...")
    get_relationship_types_start = time.time()
    relationship_names = self.mr.get_relationship_types()
    get_relationship_types_end = time.time() - get_relationship_types_start
    logger.debug(f"Get relationship types took {get_relationship_types_end} s")

    for relationship_name in relationship_names:
        logger.debug(f"Get relationship info for type {relationship_name} ...")

        get_relationship_info_start = time.time()
        headers = self.mr.get_relationship_properties_by_type(relationship_name)
        relationship_result = self.mr.get_relationships_by_type(relationship_name)

        get_relationship_info_end = time.time() - get_relationship_info_start
        logger.debug(f"Get relationship info for type {relationship_name} took {get_relationship_info_end} s")
        logger.debug(f"Export relationship info for type {relationship_name} ...")

        export_relationship_info_start = time.time()
        CsvWriter.write_relationship_data(self.export_dir, relationship_name, relationship_result)
        export_relationship_info_end = time.time() - export_relationship_info_start

        logger.debug(f"Export relationship info for type {relationship_name} took {export_relationship_info_end} s")
        logger.debug(f"Relationships exported :{relationship_name}")
    logger.debug("... End exporting relationships")

export_all_twins()

Export all twins :return: Csv files containing all twin instances exported into {export_dir} folder named by twin type

Source code in CosmoTech_Acceleration_Library/Modelops/core/io/model_exporter.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@do_if_graph_exist
def export_all_twins(self):
    """
    Export all twins
    :return: Csv files containing all twin instances exported into {export_dir} folder named by twin type
    """
    logger.debug("Start exporting twins...")
    logger.debug("Get twin types...")
    get_types_start = time.time()
    twin_names = self.mr.get_twin_types()
    get_types_end = time.time() - get_types_start
    logger.debug(f"Get twin types took {get_types_end} s")

    for twin_name in twin_names:
        logger.debug(f"Get twin info for type {twin_name} ...")

        get_twin_info_start = time.time()
        twin_results = self.mr.get_twins_by_type(twin_name)
        get_twin_info_end = time.time() - get_twin_info_start

        logger.debug(f"Get twin info for type {twin_name} took {get_twin_info_end} s")
        logger.debug(f"Export twin info for type {twin_name} ...")

        export_twin_info_start = time.time()
        CsvWriter.write_twin_data(self.export_dir, twin_name, twin_results)
        export_twin_info_end = time.time() - export_twin_info_start

        logger.debug(f"Export twin info for type {twin_name} took {export_twin_info_end} s")
        logger.debug(f"Twins exported :{twin_name}")
    logger.debug("... End exporting twins")