wd_orca_api

This is a wrapper to communicate with the World Direct ORCA API, both Finish and Swedish instances.

WDOrcaAPI

Source code in physical_operations_utils/wd_orca_utils/WDOrcaAPI.py

class WDOrcaAPI:

    def __init__(self, country: str):
        """Initialize a World Direct Orca API client for a specific asset.

        Args:
            country (str): The country for which to fetch data (eg. "SE", "FI").

        Example:
            ```python
            from physical_operations_utils.wd_orca_utils.WDOrcaAPI import WDOrcaAPI

            api = WDOrcaAPI("SE")
            device_id = 23  # Hultema -> 32
            signals_dict = {
                249: "kw",
                126: "m/s",
                101: "boolean",
            }
            df = api.fetch_monitoring_data(
                device_id=device_id,
                signals=signals_dict.keys(),
                start_time_lb_utc=pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=2),
                end_time_lb_utc=pd.Timestamp.now(tz="UTC"),
            )
            df_dict = api.reformat(df, signals_dict=signals_dict, variable_id="735999100016826260", resolution_seconds=900)
            for signal_id, df_signal in df_dict.items():
                print(f"Signal ID: {signal_id}")
                print(df_signal)
            ```
        """
        self.environment = setup_environment()
        self.country = country.upper()

        if self.country == "SE":
            self.base_url = (
                SE_BASE_URL_PROD
                if self.environment.lower() == "prod"
                else SE_BASE_URL_NONPROD
            )
            self.authorize_url = (
                SE_AUTHORIZE_URL_PROD
                if self.environment.lower() == "prod"
                else SE_AUTHORIZE_URL_NONPROD
            )
            self.monitoring_url = self.base_url + "api/monitoring/devices/"
        elif self.country == "FI":
            self.base_url = (
                FI_BASE_URL_PROD
                if self.environment.lower() == "prod"
                else FI_BASE_URL_NONPROD
            )
            self.authorize_url = (
                FI_AUTHORIZE_URL_PROD
                if self.environment.lower() == "prod"
                else FI_AUTHORIZE_URL_NONPROD
            )
            self.monitoring_url = self.base_url + "admin/devices/monitoring/"
        else:
            raise ValueError(f"Unsupported country: {country}")

        self._authorize()

    def _authorize(self):
        """
        Helper method to create the authorization header for API requests.
        This method retrieves the necessary credentials from Azure Key Vault and obtains an access token from the World Direct API.
        The access token is then stored in the instance variable `_headers` for use in subsequent API calls.
        """
        data = {
            "grant_type": "client_credentials",
        }
        username = (
            get_keys_yaml_file()
            .get(f"wd_full_access_client_{self.country}")
            .get("username")
        )
        secret_name = (
            get_keys_yaml_file()
            .get(f"wd_full_access_client_{self.country}")
            .get("secret")
        )
        secret = get_secret(secret_name)

        auth = (
            username,
            secret,
        )

        response = requests.post(self.authorize_url, data=data, auth=auth)
        if response.status_code == 200:
            self._headers = {
                "Authorization": f"Bearer {response.json()['access_token']}"
            }
        else:
            raise Exception(
                f"Failed to authorize with World Direct API. Status code: {response.status_code}, Response: {response.text}"
            )

    def fetch_monitoring_data(
        self,
        device_id: int,
        signals: list,
        start_time_lb_utc: pd.Timestamp = None,
        end_time_lb_utc: pd.Timestamp = None,
    ) -> pd.DataFrame:
        """Returns monitoring data from the World Direct API for a given signal, device and time range.

        Args:
            start_time_lb_utc (pd.Timestamp, optional): Start time in UTC. Default to 10 minutes before end_time_lb_utc.
            end_time_lb_utc (pd.Timestamp, optional): End time in UTC. Defaults to current time.
            device_id (int): The ID of the device. Stored in Ancillary Services Database [dbo.asset]
            signals (list): List of signals to fetch (e.g., [101, 331]).

        Returns:
            pd.DataFrame: DataFrame containing the monitoring data. (timestamp_utc, variable_value, signal_id)
        """

        # If start_time_lb_utc and end_time_lb_utc are not specified, set default to last 10 minutes
        if end_time_lb_utc is None:
            end_time_lb_utc = pd.Timestamp.now(tz="UTC")
        if start_time_lb_utc is None:
            start_time_lb_utc = end_time_lb_utc - pd.Timedelta(minutes=10)

        assert (
            end_time_lb_utc > start_time_lb_utc
        ), "end_time_lb_utc must be a later time than start_time_lb_utc."

        res = requests.get(
            url=self.monitoring_url + f"{device_id}/signals/download",
            headers=self._headers,
            params={
                "deviceId": device_id,
                "From": start_time_lb_utc.isoformat(),
                "To": end_time_lb_utc.isoformat(),
                "Signals": signals,
            },
        )

        if res.status_code != 200:
            raise Exception(
                f"Failed to fetch monitoring data from {self.monitoring_url}. Status code: {res.status_code}"
            )

        # The Swedish instance of the API uses "id" while the Finish uses "signalId"
        signal_id_name = "id" if self.country == "SE" else "signalId"

        # Create the DataFrame from the response (timestamp_utc, variable_value, signal_id).
        df = pd.json_normalize(
            res.json()["signals"],
            record_path="records",
            meta=[signal_id_name],
        )

        if df.empty:
            raise Exception(
                "WD API error: No monitoring data found for the given parameters."
            )

        df.rename(
            columns={
                "timestamp": "timestamp_lb_utc",
                "value": "variable_value",
                signal_id_name: "signal_id",
            },
            inplace=True,
        )
        df["timestamp_lb_utc"] = pd.to_datetime(df["timestamp_lb_utc"], utc=True)

        return df

    def reformat(
        self,
        df: pd.DataFrame,
        signals_dict: dict,
        variable_id: str,
        resolution_seconds: int = 900,
    ) -> dict[int, pd.DataFrame]:
        """Reformats the DataFrame from the World Direct API to a standardized CTDB format.

        Args:
            df (pd.DataFrame): The input DataFrame containing columns 'timestamp_lb_utc', 'variable_value', and 'signal_id'.
            signals_dict (dict): Dictionary mapping signal IDs to signal unit.
            variable_id (str): The ID of the variable for which to reformat data.
            resolution_seconds (int, optional): The desired time resolution in seconds (e.g., 900 for 15 minutes, 3600 for 1 hour).

        Returns:
            dict[int, pd.DataFrame]: A dictionary where the keys are signal IDs and the values are DataFrames reformatted to the CommonTradingData format, containing columns 'start_time_lb_utc', 'stop_time_lb_utc', 'db_updated_utc', 'variable_id', 'variable_value', 'variable_unit', and 'resolution_seconds'.
        """

        assert (
            "timestamp_lb_utc" in df.columns
            and "variable_value" in df.columns
            and "signal_id" in df.columns
        ), "Input DataFrame must contain 'timestamp_lb_utc', 'variable_value', and 'signal_id' columns."
        assert resolution_seconds in [
            900,
            3600,
        ], "resolution_seconds must be either 900 (15 minutes) or 3600 (1 hour)."

        # Divide the df into different dfs for each signal_id and add a column for the signal unit using the signals_dict, save them in a dictionary with the signal_id as the key
        dfs = {}
        print(
            f"Reformatting data for {len(df['signal_id'].unique())} unique signal_ids..."
        )
        for signal_id in df["signal_id"].unique():
            print(f"Processing signal_id {signal_id}...")
            df_signal = df[df["signal_id"] == signal_id].copy()

            # Drop values that are not part of a whole interval.
            start_time = (
                df_signal["timestamp_lb_utc"].min().ceil(f"{resolution_seconds}s")
            )
            end_time = (
                df_signal["timestamp_lb_utc"].max().floor(f"{resolution_seconds}s")
            )
            df_signal = df_signal[
                (df_signal["timestamp_lb_utc"] >= start_time)
                & (df_signal["timestamp_lb_utc"] <= end_time)
            ]

            # Resample the data to the desired resolution by taking the average of each interval. For example, if resolution_seconds is 900, take the average of each 15-minute interval
            df_signal.set_index("timestamp_lb_utc", inplace=True)
            df_signal = (
                df_signal.resample(f"{resolution_seconds}s")
                .agg(
                    {
                        "variable_value": "mean",
                        "signal_id": "first",
                    }
                )
                .dropna()
                .reset_index()
            )

            df_signal["variable_unit"] = signals_dict.get(signal_id, "unknown")

            # Create the new columns following the CommonTradingData format.
            df_signal.rename(
                columns={"timestamp_lb_utc": "start_time_lb_utc"}, inplace=True
            )
            print(df_signal)
            df_signal["stop_time_lb_utc"] = df_signal[
                "start_time_lb_utc"
            ] + pd.Timedelta(seconds=resolution_seconds)
            df_signal["db_updated_utc"] = pd.Timestamp.now(tz="UTC")
            df_signal["variable_id"] = variable_id
            df_signal["resolution_seconds"] = resolution_seconds
            df_signal = df_signal[
                [
                    "start_time_lb_utc",
                    "stop_time_lb_utc",
                    "db_updated_utc",
                    "variable_id",
                    "variable_value",
                    "variable_unit",
                    "resolution_seconds",
                ]
            ]
            dfs[signal_id] = df_signal
        return dfs

__init__(country)

Initialize a World Direct Orca API client for a specific asset.

Parameters:

Name	Type	Description	Default
country	str	The country for which to fetch data (eg. "SE", "FI").	required

Example

from physical_operations_utils.wd_orca_utils.WDOrcaAPI import WDOrcaAPI

api = WDOrcaAPI("SE")
device_id = 23  # Hultema -> 32
signals_dict = {
    249: "kw",
    126: "m/s",
    101: "boolean",
}
df = api.fetch_monitoring_data(
    device_id=device_id,
    signals=signals_dict.keys(),
    start_time_lb_utc=pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=2),
    end_time_lb_utc=pd.Timestamp.now(tz="UTC"),
)
df_dict = api.reformat(df, signals_dict=signals_dict, variable_id="735999100016826260", resolution_seconds=900)
for signal_id, df_signal in df_dict.items():
    print(f"Signal ID: {signal_id}")
    print(df_signal)

Source code in physical_operations_utils/wd_orca_utils/WDOrcaAPI.py

def __init__(self, country: str):
    """Initialize a World Direct Orca API client for a specific asset.

    Args:
        country (str): The country for which to fetch data (eg. "SE", "FI").

    Example:
        ```python
        from physical_operations_utils.wd_orca_utils.WDOrcaAPI import WDOrcaAPI

        api = WDOrcaAPI("SE")
        device_id = 23  # Hultema -> 32
        signals_dict = {
            249: "kw",
            126: "m/s",
            101: "boolean",
        }
        df = api.fetch_monitoring_data(
            device_id=device_id,
            signals=signals_dict.keys(),
            start_time_lb_utc=pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=2),
            end_time_lb_utc=pd.Timestamp.now(tz="UTC"),
        )
        df_dict = api.reformat(df, signals_dict=signals_dict, variable_id="735999100016826260", resolution_seconds=900)
        for signal_id, df_signal in df_dict.items():
            print(f"Signal ID: {signal_id}")
            print(df_signal)
        ```
    """
    self.environment = setup_environment()
    self.country = country.upper()

    if self.country == "SE":
        self.base_url = (
            SE_BASE_URL_PROD
            if self.environment.lower() == "prod"
            else SE_BASE_URL_NONPROD
        )
        self.authorize_url = (
            SE_AUTHORIZE_URL_PROD
            if self.environment.lower() == "prod"
            else SE_AUTHORIZE_URL_NONPROD
        )
        self.monitoring_url = self.base_url + "api/monitoring/devices/"
    elif self.country == "FI":
        self.base_url = (
            FI_BASE_URL_PROD
            if self.environment.lower() == "prod"
            else FI_BASE_URL_NONPROD
        )
        self.authorize_url = (
            FI_AUTHORIZE_URL_PROD
            if self.environment.lower() == "prod"
            else FI_AUTHORIZE_URL_NONPROD
        )
        self.monitoring_url = self.base_url + "admin/devices/monitoring/"
    else:
        raise ValueError(f"Unsupported country: {country}")

    self._authorize()

fetch_monitoring_data(device_id, signals, start_time_lb_utc=None, end_time_lb_utc=None)

Returns monitoring data from the World Direct API for a given signal, device and time range.

Parameters:

Name	Type	Description	Default
start_time_lb_utc	Timestamp	Start time in UTC. Default to 10 minutes before end_time_lb_utc.	None
end_time_lb_utc	Timestamp	End time in UTC. Defaults to current time.	None
device_id	int	The ID of the device. Stored in Ancillary Services Database [dbo.asset]	required
signals	list	List of signals to fetch (e.g., [101, 331]).	required

Returns:

Type	Description
DataFrame	pd.DataFrame: DataFrame containing the monitoring data. (timestamp_utc, variable_value, signal_id)

Source code in physical_operations_utils/wd_orca_utils/WDOrcaAPI.py

def fetch_monitoring_data(
    self,
    device_id: int,
    signals: list,
    start_time_lb_utc: pd.Timestamp = None,
    end_time_lb_utc: pd.Timestamp = None,
) -> pd.DataFrame:
    """Returns monitoring data from the World Direct API for a given signal, device and time range.

    Args:
        start_time_lb_utc (pd.Timestamp, optional): Start time in UTC. Default to 10 minutes before end_time_lb_utc.
        end_time_lb_utc (pd.Timestamp, optional): End time in UTC. Defaults to current time.
        device_id (int): The ID of the device. Stored in Ancillary Services Database [dbo.asset]
        signals (list): List of signals to fetch (e.g., [101, 331]).

    Returns:
        pd.DataFrame: DataFrame containing the monitoring data. (timestamp_utc, variable_value, signal_id)
    """

    # If start_time_lb_utc and end_time_lb_utc are not specified, set default to last 10 minutes
    if end_time_lb_utc is None:
        end_time_lb_utc = pd.Timestamp.now(tz="UTC")
    if start_time_lb_utc is None:
        start_time_lb_utc = end_time_lb_utc - pd.Timedelta(minutes=10)

    assert (
        end_time_lb_utc > start_time_lb_utc
    ), "end_time_lb_utc must be a later time than start_time_lb_utc."

    res = requests.get(
        url=self.monitoring_url + f"{device_id}/signals/download",
        headers=self._headers,
        params={
            "deviceId": device_id,
            "From": start_time_lb_utc.isoformat(),
            "To": end_time_lb_utc.isoformat(),
            "Signals": signals,
        },
    )

    if res.status_code != 200:
        raise Exception(
            f"Failed to fetch monitoring data from {self.monitoring_url}. Status code: {res.status_code}"
        )

    # The Swedish instance of the API uses "id" while the Finish uses "signalId"
    signal_id_name = "id" if self.country == "SE" else "signalId"

    # Create the DataFrame from the response (timestamp_utc, variable_value, signal_id).
    df = pd.json_normalize(
        res.json()["signals"],
        record_path="records",
        meta=[signal_id_name],
    )

    if df.empty:
        raise Exception(
            "WD API error: No monitoring data found for the given parameters."
        )

    df.rename(
        columns={
            "timestamp": "timestamp_lb_utc",
            "value": "variable_value",
            signal_id_name: "signal_id",
        },
        inplace=True,
    )
    df["timestamp_lb_utc"] = pd.to_datetime(df["timestamp_lb_utc"], utc=True)

    return df

reformat(df, signals_dict, variable_id, resolution_seconds=900)

Reformats the DataFrame from the World Direct API to a standardized CTDB format.

Parameters:

Name	Type	Description	Default
df	DataFrame	The input DataFrame containing columns 'timestamp_lb_utc', 'variable_value', and 'signal_id'.	required
signals_dict	dict	Dictionary mapping signal IDs to signal unit.	required
variable_id	str	The ID of the variable for which to reformat data.	required
resolution_seconds	int	The desired time resolution in seconds (e.g., 900 for 15 minutes, 3600 for 1 hour).	900

Returns:

Type	Description
dict[int, DataFrame]	dict[int, pd.DataFrame]: A dictionary where the keys are signal IDs and the values are DataFrames reformatted to the CommonTradingData format, containing columns 'start_time_lb_utc', 'stop_time_lb_utc', 'db_updated_utc', 'variable_id', 'variable_value', 'variable_unit', and 'resolution_seconds'.

Source code in physical_operations_utils/wd_orca_utils/WDOrcaAPI.py

def reformat(
    self,
    df: pd.DataFrame,
    signals_dict: dict,
    variable_id: str,
    resolution_seconds: int = 900,
) -> dict[int, pd.DataFrame]:
    """Reformats the DataFrame from the World Direct API to a standardized CTDB format.

    Args:
        df (pd.DataFrame): The input DataFrame containing columns 'timestamp_lb_utc', 'variable_value', and 'signal_id'.
        signals_dict (dict): Dictionary mapping signal IDs to signal unit.
        variable_id (str): The ID of the variable for which to reformat data.
        resolution_seconds (int, optional): The desired time resolution in seconds (e.g., 900 for 15 minutes, 3600 for 1 hour).

    Returns:
        dict[int, pd.DataFrame]: A dictionary where the keys are signal IDs and the values are DataFrames reformatted to the CommonTradingData format, containing columns 'start_time_lb_utc', 'stop_time_lb_utc', 'db_updated_utc', 'variable_id', 'variable_value', 'variable_unit', and 'resolution_seconds'.
    """

    assert (
        "timestamp_lb_utc" in df.columns
        and "variable_value" in df.columns
        and "signal_id" in df.columns
    ), "Input DataFrame must contain 'timestamp_lb_utc', 'variable_value', and 'signal_id' columns."
    assert resolution_seconds in [
        900,
        3600,
    ], "resolution_seconds must be either 900 (15 minutes) or 3600 (1 hour)."

    # Divide the df into different dfs for each signal_id and add a column for the signal unit using the signals_dict, save them in a dictionary with the signal_id as the key
    dfs = {}
    print(
        f"Reformatting data for {len(df['signal_id'].unique())} unique signal_ids..."
    )
    for signal_id in df["signal_id"].unique():
        print(f"Processing signal_id {signal_id}...")
        df_signal = df[df["signal_id"] == signal_id].copy()

        # Drop values that are not part of a whole interval.
        start_time = (
            df_signal["timestamp_lb_utc"].min().ceil(f"{resolution_seconds}s")
        )
        end_time = (
            df_signal["timestamp_lb_utc"].max().floor(f"{resolution_seconds}s")
        )
        df_signal = df_signal[
            (df_signal["timestamp_lb_utc"] >= start_time)
            & (df_signal["timestamp_lb_utc"] <= end_time)
        ]

        # Resample the data to the desired resolution by taking the average of each interval. For example, if resolution_seconds is 900, take the average of each 15-minute interval
        df_signal.set_index("timestamp_lb_utc", inplace=True)
        df_signal = (
            df_signal.resample(f"{resolution_seconds}s")
            .agg(
                {
                    "variable_value": "mean",
                    "signal_id": "first",
                }
            )
            .dropna()
            .reset_index()
        )

        df_signal["variable_unit"] = signals_dict.get(signal_id, "unknown")

        # Create the new columns following the CommonTradingData format.
        df_signal.rename(
            columns={"timestamp_lb_utc": "start_time_lb_utc"}, inplace=True
        )
        print(df_signal)
        df_signal["stop_time_lb_utc"] = df_signal[
            "start_time_lb_utc"
        ] + pd.Timedelta(seconds=resolution_seconds)
        df_signal["db_updated_utc"] = pd.Timestamp.now(tz="UTC")
        df_signal["variable_id"] = variable_id
        df_signal["resolution_seconds"] = resolution_seconds
        df_signal = df_signal[
            [
                "start_time_lb_utc",
                "stop_time_lb_utc",
                "db_updated_utc",
                "variable_id",
                "variable_value",
                "variable_unit",
                "resolution_seconds",
            ]
        ]
        dfs[signal_id] = df_signal
    return dfs