Skip to content

Demo

cognite.pygen.demo

SolarFarmAPM

Demo class for generating Solar Farm APM model in Python.

Parameters:

Name Type Description Default
space str

The space to deploy the APM model to.

 DEFAULT_SPACE
model_external_id str

The external ID of the APM model.

 'SolarFarmAPM'
model_version str

The version of the APM model.

 '1'
data_set_external_id str | None

The external ID of the data set to use for CDF Resources such as Time Series and Files.

 DEFAULT_DATA_SET
Source code in cognite/pygen/demo/_solar_apm.py 
 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
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
class SolarFarmAPM:
    """
    Demo class for generating Solar Farm APM model in Python.

    Args:
        space: The space to deploy the APM model to.
        model_external_id: The external ID of the APM model.
        model_version: The version of the APM model.
        data_set_external_id: The external ID of the data set to use for CDF Resources such as Time Series and Files.
    """

    def __init__(
        self,
        space: str = DEFAULT_SPACE,
        model_external_id: str = "SolarFarmAPM",
        model_version: str = "1",
        data_set_external_id: str | None = DEFAULT_DATA_SET,
    ):
        self._graphql = (_DATA_FOLDER / "model.graphql").read_text()
        self._data_model_id = DataModelId(space=space, external_id=model_external_id, version=model_version)
        self._echo: Callable[[str], None] = print
        self._data_model: DataModel[View] | None = None
        self._data_set_external_id = data_set_external_id

    def display(self):
        """
        Display the model in GraphQL format in a Jupyter notebook environment.
        """
        try:
            from IPython.display import Markdown, display

            display(Markdown(f"### {self._data_model_id}:\n ```\n{self._graphql}```"))
        except ImportError:
            print(self._graphql)

    def create(self, client: CogniteClient, populate: bool = True) -> Any:
        """
        Deploy, populate (optional) and generate a Python SDK for the APM demo model.

        This method will do the following three steps

        1. Deploy the APM model to the CDF project the client is connected to.
        2. Generate a Python SDK for the APM model.
        3. Populate the APM model with mock data included in pygen.

        Args:
            client: Connected CogniteClient
            populate: Whether to populate the APM model with mock data included in pygen.

        Returns:
            An instantiated SDK client for the APM model.
        """
        self._data_model = self.deploy(client)
        self._echo("✅  Data Model Ready!")
        if populate:
            self.populate(client)
            self._echo("✅  Population Complete!")
        client = self.generate_sdk(client)
        self._echo("✅  SDK Generated!")
        return client

    def deploy(self, client: CogniteClient) -> DataModel[View]:
        """
        Deploy the APM model to the CDF project the client is connected to.

        Args:
            client: Connected CogniteClient

        Returns:
            The DMS representation of the deployed model.
        """
        space = client.data_modeling.spaces.retrieve(self._data_model_id.space)
        if not space:
            space_apply = SpaceApply(
                space=self._data_model_id.space,
                name=self._data_model_id.space,
                description="This space was created by pygen to host demo data models.",
            )
            space = client.data_modeling.spaces.apply(space_apply)
            self._echo(f"Created space {space.space}")

        retrieved = client.data_modeling.data_models.retrieve(self._data_model_id, inline_views=True)
        if retrieved:
            self._echo(f"Data model {self._data_model_id} already exists, skipping deployment")
            return retrieved.latest_version()

        _ = client.data_modeling.graphql.apply_dml(
            self._data_model_id,
            self._graphql,
            name=self._data_model_id.external_id,
            description="This data model was created by pygen for demo purposes.",
        )
        self._echo(f"Deployed data model {self._data_model_id}")
        return client.data_modeling.data_models.retrieve(self._data_model_id, inline_views=True).latest_version()

    def populate(self, client: CogniteClient):
        if self._data_model is None:
            raise ValueError("Cannot populate model before deploying it, please call deploy() first")
        loader = self._create_csv_loader(client)
        loader.populate(client)

    def _create_csv_loader(self, client: CogniteClient) -> CSVLoader:
        if self._data_model is None:
            raise ValueError("Cannot populate model before deploying it, please call deploy() first")
        data_set_id = self._data_set_id(client)
        loader = CSVLoader(
            _DATA_FOLDER,
            self._echo,
            data_set_id,
            self._data_model,
        )
        return loader

    def _data_set_id(self, client: CogniteClient) -> int | None:
        if self._data_set_external_id is None:
            return None
        dataset = client.data_sets.retrieve(external_id=self._data_set_external_id)
        if dataset:
            return dataset.id

        try:
            new_dataset = cast(
                DataSet,
                client.data_sets.create(
                    DataSet(
                        external_id=self._data_set_external_id,
                        name=self._data_set_external_id,
                        description="This data set was created by pygen for demo purposes.",
                    )
                ),
            )
        except CogniteAPIError as e:
            self._echo(f"Failed to create data set {self._data_set_external_id}: {e}")
            return None
        self._echo(f"Created data set {new_dataset.external_id}")
        return new_dataset.id

    def generate_sdk(self, client: CogniteClient) -> Any:
        """
        Generate a Python SDK for the APM demo model.

        !!! warning "Assumes APM Model Deployed"
            This method assumes the APM model has been deployed to the CDF project the client is connected to.

        Args:
            client: Connected CogniteClient

        Returns:
            An instantiated SDK client for the APM model.

        """
        if self._data_model is None:
            return generate_sdk_notebook(self._data_model_id, client)
        return generate_sdk_notebook(self._data_model, client)

    def clean(self, client: CogniteClient, delete_space: bool = True, auto_confirm: bool = False):
        """
        Clean the Solar Farm Model from the CDF project the client is connected to.

        This means removing the data model, views, and containers generated by pygen, as well as
        all CDF resources used in the population of the model.

        Args:
            client: Connected CogniteClient
            delete_space: Whether to try to delete the space the APM model was deployed to. This will only work if the
                          space does not contain any other data models, views or containers.
            auto_confirm: Whether to skip the confirmation prompt.

        """
        data_models = client.data_modeling.data_models.retrieve(self._data_model_id, inline_views=True)
        if not data_models:
            self._echo(f"Data model {self._data_model_id} does not exist, skipping clean")
            return
        data_model = data_models.latest_version()
        view_ids = list({view.as_id() for view in data_model.views})
        container_ids = list(
            {
                prop.container
                for view in data_model.views
                for prop in view.properties.values()
                if isinstance(prop, MappedProperty)
            }
        )
        self._data_model = data_model
        loader = self._create_csv_loader(client)
        if not auto_confirm:
            self._echo(f"About to delete data model {self._data_model_id}")
            self._echo(f"About to delete views {view_ids}")
            self._echo(f"About to delete containers {container_ids} along with all nodes and edges")
            if delete_space:
                self._echo(f"About to delete space {self._data_model_id.space}")
            answer = input("Are you sure you want to continue? [y/N] ")
            if not isinstance(answer, str) and hasattr(answer, "result"):
                # PyodideFuture
                self._echo(
                    "The parameter auto_confirm is not supported in a Pyodide environment,"
                    "please set auto_confirm=True"
                )
                answer = "n"

            if not answer.lower().startswith("y"):
                self._echo("Aborting")
                return

        loader.clean(client)
        client.data_modeling.data_models.delete(self._data_model_id)
        self._echo(f"Deleted data model {self._data_model_id}")
        client.data_modeling.views.delete(view_ids)
        self._echo(f"Deleted views {view_ids}")
        client.data_modeling.containers.delete(container_ids)
        self._echo(f"Deleted containers {container_ids}")
        if delete_space:
            client.data_modeling.spaces.delete(self._data_model_id.space)
            self._echo(f"Deleted space {self._data_model_id.space}")

clean(client, delete_space=True, auto_confirm=False)

Clean the Solar Farm Model from the CDF project the client is connected to.

This means removing the data model, views, and containers generated by pygen, as well as all CDF resources used in the population of the model.

Parameters:

Name Type Description Default
client CogniteClient

Connected CogniteClient

 required
delete_space bool

Whether to try to delete the space the APM model was deployed to. This will only work if the space does not contain any other data models, views or containers.

 True
auto_confirm bool

Whether to skip the confirmation prompt.

 False
Source code in cognite/pygen/demo/_solar_apm.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def clean(self, client: CogniteClient, delete_space: bool = True, auto_confirm: bool = False):
    """
    Clean the Solar Farm Model from the CDF project the client is connected to.

    This means removing the data model, views, and containers generated by pygen, as well as
    all CDF resources used in the population of the model.

    Args:
        client: Connected CogniteClient
        delete_space: Whether to try to delete the space the APM model was deployed to. This will only work if the
                      space does not contain any other data models, views or containers.
        auto_confirm: Whether to skip the confirmation prompt.

    """
    data_models = client.data_modeling.data_models.retrieve(self._data_model_id, inline_views=True)
    if not data_models:
        self._echo(f"Data model {self._data_model_id} does not exist, skipping clean")
        return
    data_model = data_models.latest_version()
    view_ids = list({view.as_id() for view in data_model.views})
    container_ids = list(
        {
            prop.container
            for view in data_model.views
            for prop in view.properties.values()
            if isinstance(prop, MappedProperty)
        }
    )
    self._data_model = data_model
    loader = self._create_csv_loader(client)
    if not auto_confirm:
        self._echo(f"About to delete data model {self._data_model_id}")
        self._echo(f"About to delete views {view_ids}")
        self._echo(f"About to delete containers {container_ids} along with all nodes and edges")
        if delete_space:
            self._echo(f"About to delete space {self._data_model_id.space}")
        answer = input("Are you sure you want to continue? [y/N] ")
        if not isinstance(answer, str) and hasattr(answer, "result"):
            # PyodideFuture
            self._echo(
                "The parameter auto_confirm is not supported in a Pyodide environment,"
                "please set auto_confirm=True"
            )
            answer = "n"

        if not answer.lower().startswith("y"):
            self._echo("Aborting")
            return

    loader.clean(client)
    client.data_modeling.data_models.delete(self._data_model_id)
    self._echo(f"Deleted data model {self._data_model_id}")
    client.data_modeling.views.delete(view_ids)
    self._echo(f"Deleted views {view_ids}")
    client.data_modeling.containers.delete(container_ids)
    self._echo(f"Deleted containers {container_ids}")
    if delete_space:
        client.data_modeling.spaces.delete(self._data_model_id.space)
        self._echo(f"Deleted space {self._data_model_id.space}")

create(client, populate=True)

Deploy, populate (optional) and generate a Python SDK for the APM demo model.

This method will do the following three steps

  1. Deploy the APM model to the CDF project the client is connected to.
  2. Generate a Python SDK for the APM model.
  3. Populate the APM model with mock data included in pygen.

Parameters:

Name Type Description Default
client CogniteClient

Connected CogniteClient

 required
populate bool

Whether to populate the APM model with mock data included in pygen.

 True

Returns:

Type Description
Any

An instantiated SDK client for the APM model.
Source code in cognite/pygen/demo/_solar_apm.py 
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
def create(self, client: CogniteClient, populate: bool = True) -> Any:
    """
    Deploy, populate (optional) and generate a Python SDK for the APM demo model.

    This method will do the following three steps

    1. Deploy the APM model to the CDF project the client is connected to.
    2. Generate a Python SDK for the APM model.
    3. Populate the APM model with mock data included in pygen.

    Args:
        client: Connected CogniteClient
        populate: Whether to populate the APM model with mock data included in pygen.

    Returns:
        An instantiated SDK client for the APM model.
    """
    self._data_model = self.deploy(client)
    self._echo("✅  Data Model Ready!")
    if populate:
        self.populate(client)
        self._echo("✅  Population Complete!")
    client = self.generate_sdk(client)
    self._echo("✅  SDK Generated!")
    return client

deploy(client)

Deploy the APM model to the CDF project the client is connected to.

Parameters:

Name Type Description Default
client CogniteClient

Connected CogniteClient

 required

Returns:

Type Description
DataModel[View]

The DMS representation of the deployed model.
Source code in cognite/pygen/demo/_solar_apm.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def deploy(self, client: CogniteClient) -> DataModel[View]:
    """
    Deploy the APM model to the CDF project the client is connected to.

    Args:
        client: Connected CogniteClient

    Returns:
        The DMS representation of the deployed model.
    """
    space = client.data_modeling.spaces.retrieve(self._data_model_id.space)
    if not space:
        space_apply = SpaceApply(
            space=self._data_model_id.space,
            name=self._data_model_id.space,
            description="This space was created by pygen to host demo data models.",
        )
        space = client.data_modeling.spaces.apply(space_apply)
        self._echo(f"Created space {space.space}")

    retrieved = client.data_modeling.data_models.retrieve(self._data_model_id, inline_views=True)
    if retrieved:
        self._echo(f"Data model {self._data_model_id} already exists, skipping deployment")
        return retrieved.latest_version()

    _ = client.data_modeling.graphql.apply_dml(
        self._data_model_id,
        self._graphql,
        name=self._data_model_id.external_id,
        description="This data model was created by pygen for demo purposes.",
    )
    self._echo(f"Deployed data model {self._data_model_id}")
    return client.data_modeling.data_models.retrieve(self._data_model_id, inline_views=True).latest_version()

display()

Display the model in GraphQL format in a Jupyter notebook environment.

Source code in cognite/pygen/demo/_solar_apm.py 
49
50
51
52
53
54
55
56
57
58
def display(self):
    """
    Display the model in GraphQL format in a Jupyter notebook environment.
    """
    try:
        from IPython.display import Markdown, display

        display(Markdown(f"### {self._data_model_id}:\n ```\n{self._graphql}```"))
    except ImportError:
        print(self._graphql)

generate_sdk(client)

Generate a Python SDK for the APM demo model.

Assumes APM Model Deployed

This method assumes the APM model has been deployed to the CDF project the client is connected to.

Parameters:

Name Type Description Default
client CogniteClient

Connected CogniteClient

 required

Returns:

Type Description
Any

An instantiated SDK client for the APM model.
Source code in cognite/pygen/demo/_solar_apm.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def generate_sdk(self, client: CogniteClient) -> Any:
    """
    Generate a Python SDK for the APM demo model.

    !!! warning "Assumes APM Model Deployed"
        This method assumes the APM model has been deployed to the CDF project the client is connected to.

    Args:
        client: Connected CogniteClient

    Returns:
        An instantiated SDK client for the APM model.

    """
    if self._data_model is None:
        return generate_sdk_notebook(self._data_model_id, client)
    return generate_sdk_notebook(self._data_model, client)