Skip to content

SHOP Classes Returned by the SHOP API

cognite.powerops.client.shop.data_classes.shop_run.SHOPRun dataclass

This represents a single SHOP run.

A SHOP run is represented by an event in CDF. This class is a wrapper around the event.

Parameters:

Name Type Description Default
external_id str

The external ID of the SHOP run. This matches the external ID of the event in CDF.

 required
watercourse str

The watercourse of the SHOP run.

 required
start datetime | None

The start time of the SHOP run.

 required
end datetime | None

The end time of the SHOP run.

 required
shop_version str

The version of SHOP used for the SHOP run.

 required
user_id

The user ID of the user that triggered the SHOP run.

 required
Source code in cognite/powerops/client/shop/data_classes/shop_run.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
 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
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
@dataclass
class SHOPRun:
    """
    This represents a single SHOP run.

    A SHOP run is represented by an event in CDF. This class is a wrapper around the event.

    Args:
        external_id: The external ID of the SHOP run. This matches the external ID of the event in CDF.
        watercourse: The watercourse of the SHOP run.
        start: The start time of the SHOP run.
        end: The end time of the SHOP run.
        shop_version: The version of SHOP used for the SHOP run.
        user_id: The user ID of the user that triggered the SHOP run.
    """

    external_id: str
    watercourse: str
    data_set_id: int
    start: datetime | None
    end: datetime | None
    shop_version: str
    source: str | None
    _case_file_external_id: str | None
    _shop_files: list[SHOPFileReference]
    _client: CogniteClient = field(repr=False)
    _run_event_types: set[str] = field(init=False, default_factory=set)
    _manual_run: bool = True
    plants: Optional[str] = ""
    price_scenario: Optional[str] = ""

    @classmethod
    def load(cls, event: Event) -> Self:
        """
        Load a SHOP run from an event.

        Args:
            event: The event to load from.

        Returns:

        """
        metadata = event.metadata or {}
        if event.type != SHOPRunEvent.event_type:
            raise ValueError(f"Event {event.external_id} is not a SHOP run event!")

        if event._cognite_client is None:
            raise ValueError(f"Event {event.external_id} is not loaded with a cognite client!")

        if preprocessor_raw := metadata.get(SHOPRunEvent.preprocessor_data, None):
            preprocessor_data = json.loads(preprocessor_raw)
        else:
            preprocessor_data = {}

        return cls(
            external_id=event.external_id,
            data_set_id=event.data_set_id,
            watercourse=metadata.get(SHOPRunEvent.watercourse, ""),
            start=ms_to_datetime(event.start_time) if event.start_time else None,
            end=ms_to_datetime(event.end_time) if event.end_time else None,
            shop_version=preprocessor_data.get(SHOPRunEvent.shop_version, ""),
            _case_file_external_id=preprocessor_data.get(SHOPRunEvent.case_file, {}).get("external_id"),
            _shop_files=[SHOPFileReference.load(item) for item in preprocessor_data.get(SHOPRunEvent.shop_files, [])],
            _client=event._cognite_client,
            source=event.source,
            plants=metadata.get(SHOPRunEvent.plants),
            price_scenario=metadata.get(SHOPRunEvent.price_scenario),
        )

    @property
    def case_file_external_id(self) -> str:
        return self._case_file_external_id

    @property
    def shop_files_external_ids(self) -> list[dict[str, Any]]:
        return [shop_file.dump() for shop_file in self._shop_files]

    def as_cdf_event(self) -> Event:
        shop_version_spec = {SHOPRunEvent.shop_version: self.shop_version} if self.shop_version else {}
        return Event(
            external_id=self.external_id,
            type=SHOPRunEvent.event_type,
            subtype=SHOPRunEvent.event_type,
            data_set_id=self.data_set_id,
            start_time=datetime_to_ms(self.start),
            end_time=datetime_to_ms(self.end) if self.end else None,
            metadata={
                SHOPRunEvent.watercourse: self.watercourse,
                SHOPRunEvent.manual_run: self._manual_run,
                SHOPRunEvent.plants: self.plants,
                SHOPRunEvent.price_scenario: self.price_scenario,
                SHOPRunEvent.preprocessor_data: json.dumps(
                    {
                        **shop_version_spec,
                        SHOPRunEvent.case_file: {"external_id": self._case_file_external_id},
                        SHOPRunEvent.shop_files: [shop_file.dump() for shop_file in self._shop_files],
                    }
                ),
                # These are required by the SHOP container
                # In the functions, create_bid_process_event the end is by default 2 weeks into the future.
                SHOPRunEvent.file_source: self.source,
                SHOPRunEvent.shopstart: self.start.isoformat(),
                SHOPRunEvent.shopend: (self.start + timedelta(days=14)).isoformat(),
            },
            source=self.source,
        )

    def dump(self) -> dict[str, Any]:
        """
        Dump the SHOP run to a dictionary.

        Returns:
            A dictionary representation of the SHOP run.
        """
        return {
            "external_id": self.external_id,
            "watercourse": self.watercourse,
            "start": self.start,
            "end": self.end,
            "case_file_external_id": self.case_file_external_id,
            "shop_files_external_ids": self.shop_files_external_ids,
            "shop_version": self.shop_version,
        }

    def to_case(self):
        """Make a new SHOPCase from this SHOPRun."""
        return SHOPCase(
            data=self.get_case_file(),
            shop_files=self._shop_files.copy(),
            watercourse=self.watercourse.lower(),
            shop_version=self.shop_version,
        )

    def _download_file(self, external_id: str) -> str:
        content_bytes = self._client.files.download_bytes(external_id=external_id)
        try:
            return content_bytes.decode("utf-8")
        except UnicodeDecodeError:
            # SHOP sometimes writes files with Windows-1252 encoding (ASCII)
            # Trying to recover.
            return content_bytes.decode("Windows-1252")

    def get_case_file(self) -> str:
        """
        Get the case file for the SHOP run.

        Returns:
            The case file as a string.
        """
        return self._download_file(self._case_file_external_id)

    def get_shop_files(self) -> Iterable[str]:
        """
        Get the SHOP files for the SHOP run.

        Returns:
            A generator of strings, where each string is a SHOP file.
        """
        for shop_file in self._shop_files:
            yield self._download_file(shop_file.external_id)

    def get_pre_run_file(self) -> str:
        """
        Get the pre-run file generated by CogSHOP. This file is linked to the SHOP run event through the relationship
        "relationship_to.pre_run_file" and contains the input data for the SHOP run.

        Returns:
            The pre-run file as a string.
        """
        pre_run_files = self._client.relationships.list(
            source_external_ids=[self.external_id],
            source_types=["event"],
            target_types=["file"],
            labels=LabelFilter(contains_all=["relationship_to.pre_run_file"]),
        )
        if not pre_run_files:
            raise ValueError(f"No pre-run file found for SHOP run with external ID: {self.external_id}")
        if len(pre_run_files) > 1:
            raise ValueError(f"Multiple pre-run files found for SHOP run with external ID: {self.external_id}")

        return self._download_file(pre_run_files[0].target_external_id)

    def get_log_files(self) -> Iterable[tuple[str, str]]:
        """
        Get the log files for the SHOP run. This is, for exampl,e the CPLEX output.

        Returns:
            A generator of tuples of the form (external_id, file_content).
        """
        if self.check_status() is not SHOPRunStatus.SUCCESS:
            raise ValueError("Cannot retrieve result files for a SHOP run that has not finished successfully.")
        relationship = self._client.relationships.list(source_external_ids=[self.external_id], target_types=["file"])
        for rel in relationship:
            yield rel.target_external_id, self._download_file(rel.target_external_id)

    def check_status(self) -> SHOPRunStatus:
        """
        Check the status of the SHOP run.

        This does a request to CDF to check whether the SHOP run has finished or failed.

        Returns:
            A SHOP run status.

        """
        return self._check_status(update_events=True)

    def _check_status(self, update_events: bool = False) -> SHOPRunStatus:
        # If the run is finished or failed, we don't need to check the events.
        if SHOPProcessEvents.failed in self._run_event_types:
            return SHOPRunStatus.FAILED
        elif SHOPProcessEvents.finished in self._run_event_types:
            return SHOPRunStatus.SUCCESS

        if not update_events:
            return SHOPRunStatus.IN_PROGRESS

        self._update_run_events()
        return self._check_status(update_events=False)

    def get_failure_info(self) -> dict | None:
        relationships = self._client.relationships.list(
            source_external_ids=[self.external_id], target_types=["event"], fetch_resources=True
        )
        failure_events = [rel.target for rel in relationships if rel.target.type == SHOPProcessEvents.failed]
        if not failure_events:
            return None
        failure_events = sorted(failure_events, key=lambda event: -event.created_time)
        cogshop_failure_events = [event for event in failure_events if event.source == "CogShop"]
        argo_failure_events = [event for event in failure_events if event.source == "CogShopExitHandler"]
        error = None
        failures = None
        if cogshop_failure_events:
            error = cogshop_failure_events[0].metadata.get("errorStackTrace")
        if argo_failure_events:
            failures_json = argo_failure_events[0].metadata.get("failures")
            try:
                failures = json.loads(failures_json)
            except (json.JSONDecodeError, TypeError):
                failures = failures_json
        return {
            "error": error,
            "failures": failures,
        }

    def _update_run_events(self) -> None:
        relationships = self._client.relationships.list(
            source_external_ids=[self.external_id], target_types=["event"], fetch_resources=True
        )
        self._run_event_types |= {cast(Event, rel.target).type for rel in relationships}

    def _repr_html_(self) -> str:
        return pd.Series(self.dump()).to_frame().rename(columns={0: "Value"})._repr_html_()

    def __str__(self) -> str:
        """
        Returns a human-readable dump of the SHOP run.
        """
        return json.dumps(self.dump(), indent=4, default=str)

load(event) classmethod

Load a SHOP run from an event.

Parameters:

Name Type Description Default
event Event

The event to load from.

 required

Returns:

Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
 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
@classmethod
def load(cls, event: Event) -> Self:
    """
    Load a SHOP run from an event.

    Args:
        event: The event to load from.

    Returns:

    """
    metadata = event.metadata or {}
    if event.type != SHOPRunEvent.event_type:
        raise ValueError(f"Event {event.external_id} is not a SHOP run event!")

    if event._cognite_client is None:
        raise ValueError(f"Event {event.external_id} is not loaded with a cognite client!")

    if preprocessor_raw := metadata.get(SHOPRunEvent.preprocessor_data, None):
        preprocessor_data = json.loads(preprocessor_raw)
    else:
        preprocessor_data = {}

    return cls(
        external_id=event.external_id,
        data_set_id=event.data_set_id,
        watercourse=metadata.get(SHOPRunEvent.watercourse, ""),
        start=ms_to_datetime(event.start_time) if event.start_time else None,
        end=ms_to_datetime(event.end_time) if event.end_time else None,
        shop_version=preprocessor_data.get(SHOPRunEvent.shop_version, ""),
        _case_file_external_id=preprocessor_data.get(SHOPRunEvent.case_file, {}).get("external_id"),
        _shop_files=[SHOPFileReference.load(item) for item in preprocessor_data.get(SHOPRunEvent.shop_files, [])],
        _client=event._cognite_client,
        source=event.source,
        plants=metadata.get(SHOPRunEvent.plants),
        price_scenario=metadata.get(SHOPRunEvent.price_scenario),
    )

dump()

Dump the SHOP run to a dictionary.

Returns:

Type Description
dict[str, Any]

A dictionary representation of the SHOP run.
Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def dump(self) -> dict[str, Any]:
    """
    Dump the SHOP run to a dictionary.

    Returns:
        A dictionary representation of the SHOP run.
    """
    return {
        "external_id": self.external_id,
        "watercourse": self.watercourse,
        "start": self.start,
        "end": self.end,
        "case_file_external_id": self.case_file_external_id,
        "shop_files_external_ids": self.shop_files_external_ids,
        "shop_version": self.shop_version,
    }

to_case()

Make a new SHOPCase from this SHOPRun.

Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
178
179
180
181
182
183
184
185
def to_case(self):
    """Make a new SHOPCase from this SHOPRun."""
    return SHOPCase(
        data=self.get_case_file(),
        shop_files=self._shop_files.copy(),
        watercourse=self.watercourse.lower(),
        shop_version=self.shop_version,
    )

get_case_file()

Get the case file for the SHOP run.

Returns:

Type Description
str

The case file as a string.
Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
196
197
198
199
200
201
202
203
def get_case_file(self) -> str:
    """
    Get the case file for the SHOP run.

    Returns:
        The case file as a string.
    """
    return self._download_file(self._case_file_external_id)

get_shop_files()

Get the SHOP files for the SHOP run.

Returns:

Type Description
Iterable[str]

A generator of strings, where each string is a SHOP file.
Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
205
206
207
208
209
210
211
212
213
def get_shop_files(self) -> Iterable[str]:
    """
    Get the SHOP files for the SHOP run.

    Returns:
        A generator of strings, where each string is a SHOP file.
    """
    for shop_file in self._shop_files:
        yield self._download_file(shop_file.external_id)

get_pre_run_file()

Get the pre-run file generated by CogSHOP. This file is linked to the SHOP run event through the relationship "relationship_to.pre_run_file" and contains the input data for the SHOP run.

Returns:

Type Description
str

The pre-run file as a string.
Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def get_pre_run_file(self) -> str:
    """
    Get the pre-run file generated by CogSHOP. This file is linked to the SHOP run event through the relationship
    "relationship_to.pre_run_file" and contains the input data for the SHOP run.

    Returns:
        The pre-run file as a string.
    """
    pre_run_files = self._client.relationships.list(
        source_external_ids=[self.external_id],
        source_types=["event"],
        target_types=["file"],
        labels=LabelFilter(contains_all=["relationship_to.pre_run_file"]),
    )
    if not pre_run_files:
        raise ValueError(f"No pre-run file found for SHOP run with external ID: {self.external_id}")
    if len(pre_run_files) > 1:
        raise ValueError(f"Multiple pre-run files found for SHOP run with external ID: {self.external_id}")

    return self._download_file(pre_run_files[0].target_external_id)

get_log_files()

Get the log files for the SHOP run. This is, for exampl,e the CPLEX output.

Returns:

Type Description
Iterable[tuple[str, str]]

A generator of tuples of the form (external_id, file_content).
Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
236
237
238
239
240
241
242
243
244
245
246
247
def get_log_files(self) -> Iterable[tuple[str, str]]:
    """
    Get the log files for the SHOP run. This is, for exampl,e the CPLEX output.

    Returns:
        A generator of tuples of the form (external_id, file_content).
    """
    if self.check_status() is not SHOPRunStatus.SUCCESS:
        raise ValueError("Cannot retrieve result files for a SHOP run that has not finished successfully.")
    relationship = self._client.relationships.list(source_external_ids=[self.external_id], target_types=["file"])
    for rel in relationship:
        yield rel.target_external_id, self._download_file(rel.target_external_id)

check_status()

Check the status of the SHOP run.

This does a request to CDF to check whether the SHOP run has finished or failed.

Returns:

Type Description
SHOPRunStatus

A SHOP run status.
Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
249
250
251
252
253
254
255
256
257
258
259
def check_status(self) -> SHOPRunStatus:
    """
    Check the status of the SHOP run.

    This does a request to CDF to check whether the SHOP run has finished or failed.

    Returns:
        A SHOP run status.

    """
    return self._check_status(update_events=True)

__str__()

Returns a human-readable dump of the SHOP run.

Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
308
309
310
311
312
def __str__(self) -> str:
    """
    Returns a human-readable dump of the SHOP run.
    """
    return json.dumps(self.dump(), indent=4, default=str)

cognite.powerops.client.shop.data_classes.shop_run.SHOPRunList

Bases: UserList

This represents a list of SHOP runs.

Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
class SHOPRunList(UserList):
    """
    This represents a list of SHOP runs.
    """

    @overload
    def __getitem__(self, item: int) -> SHOPRun: ...

    @overload
    def __getitem__(self, item: slice) -> SHOPRunList: ...

    # The dunder implementations is to get proper type hints
    def __getitem__(self, item: int | slice) -> SHOPRunList | SHOPRun:
        if isinstance(item, slice):
            return type(self)(self.data[item])
        return self.data[item]

    def __iter__(self) -> Iterator[SHOPRun]:
        return super().__iter__()

    @classmethod
    def load(cls, events: Sequence[Event]) -> Self:
        """
        Load a SHOP run list from a list of events.

        Args:
            events: The events to load from. Must be SHOP run events.

        Returns:
            A SHOP run list.
        """
        return cls([SHOPRun.load(event) for event in events])

    def to_pandas(self) -> pd.DataFrame:
        """
        Convert the SHOP run list to a pandas DataFrame.

        Returns:
            A pandas DataFrame.
        """
        return pd.DataFrame([run.dump() for run in self.data])

    def _repr_html_(self) -> str:
        return self.to_pandas()._repr_html_()

load(events) classmethod

Load a SHOP run list from a list of events.

Parameters:

Name Type Description Default
events Sequence[Event]

The events to load from. Must be SHOP run events.

 required

Returns:

Type Description
Self

A SHOP run list.
Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
335
336
337
338
339
340
341
342
343
344
345
346
@classmethod
def load(cls, events: Sequence[Event]) -> Self:
    """
    Load a SHOP run list from a list of events.

    Args:
        events: The events to load from. Must be SHOP run events.

    Returns:
        A SHOP run list.
    """
    return cls([SHOPRun.load(event) for event in events])

to_pandas()

Convert the SHOP run list to a pandas DataFrame.

Returns:

Type Description
DataFrame

A pandas DataFrame.
Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
348
349
350
351
352
353
354
355
def to_pandas(self) -> pd.DataFrame:
    """
    Convert the SHOP run list to a pandas DataFrame.

    Returns:
        A pandas DataFrame.
    """
    return pd.DataFrame([run.dump() for run in self.data])

cognite.powerops.client.shop.data_classes.shop_run.SHOPRunStatus

Bases: StrEnum

Source code in cognite/powerops/client/shop/data_classes/shop_run.py 
26
27
28
29
class SHOPRunStatus(StrEnum):
    SUCCESS = "success"
    FAILED = "failed"
    IN_PROGRESS = "in_progress"