ems_rest_api

This is a wrapper to communicate with the EMS REST API.

EMSRestApi

A class to interact with the EMS REST API.

Provides methods for authenticating, retrieving time series data for specific data groups, and posting data back to the EMS system. The API supports working with both JSON responses.

The API documentation can be found at: https://ems.energymarketservices.volue.com/ExternalData/swagger/index.html

Data groups can be managed in the EMS in the External data section.

Authentication can either be handled via the KEYS_FILE or by passing in a username and password.

Source code in physical_operations_utils/ems_utils/EMSRestApi.py

class EMSRestApi:
    """
    A class to interact with the EMS REST API.

    Provides methods for authenticating, retrieving time series data for specific data groups,
    and posting data back to the EMS system. The API supports working with both JSON responses.

    The API documentation can be found at: https://ems.energymarketservices.volue.com/ExternalData/swagger/index.html

    Data groups can be managed in the EMS in the External data section.

    Authentication can either be handled via the KEYS_FILE or by passing in a username and password.
    """

    def __init__(
        self,
        use_keys_file: bool,
        username: str | None = None,
        password: str | None = None,
    ):
        """
        Initialize the EMSRestApi client.

        Args:
            use_keys_file (bool): If True, credentials are loaded from the KEYS_FILE.yml using Azure Key Vault.
            username (str | None): Username for authentication. Required if not using keys file.
            password (str | None): Password for authentication. Required if not using keys file.

        Raises:
            ValueError: If credentials are missing or keys file is misconfigured.

        Example:
            ```python
            from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

            ems_api = EMSRestApi(use_keys_file=True)
            ```
        """
        self._base_url = "https://ems.energymarketservices.volue.com/ExternalData"
        self._auth_url = f"{self._base_url}/Users/Authenticate"
        self._token = None
        if use_keys_file:
            setup_environment()
            keys = get_keys_yaml_file().get("volue_rest_api", None)
            if keys is None:
                raise ValueError("No keys found in KEYS_FILE.yml.")
            self.username = keys["username"]
            self.password = get_secret(keys["secret"])
        else:
            if username is None or password is None:
                raise ValueError("Username and password must be provided.")
            self.username = username
            self.password = password
        self.authenticate()

    @retry(
        wait=wait_fixed(2),
        stop=stop_after_attempt(3),
        reraise=True,
    )
    def authenticate(self) -> None:
        """
        Authenticates against the EMS REST API and stores the bearer token.

        Raises:
            ValueError: If the API returns 200 but no token is present.
            HTTPError: If authentication fails with a non-200 response.

        Example:
            ```python
            from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

            ems_api = EMSRestApi(use_keys_file=True)
            ems_api.authenticate()
            print(ems_api._token)  # Prints the bearer token
            ```
        """
        data = json.dumps({"username": self.username, "password": self.password})
        headers = {"Content-Type": "application/json"}
        response = requests.post(url=self._auth_url, data=data, headers=headers)
        if response.status_code == 200:
            if response.text:
                self._token = response.json().get("token")
            else:
                raise ValueError(
                    "Authentication failed: No token received with response code 200."
                )
        else:
            logging.error(
                f"Authentication failed with status code {response.status_code}: {response.text}"
            )
        response.raise_for_status()

    @retry(
        wait=wait_fixed(2),
        stop=stop_after_attempt(3),
        reraise=True,
    )
    def get_available_data_groups(self) -> List[str]:
        """
        Retrieves a list of all available data group names.

        Returns:
            List[str]: A list of data group names.

        Raises:
            ValueError: If the response is 200 but empty.
            HTTPError: If the request fails.

        Example:
            ```
            python
            from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

            ems_api = EMSRestApi(use_keys_file=True)
            data_groups = ems_api.get_available_data_groups()
            print(data_groups)
            ```
        """
        url = f"{self._base_url}/DataGroups"
        self.authenticate()
        headers = {"Authorization": f"Bearer {self._token}"}
        response = requests.get(url=url, headers=headers)
        if response.status_code == 200:
            if response.text:
                return [group for group in response.json()]
            raise ValueError("Empty response with status code 200.")
        elif response.status_code == 204:
            return []  # No content, return an empty list
        else:
            logging.error(
                f"Failed to retrieve data groups with status code {response.status_code}: {response.text}"
            )
        response.raise_for_status()

    def get_data_group_as_df(
        self,
        data_group: str,
        start_time_lb_utc: datetime,
        stop_time_lb_utc: datetime,
        data_point_ids: List[str] | None = None,
    ) -> pd.DataFrame:
        """
        Fetches data from the specified data group within the given time range.
        This method retrieves data from the EMS API and returns it as a DataFrame. A list of data point
        names can be provided to filter the results.

        Args:
            data_group (str): The data group name to fetch.
            start_time_lb_utc (datetime): UTC start time (lower bound).
            stop_time_lb_utc (datetime): UTC stop time (upper bound).
            data_point_ids (List[str] | None): Optional list of data point names to filter.

        Returns:
            pd.DataFrame: A DataFrame containing the data points and their values.

        Raises:
            ValueError: If the response is 200 but empty.
            HTTPError: If the request fails.

        Example:
            ```python
            from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi
            from zoneinfo import ZoneInfo
            from datetime import datetime

            ems_api = EMSRestApi(use_keys_file=True)
            start_time = datetime(2024, 1, 1, 0, 0, tzinfo=ZoneInfo("UTC"))
            stop_time = datetime(2024, 1, 2, 0, 0, tzinfo=ZoneInfo("UTC"))
            df = ems_api.get_data_group_as_df("MOCK_GROUP", start_time, stop_time)
            print(df)
            ```
        """
        data = self._get_data_group(
            data_group=data_group,
            start_time_lb_utc=start_time_lb_utc,
            stop_time_lb_utc=stop_time_lb_utc,
            data_point_ids=data_point_ids,
        )
        res_columns = [
            "variable_id",
            "start_time_lb_utc",
            "resolution_seconds",
            "variable_value",
            "variable_unit",
            "variable_status",
            "last_updated_utc",
            "updated_by",
        ]
        df = pd.DataFrame(
            columns=res_columns,
        )
        df["start_time_lb_utc"] = pd.to_datetime(df["start_time_lb_utc"], utc=True)
        if not data or not data.get("dataPoints"):
            return df
        dfs = []
        data_points: List[dict] = data.get("dataPoints", [])
        for data_point in data_points:
            variable_id = data_point.get("name", "")
            unit = data_point.get("unit", "")
            time_level = data_point.get("timeLevel", "")
            if time_level == "Hour":
                resolution_seconds = 3600
            elif time_level == "15Min":
                resolution_seconds = 900
            else:
                resolution_seconds = 0
            time_series = data_point.get("timeSeries", [])
            temp_df = pd.DataFrame(
                time_series,
            )
            if not temp_df.empty:
                temp_df["variable_id"] = variable_id
                temp_df["start_time_lb_utc"] = pd.to_datetime(
                    temp_df["timeStamp"], utc=True, format=ZULU_STRING_FORMAT
                )
                temp_df["resolution_seconds"] = resolution_seconds
                temp_df["variable_unit"] = unit
                temp_df["variable_value"] = temp_df["value"]
                # temp_df["variable_status"] = temp_df["status"].map(EMS_STATUS_MAP)
                temp_df["variable_status"] = (
                    temp_df["status"]
                    .astype("string")
                    .str.strip()
                    .str.lower()
                    .str.replace(r"\s+", "", regex=True)
                    .map(EMS_STATUS_MAP)
                    .fillna(EMS_STATUS_MAP["unknownstatus"])
                    .astype("int64")
                )
                temp_df["last_updated_utc"] = pd.to_datetime(
                    temp_df["updated"]
                    .astype("string")
                    .str.replace(r"\.\d+(?=Z$)", "", regex=True),
                    utc=True,
                    format=ZULU_STRING_FORMAT,
                    errors="coerce",
                )
                temp_df["updated_by"] = temp_df["updatedBy"]
                dfs.append(temp_df)
        if len(dfs) == 0:
            final_df = df.copy(deep=True)
        else:
            final_df = pd.concat(
                dfs,
                ignore_index=True,
            ).reset_index(drop=True)
        final_df["resolution_seconds"] = final_df["resolution_seconds"].astype("int64")
        return (
            final_df[res_columns]
            .sort_values(by=["variable_id", "start_time_lb_utc"])
            .reset_index(drop=True)
        )

    def get_data_group_as_dict(
        self,
        data_group: str,
        start_time_lb_utc: datetime,
        stop_time_lb_utc: datetime,
        data_point_ids: List[str] | None = None,
    ) -> dict:
        """
        Fetches data from the specified data group within the given time range.

        This method retrieves data from the EMS API and returns it as a dictionary. A list of data point
        names can be provided to filter the results.

        Args:
            data_group (str): The data group name to fetch.
            start_time_lb_utc (datetime): UTC start time (lower bound).
            stop_time_lb_utc (datetime): UTC stop time (upper bound).
            data_point_ids (List[str] | None): Optional list of data point names to filter.

        Returns:
            dict: The raw JSON response from the EMS API.

        Raises:
            ValueError: If the response is 200 but empty.
            HTTPError: If the request fails.

        Example:
            ```python
            from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

            ems_api = EMSRestApi(use_keys_file=True)
            data = ems_api.get_data_group_as_dict("MOCK_GROUP", start, end)
            print(data)
            ```
        """
        return self._get_data_group(
            data_group=data_group,
            start_time_lb_utc=start_time_lb_utc,
            stop_time_lb_utc=stop_time_lb_utc,
            data_point_ids=data_point_ids,
        )

    @retry(
        wait=wait_fixed(2),
        stop=stop_after_attempt(3),
        reraise=True,
    )
    def post_to_data_group(self, data_group: str, data_points: List[dict]) -> int:
        """
        Posts data points to the specified data group.

        The data points must be in the format expected by the EMS API. To generate data points from a DataFrame,
        use the `generate_data_points_from_df` function.

        Args:
            data_group (str): The data group name to post to.
            data_points (List[dict]): A list of data points to post.
                Each data point must have a 'name' key and a 'timeSeries' key.
                The 'timeSeries' key must contain a list of dictionaries with 'timeStamp', 'value', and 'status'.
                The 'status' must be one of the allowed statuses "Good", "Blocked entered" or "Entered"
                The 'timeStamp' must be in ISO 8601 format.
                The 'value' must be a number.

        Returns:
            int: The HTTP status code of the response.

        Raises:
            ValueError: If the data points are not valid.
            HTTPError: If the request fails.

        Example:
            ```python
            from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

            ems_api = EMSRestApi(use_keys_file=True)
            data_points = [
                {
                    "name": "MOCK_POINT",
                    "unit": "kWh",
                    "timeLevel": "Hour",
                    "timeSeries": [
                        {
                            "timeStamp": "2024-01-01T00:00:00Z",
                            "value": 1.0,
                            "status": "Good",
                        }
                    ],
                }
            ]
            status_code = ems_api.post_to_data_group("MOCK_GROUP", data_points)
            print(status_code)
            ```
        """
        _validate_data_points(data_points)
        if not isinstance(data_group, str) or not data_group.strip():
            raise ValueError("data_group must be a non-empty string.")
        self.authenticate()
        url = f"{self._base_url}/DataGroups/"
        headers = {
            "Authorization": f"Bearer {self._token}",
            "Content-Type": "application/json",
        }
        payload = {
            "name": data_group,
            "isWriteAllowed": True,
            "dataPoints": data_points,
        }
        response = requests.post(url=url, headers=headers, json=payload)
        if not response.ok:
            logging.error(
                f"Failed to post data group with status code {response.status_code}: {response.text}"
            )
        response.raise_for_status()
        return response.status_code

    @retry(
        wait=wait_fixed(2),
        stop=stop_after_attempt(3),
        reraise=True,
    )
    def _get_data_group(
        self,
        data_group: str,
        start_time_lb_utc: datetime,
        stop_time_lb_utc: datetime,
        data_point_ids: List[str] | None = None,
    ) -> dict:
        """
        Fetches data from the specified data group within the given time range.

        Args:
            data_group (str): The data group name to fetch.
            start_time_lb_utc (datetime): UTC start time (lower bound).
            stop_time_lb_utc (datetime): UTC stop time (upper bound).
            data_point_ids (List[str] | None): Optional list of data point names to filter.

        Returns:
            dict: The raw JSON response from the EMS API.

        Raises:
            ValueError: If the response is 200 but empty.
            HTTPError: If the request fails.

        Example:
            ```python
            from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

            ems_api = EMSRestApi(use_keys_file=True)
            data = ems_api._get_data_group("MOCK_GROUP", start, end)
            print(data)
            ```
        """
        validate_datetime_in_utc(start_time_lb_utc)
        validate_datetime_in_utc(stop_time_lb_utc)
        validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)
        self.authenticate()
        url = f"{self._base_url}/DataGroups/{data_group}"
        headers = {"Authorization": f"Bearer {self._token}"}
        params = [
            ("start", convert_datetime_to_zulu_string(start_time_lb_utc)),
            ("end", convert_datetime_to_zulu_string(stop_time_lb_utc)),
        ]
        if data_point_ids:
            params.extend([("dataPointName", name) for name in data_point_ids])
        response = requests.get(url=url, headers=headers, params=params)
        if response.status_code == 200:
            if response.text:
                return response.json()
            raise ValueError("Empty response with status code 200.")
        elif response.status_code == 204:
            return {}
        else:
            logging.error(
                f"Failed to data group status code {response.status_code}: {response.text}"
            )
        response.raise_for_status()

__init__(use_keys_file, username=None, password=None)

Initialize the EMSRestApi client.

Parameters:

Name	Type	Description	Default
use_keys_file	bool	If True, credentials are loaded from the KEYS_FILE.yml using Azure Key Vault.	required
username	str | None	Username for authentication. Required if not using keys file.	None
password	str | None	Password for authentication. Required if not using keys file.	None

Raises:

Type	Description
ValueError	If credentials are missing or keys file is misconfigured.

Example

from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

ems_api = EMSRestApi(use_keys_file=True)

Source code in physical_operations_utils/ems_utils/EMSRestApi.py

def __init__(
    self,
    use_keys_file: bool,
    username: str | None = None,
    password: str | None = None,
):
    """
    Initialize the EMSRestApi client.

    Args:
        use_keys_file (bool): If True, credentials are loaded from the KEYS_FILE.yml using Azure Key Vault.
        username (str | None): Username for authentication. Required if not using keys file.
        password (str | None): Password for authentication. Required if not using keys file.

    Raises:
        ValueError: If credentials are missing or keys file is misconfigured.

    Example:
        ```python
        from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

        ems_api = EMSRestApi(use_keys_file=True)
        ```
    """
    self._base_url = "https://ems.energymarketservices.volue.com/ExternalData"
    self._auth_url = f"{self._base_url}/Users/Authenticate"
    self._token = None
    if use_keys_file:
        setup_environment()
        keys = get_keys_yaml_file().get("volue_rest_api", None)
        if keys is None:
            raise ValueError("No keys found in KEYS_FILE.yml.")
        self.username = keys["username"]
        self.password = get_secret(keys["secret"])
    else:
        if username is None or password is None:
            raise ValueError("Username and password must be provided.")
        self.username = username
        self.password = password
    self.authenticate()

authenticate()

Authenticates against the EMS REST API and stores the bearer token.

Raises:

Type	Description
ValueError	If the API returns 200 but no token is present.
HTTPError	If authentication fails with a non-200 response.

Example

from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

ems_api = EMSRestApi(use_keys_file=True)
ems_api.authenticate()
print(ems_api._token)  # Prints the bearer token

Source code in physical_operations_utils/ems_utils/EMSRestApi.py

@retry(
    wait=wait_fixed(2),
    stop=stop_after_attempt(3),
    reraise=True,
)
def authenticate(self) -> None:
    """
    Authenticates against the EMS REST API and stores the bearer token.

    Raises:
        ValueError: If the API returns 200 but no token is present.
        HTTPError: If authentication fails with a non-200 response.

    Example:
        ```python
        from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

        ems_api = EMSRestApi(use_keys_file=True)
        ems_api.authenticate()
        print(ems_api._token)  # Prints the bearer token
        ```
    """
    data = json.dumps({"username": self.username, "password": self.password})
    headers = {"Content-Type": "application/json"}
    response = requests.post(url=self._auth_url, data=data, headers=headers)
    if response.status_code == 200:
        if response.text:
            self._token = response.json().get("token")
        else:
            raise ValueError(
                "Authentication failed: No token received with response code 200."
            )
    else:
        logging.error(
            f"Authentication failed with status code {response.status_code}: {response.text}"
        )
    response.raise_for_status()

get_available_data_groups()

Retrieves a list of all available data group names.

Returns:

Type	Description
List[str]	List[str]: A list of data group names.

Raises:

Type	Description
ValueError	If the response is 200 but empty.
HTTPError	If the request fails.

Example

python
from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

ems_api = EMSRestApi(use_keys_file=True)
data_groups = ems_api.get_available_data_groups()
print(data_groups)

Source code in physical_operations_utils/ems_utils/EMSRestApi.py

@retry(
    wait=wait_fixed(2),
    stop=stop_after_attempt(3),
    reraise=True,
)
def get_available_data_groups(self) -> List[str]:
    """
    Retrieves a list of all available data group names.

    Returns:
        List[str]: A list of data group names.

    Raises:
        ValueError: If the response is 200 but empty.
        HTTPError: If the request fails.

    Example:
        ```
        python
        from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

        ems_api = EMSRestApi(use_keys_file=True)
        data_groups = ems_api.get_available_data_groups()
        print(data_groups)
        ```
    """
    url = f"{self._base_url}/DataGroups"
    self.authenticate()
    headers = {"Authorization": f"Bearer {self._token}"}
    response = requests.get(url=url, headers=headers)
    if response.status_code == 200:
        if response.text:
            return [group for group in response.json()]
        raise ValueError("Empty response with status code 200.")
    elif response.status_code == 204:
        return []  # No content, return an empty list
    else:
        logging.error(
            f"Failed to retrieve data groups with status code {response.status_code}: {response.text}"
        )
    response.raise_for_status()

get_data_group_as_df(data_group, start_time_lb_utc, stop_time_lb_utc, data_point_ids=None)

Fetches data from the specified data group within the given time range. This method retrieves data from the EMS API and returns it as a DataFrame. A list of data point names can be provided to filter the results.

Parameters:

Name	Type	Description	Default
data_group	str	The data group name to fetch.	required
start_time_lb_utc	datetime	UTC start time (lower bound).	required
stop_time_lb_utc	datetime	UTC stop time (upper bound).	required
data_point_ids	List[str] | None	Optional list of data point names to filter.	None

Returns:

Type	Description
DataFrame	pd.DataFrame: A DataFrame containing the data points and their values.

Raises:

Type	Description
ValueError	If the response is 200 but empty.
HTTPError	If the request fails.

Example

from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi
from zoneinfo import ZoneInfo
from datetime import datetime

ems_api = EMSRestApi(use_keys_file=True)
start_time = datetime(2024, 1, 1, 0, 0, tzinfo=ZoneInfo("UTC"))
stop_time = datetime(2024, 1, 2, 0, 0, tzinfo=ZoneInfo("UTC"))
df = ems_api.get_data_group_as_df("MOCK_GROUP", start_time, stop_time)
print(df)

Source code in physical_operations_utils/ems_utils/EMSRestApi.py

def get_data_group_as_df(
    self,
    data_group: str,
    start_time_lb_utc: datetime,
    stop_time_lb_utc: datetime,
    data_point_ids: List[str] | None = None,
) -> pd.DataFrame:
    """
    Fetches data from the specified data group within the given time range.
    This method retrieves data from the EMS API and returns it as a DataFrame. A list of data point
    names can be provided to filter the results.

    Args:
        data_group (str): The data group name to fetch.
        start_time_lb_utc (datetime): UTC start time (lower bound).
        stop_time_lb_utc (datetime): UTC stop time (upper bound).
        data_point_ids (List[str] | None): Optional list of data point names to filter.

    Returns:
        pd.DataFrame: A DataFrame containing the data points and their values.

    Raises:
        ValueError: If the response is 200 but empty.
        HTTPError: If the request fails.

    Example:
        ```python
        from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi
        from zoneinfo import ZoneInfo
        from datetime import datetime

        ems_api = EMSRestApi(use_keys_file=True)
        start_time = datetime(2024, 1, 1, 0, 0, tzinfo=ZoneInfo("UTC"))
        stop_time = datetime(2024, 1, 2, 0, 0, tzinfo=ZoneInfo("UTC"))
        df = ems_api.get_data_group_as_df("MOCK_GROUP", start_time, stop_time)
        print(df)
        ```
    """
    data = self._get_data_group(
        data_group=data_group,
        start_time_lb_utc=start_time_lb_utc,
        stop_time_lb_utc=stop_time_lb_utc,
        data_point_ids=data_point_ids,
    )
    res_columns = [
        "variable_id",
        "start_time_lb_utc",
        "resolution_seconds",
        "variable_value",
        "variable_unit",
        "variable_status",
        "last_updated_utc",
        "updated_by",
    ]
    df = pd.DataFrame(
        columns=res_columns,
    )
    df["start_time_lb_utc"] = pd.to_datetime(df["start_time_lb_utc"], utc=True)
    if not data or not data.get("dataPoints"):
        return df
    dfs = []
    data_points: List[dict] = data.get("dataPoints", [])
    for data_point in data_points:
        variable_id = data_point.get("name", "")
        unit = data_point.get("unit", "")
        time_level = data_point.get("timeLevel", "")
        if time_level == "Hour":
            resolution_seconds = 3600
        elif time_level == "15Min":
            resolution_seconds = 900
        else:
            resolution_seconds = 0
        time_series = data_point.get("timeSeries", [])
        temp_df = pd.DataFrame(
            time_series,
        )
        if not temp_df.empty:
            temp_df["variable_id"] = variable_id
            temp_df["start_time_lb_utc"] = pd.to_datetime(
                temp_df["timeStamp"], utc=True, format=ZULU_STRING_FORMAT
            )
            temp_df["resolution_seconds"] = resolution_seconds
            temp_df["variable_unit"] = unit
            temp_df["variable_value"] = temp_df["value"]
            # temp_df["variable_status"] = temp_df["status"].map(EMS_STATUS_MAP)
            temp_df["variable_status"] = (
                temp_df["status"]
                .astype("string")
                .str.strip()
                .str.lower()
                .str.replace(r"\s+", "", regex=True)
                .map(EMS_STATUS_MAP)
                .fillna(EMS_STATUS_MAP["unknownstatus"])
                .astype("int64")
            )
            temp_df["last_updated_utc"] = pd.to_datetime(
                temp_df["updated"]
                .astype("string")
                .str.replace(r"\.\d+(?=Z$)", "", regex=True),
                utc=True,
                format=ZULU_STRING_FORMAT,
                errors="coerce",
            )
            temp_df["updated_by"] = temp_df["updatedBy"]
            dfs.append(temp_df)
    if len(dfs) == 0:
        final_df = df.copy(deep=True)
    else:
        final_df = pd.concat(
            dfs,
            ignore_index=True,
        ).reset_index(drop=True)
    final_df["resolution_seconds"] = final_df["resolution_seconds"].astype("int64")
    return (
        final_df[res_columns]
        .sort_values(by=["variable_id", "start_time_lb_utc"])
        .reset_index(drop=True)
    )

get_data_group_as_dict(data_group, start_time_lb_utc, stop_time_lb_utc, data_point_ids=None)

Fetches data from the specified data group within the given time range.

This method retrieves data from the EMS API and returns it as a dictionary. A list of data point names can be provided to filter the results.

Parameters:

Name	Type	Description	Default
data_group	str	The data group name to fetch.	required
start_time_lb_utc	datetime	UTC start time (lower bound).	required
stop_time_lb_utc	datetime	UTC stop time (upper bound).	required
data_point_ids	List[str] | None	Optional list of data point names to filter.	None

Returns:

Name	Type	Description
dict	dict	The raw JSON response from the EMS API.

Raises:

Type	Description
ValueError	If the response is 200 but empty.
HTTPError	If the request fails.

Example

from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

ems_api = EMSRestApi(use_keys_file=True)
data = ems_api.get_data_group_as_dict("MOCK_GROUP", start, end)
print(data)

Source code in physical_operations_utils/ems_utils/EMSRestApi.py

def get_data_group_as_dict(
    self,
    data_group: str,
    start_time_lb_utc: datetime,
    stop_time_lb_utc: datetime,
    data_point_ids: List[str] | None = None,
) -> dict:
    """
    Fetches data from the specified data group within the given time range.

    This method retrieves data from the EMS API and returns it as a dictionary. A list of data point
    names can be provided to filter the results.

    Args:
        data_group (str): The data group name to fetch.
        start_time_lb_utc (datetime): UTC start time (lower bound).
        stop_time_lb_utc (datetime): UTC stop time (upper bound).
        data_point_ids (List[str] | None): Optional list of data point names to filter.

    Returns:
        dict: The raw JSON response from the EMS API.

    Raises:
        ValueError: If the response is 200 but empty.
        HTTPError: If the request fails.

    Example:
        ```python
        from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

        ems_api = EMSRestApi(use_keys_file=True)
        data = ems_api.get_data_group_as_dict("MOCK_GROUP", start, end)
        print(data)
        ```
    """
    return self._get_data_group(
        data_group=data_group,
        start_time_lb_utc=start_time_lb_utc,
        stop_time_lb_utc=stop_time_lb_utc,
        data_point_ids=data_point_ids,
    )

post_to_data_group(data_group, data_points)

Posts data points to the specified data group.

The data points must be in the format expected by the EMS API. To generate data points from a DataFrame, use the generate_data_points_from_df function.

Parameters:

Name	Type	Description	Default
data_group	str	The data group name to post to.	required
data_points	List[dict]	A list of data points to post. Each data point must have a 'name' key and a 'timeSeries' key. The 'timeSeries' key must contain a list of dictionaries with 'timeStamp', 'value', and 'status'. The 'status' must be one of the allowed statuses "Good", "Blocked entered" or "Entered" The 'timeStamp' must be in ISO 8601 format. The 'value' must be a number.	required

Returns:

Name	Type	Description
int	int	The HTTP status code of the response.

Raises:

Type	Description
ValueError	If the data points are not valid.
HTTPError	If the request fails.

Example

from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

ems_api = EMSRestApi(use_keys_file=True)
data_points = [
    {
        "name": "MOCK_POINT",
        "unit": "kWh",
        "timeLevel": "Hour",
        "timeSeries": [
            {
                "timeStamp": "2024-01-01T00:00:00Z",
                "value": 1.0,
                "status": "Good",
            }
        ],
    }
]
status_code = ems_api.post_to_data_group("MOCK_GROUP", data_points)
print(status_code)

Source code in physical_operations_utils/ems_utils/EMSRestApi.py

@retry(
    wait=wait_fixed(2),
    stop=stop_after_attempt(3),
    reraise=True,
)
def post_to_data_group(self, data_group: str, data_points: List[dict]) -> int:
    """
    Posts data points to the specified data group.

    The data points must be in the format expected by the EMS API. To generate data points from a DataFrame,
    use the `generate_data_points_from_df` function.

    Args:
        data_group (str): The data group name to post to.
        data_points (List[dict]): A list of data points to post.
            Each data point must have a 'name' key and a 'timeSeries' key.
            The 'timeSeries' key must contain a list of dictionaries with 'timeStamp', 'value', and 'status'.
            The 'status' must be one of the allowed statuses "Good", "Blocked entered" or "Entered"
            The 'timeStamp' must be in ISO 8601 format.
            The 'value' must be a number.

    Returns:
        int: The HTTP status code of the response.

    Raises:
        ValueError: If the data points are not valid.
        HTTPError: If the request fails.

    Example:
        ```python
        from physical_operations_utils.ems_utils.EMSRestApi import EMSRestApi

        ems_api = EMSRestApi(use_keys_file=True)
        data_points = [
            {
                "name": "MOCK_POINT",
                "unit": "kWh",
                "timeLevel": "Hour",
                "timeSeries": [
                    {
                        "timeStamp": "2024-01-01T00:00:00Z",
                        "value": 1.0,
                        "status": "Good",
                    }
                ],
            }
        ]
        status_code = ems_api.post_to_data_group("MOCK_GROUP", data_points)
        print(status_code)
        ```
    """
    _validate_data_points(data_points)
    if not isinstance(data_group, str) or not data_group.strip():
        raise ValueError("data_group must be a non-empty string.")
    self.authenticate()
    url = f"{self._base_url}/DataGroups/"
    headers = {
        "Authorization": f"Bearer {self._token}",
        "Content-Type": "application/json",
    }
    payload = {
        "name": data_group,
        "isWriteAllowed": True,
        "dataPoints": data_points,
    }
    response = requests.post(url=url, headers=headers, json=payload)
    if not response.ok:
        logging.error(
            f"Failed to post data group with status code {response.status_code}: {response.text}"
        )
    response.raise_for_status()
    return response.status_code

generate_data_points_from_df(data_point_id, df, unit=None, time_level=None, status=None)

Generates a data points dictionary from a DataFrame which can be used to post data to the REST API. The DataFrame must contain 'start_time_lb_utc' and 'variable_value' columns. The 'start_time_lb_utc' column must be in UTC format. The 'variable_value' column must contain numeric values. The 'status' parameter can be used to set the status of the data points. The status must be one of the allowed statuses: "Good", "Blocked entered", "Entered". The 'unit' parameter can be used to set the unit of the data points. The 'time_level' parameter can be used to set the time level of the data points. The time level must be one of the allowed time levels: "Hour", "15Min".

Parameters:

Name	Type	Description	Default
data_point_id	str	The ID of the data point.	required
df	DataFrame	The DataFrame containing the data points.	required
unit	str | None	The unit of the data points. Optional.	None
time_level	str | None	The time level of the data points. Optional.	None
status	str | None	The status of the data points. Optional. Must be one of the allowed statuses: "Good", "Blocked entered", "Entered". If not provided, defaults to "Good".	None

Returns:

Name	Type	Description
dict	dict	A dictionary containing the data points in the format expected by the EMS API.

Raises:

Type	Description
ValueError	If the DataFrame does not contain the required columns or if the status is invalid.
TypeError	If the DataFrame is not a pandas DataFrame.
ValueError	If the data point ID is not a string or is empty.

Example

import pandas as pd
from physical_operations_utils.ems_utils.EMSRestApi import generate_data_points_from_df

df = pd.DataFrame({
    "start_time_lb_utc": ["2024-01-01T00:00:00Z", "2024-01-01T01:00:00Z"],
    "variable_value": [1.0, 2.0],
})
data_points = generate_data_points_from_df("MOCK_POINT", df)
print(data_points)

Source code in physical_operations_utils/ems_utils/EMSRestApi.py

def generate_data_points_from_df(
    data_point_id: str,
    df: pd.DataFrame,
    unit: str | None = None,
    time_level: str | None = None,
    status: str | None = None,
) -> dict:
    """
    Generates a data points dictionary from a DataFrame which can be used to post data to the REST API.
    The DataFrame must contain 'start_time_lb_utc' and 'variable_value' columns.
    The 'start_time_lb_utc' column must be in UTC format.
    The 'variable_value' column must contain numeric values.
    The 'status' parameter can be used to set the status of the data points.
    The status must be one of the allowed statuses: "Good", "Blocked entered", "Entered".
    The 'unit' parameter can be used to set the unit of the data points.
    The 'time_level' parameter can be used to set the time level of the data points.
    The time level must be one of the allowed time levels: "Hour", "15Min".

    Args:
        data_point_id (str): The ID of the data point.
        df (pd.DataFrame): The DataFrame containing the data points.
        unit (str | None): The unit of the data points. Optional.
        time_level (str | None): The time level of the data points. Optional.
        status (str | None): The status of the data points. Optional.
            Must be one of the allowed statuses: "Good", "Blocked entered", "Entered".
            If not provided, defaults to "Good".

    Returns:
        dict: A dictionary containing the data points in the format expected by the EMS API.

    Raises:
        ValueError: If the DataFrame does not contain the required columns or if the status is invalid.
        TypeError: If the DataFrame is not a pandas DataFrame.
        ValueError: If the data point ID is not a string or is empty.

    Example:
        ```python
        import pandas as pd
        from physical_operations_utils.ems_utils.EMSRestApi import generate_data_points_from_df

        df = pd.DataFrame({
            "start_time_lb_utc": ["2024-01-01T00:00:00Z", "2024-01-01T01:00:00Z"],
            "variable_value": [1.0, 2.0],
        })
        data_points = generate_data_points_from_df("MOCK_POINT", df)
        print(data_points)
        ```
    """
    if status is not None and status not in ALLOWED_STATUSES:
        raise ValueError("Invalid status value.")
    if not {"start_time_lb_utc", "variable_value"}.issubset(df.columns):
        raise ValueError(
            "DataFrame must contain 'start_time_lb_utc', 'variable_value' columns."
        )
    df_cp = df.copy(deep=True).sort_values(by=["start_time_lb_utc"])
    validate_df_column_is_utc_datetime(df_cp, "start_time_lb_utc")
    data_points = {"name": data_point_id}
    if unit:
        data_points["unit"] = unit
    if time_level:
        data_points["timeLevel"] = time_level
    data_points["timeSeries"] = [
        {
            "timeStamp": convert_datetime_to_zulu_string(row.loc["start_time_lb_utc"]),
            "value": row.loc["variable_value"],
            "status": status if status else "Good",
        }
        for _, row in df_cp.iterrows()
    ]
    return data_points