Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 42 additions & 0 deletions python/examples/print_latest_icing.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
import logging
from datetime import datetime, timedelta, timezone

from heimdall_api_client.client import HeimdallApiClient

logging.basicConfig(level=logging.WARN)

client = HeimdallApiClient(
client_id="your_client_id",
client_secret="your_client_secret",
)

assets = client.get_assets()
grid_owner = assets.data.grid_owners[0]

print(f"\nGrid Owner: {grid_owner.name}\n")

for facility in grid_owner.facilities:
line = facility.line
if not line:
print(f"Facility: {facility.name} has no lines.\n")
continue

line_id = line.id
print(f"Line: {line.name} (ID: {line_id})")

since = datetime.now(timezone.utc) - timedelta(minutes=30)
latest_icing_response = client.get_latest_icing(line_id=line_id, since=since)
latest_icing = latest_icing_response.data.icing

max_icing = latest_icing.max_
print(
f" Max Ice Weight: {max_icing.ice_weight.value} {max_icing.ice_weight.unit} (span phase {max_icing.ice_weight.span_phase_id}, {max_icing.ice_weight.timestamp})"
)
print(
f" Max Tension: {max_icing.tension.value} {max_icing.tension.unit} (span phase {max_icing.tension.span_phase_id}, {max_icing.tension.timestamp})"
)
print(
" Max Tension %: "
f"{max_icing.tension_percentage_of_break_strength.value} {max_icing.tension_percentage_of_break_strength.unit} "
f"(span phase {max_icing.tension_percentage_of_break_strength.span_phase_id}, {max_icing.tension_percentage_of_break_strength.timestamp})"
)
39 changes: 26 additions & 13 deletions python/heimdall_api_client/assets_api_client/models/assets.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,56 +4,69 @@
from attrs import define as _attrs_define
from attrs import field as _attrs_field


if TYPE_CHECKING:
from ..models.grid_owner import GridOwner
from ..models.grid_owner import GridOwner





T = TypeVar("T", bound="Assets")



@_attrs_define
class Assets:
"""
Attributes:
grid_owners (list['GridOwner']): List of grid owners the API consumer has access to.
"""
Attributes:
grid_owners (list['GridOwner']): List of grid owners the API consumer has access to.
"""

grid_owners: list["GridOwner"]
grid_owners: list['GridOwner']
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)





def to_dict(self) -> dict[str, Any]:
grid_owners = []
for grid_owners_item_data in self.grid_owners:
grid_owners_item = grid_owners_item_data.to_dict()
grid_owners.append(grid_owners_item)




field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"grid_owners": grid_owners,
}
)
field_dict.update({
"grid_owners": grid_owners,
})

return field_dict



@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
from ..models.grid_owner import GridOwner

d = dict(src_dict)
grid_owners = []
_grid_owners = d.pop("grid_owners")
for grid_owners_item_data in _grid_owners:
for grid_owners_item_data in (_grid_owners):
grid_owners_item = GridOwner.from_dict(grid_owners_item_data)



grid_owners.append(grid_owners_item)


assets = cls(
grid_owners=grid_owners,
)


assets.additional_properties = d
return assets

Expand Down
21 changes: 21 additions & 0 deletions python/heimdall_api_client/client.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
import logging
from typing import List, Optional
from uuid import UUID
import datetime
from heimdall_api_client.grid_insights_api_client.models.unit_system import UnitSystem
from heimdall_api_client.auth import AuthService
from heimdall_api_client.assets import get_assets
from heimdall_api_client.capacity_monitoring import (
Expand Down Expand Up @@ -132,3 +134,22 @@ def get_latest_current(self, line_id: UUID):
from heimdall_api_client.grid_insights import get_latest_current

return get_latest_current(client=self._get_authenticated_client(), line_id=line_id, region=self._get_region())

def get_latest_icing(
self,
line_id: UUID,
unit_system: UnitSystem | str | None = None,
since: datetime.datetime | None = None,
):
"""
Returns the latest icing measurements for a given line.
"""
from heimdall_api_client.grid_insights import get_latest_icing

return get_latest_icing(
client=self._get_authenticated_client(),
line_id=line_id,
region=self._get_region(),
unit_system=unit_system,
since=since,
)
34 changes: 33 additions & 1 deletion python/heimdall_api_client/grid_insights.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
from uuid import UUID
import datetime
from heimdall_api_client.assets_api_client.client import AuthenticatedClient
from heimdall_api_client.grid_insights_api_client.models.unit_system import UnitSystem
from heimdall_api_client.grid_insights_api_client.types import UNSET


def get_latest_conductor_temperature(client: AuthenticatedClient, line_id: UUID, region: str):
Expand All @@ -9,7 +12,7 @@ def get_latest_conductor_temperature(client: AuthenticatedClient, line_id: UUID,

response = get_latest_conductor_temperature.sync_detailed(client=client, line_id=line_id, x_region=region)
if response.status_code != 200:
raise Exception(f"Error fetching latest conductor temperature: {response.status_code} {response.text}")
raise Exception(f"Error fetching latest conductor temperature: {response.status_code} {response.content}")
return response.parsed


Expand All @@ -22,3 +25,32 @@ def get_latest_current(client: AuthenticatedClient, line_id: UUID, region: str):
if response.status_code != 200:
raise Exception(f"Error fetching latest current: {response.status_code} {response.text}")
return response.parsed


def get_latest_icing(
client: AuthenticatedClient,
line_id: UUID,
region: str,
unit_system: UnitSystem | str | None = None,
since: datetime.datetime | None = None,
):
from heimdall_api_client.grid_insights_api_client.api.line import (
grid_insights_v1_lines_get_latest_icing as get_latest_icing,
)

unit_system_value = UNSET
if unit_system is not None:
unit_system_value = unit_system if isinstance(unit_system, UnitSystem) else UnitSystem(unit_system)

since_value = UNSET if since is None else since

response = get_latest_icing.sync_detailed(
client=client,
line_id=line_id,
x_region=region,
unit_system=unit_system_value,
since=since_value,
)
if response.status_code != 200:
raise Exception(f"Error fetching latest icing: {response.status_code} {response.text}")
return response.parsed
Loading