bazefield_api

This is a wrapper to communicate with the BazeField API.

BazefieldApi

Client for the Bazefield REST API bound to a specific operator and wind farm.

The client provides high-level methods for:

• Resolving sites and turbines
• Discovering available measurement points
• Fetching aggregated turbine-level time series
• Aggregating data across turbines using MPID mappings

Attributes:

Name	Type	Description
operator		Operator name (e.g. "ox2", "arise").
farm_code		Wind farm identifier.
base_url		Base URL for the Bazefield API.
verify_ssl		Whether SSL certificates are verified.
site_id		Bazefield site object ID.
turbine_object_to_mpid		Mapping from turbine object IDs to MPIDs.

Notes

Each instance is bound to a single operator and wind farm. All time-series methods return :class:pandas.DataFrame objects.

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

class BazefieldApi:
    """
    Client for the Bazefield REST API bound to a specific operator and wind farm.

    The client provides high-level methods for:

    - Resolving sites and turbines
    - Discovering available measurement points
    - Fetching aggregated turbine-level time series
    - Aggregating data across turbines using MPID mappings

    Attributes:
        operator: Operator name (e.g. ``"ox2"``, ``"arise"``).
        farm_code: Wind farm identifier.
        base_url: Base URL for the Bazefield API.
        verify_ssl: Whether SSL certificates are verified.
        site_id: Bazefield site object ID.
        turbine_object_to_mpid: Mapping from turbine object IDs to MPIDs.

    Notes:
        Each instance is bound to a single operator and wind farm.
        All time-series methods return :class:`pandas.DataFrame` objects.
    """

    def __init__(self, operator: str, farm_code: str):
        """
        Initialize a Bazefield API client for a specific operator and wind farm.

        During initialization the client:

        - Loads configuration from disk
        - Retrieves API credentials from Azure Key Vault
        - Creates an authenticated HTTP session
        - Resolves the site object ID
        - Resolves turbine-to-MPID mappings

        Args:
            operator: Operator name.
            farm_code: Wind farm code as defined in the Bazefield configuration.

        Raises:
            KeyError: If the operator or farm code is missing from the configuration.
            RuntimeError: If the site or turbine MPIDs cannot be resolved.

        Example:
            ```python
            from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

            bz = BazefieldApi(operator="ox2", farm_code="MLK")
            ```
        """
        self.operator = operator
        self.farm_code = farm_code

        config = load_bazefield_config()
        self.cfg = config[operator][farm_code]
        self.mpid_map = self.cfg["wtg_number_mpid_map"]

        api_key = get_secret(self.cfg["secret"])
        self.base_url, self.verify_ssl = get_operator_connection(operator)

        self.session = requests.Session()
        self.session.headers.update(
            {
                "Authorization": f"Bearer {api_key}",
                "Accept": "application/json",
            }
        )

        self.site_id = self._resolve_site_id()
        self.turbine_object_to_mpid = self._resolve_turbine_mpid_map()

    def _get(self, path: str, params: dict = None) -> dict:
        r = self.session.get(
            f"{self.base_url}/{path.lstrip('/')}",
            params=params,
            timeout=30,
            verify=self.verify_ssl,
        )

        try:
            r.raise_for_status()
        except requests.HTTPError:
            logger.warning(
                "Bazefield API request failed | url=%s | status=%s | response=%s",
                r.url,
                r.status_code,
                r.text[:500],
            )
            raise

        return r.json()

    def list_sites(self) -> Dict:
        """List available sites for the configured operator.

        Returns:
            Dict: Raw API response containing site objects.

        Example:
            ```python
            from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

            bz = BazefieldApi(operator="ox2", farm_code="MLK")
            sites = bz.list_sites()
            ```
        """
        return self._get(
            "objects/structure",
            params={"type": 102, "category": 21},
        )

    def list_turbines(self, site_object_id: str) -> Dict:
        """
        List turbines related to a specific site.

        Args:
            site_object_id (str):
                Bazefield site object ID.

        Returns:
            Dict:
                Raw API response containing turbine objects.
        """
        return self._get(
            "objects/related",
            params={"objectIds": site_object_id},
        )

    def list_points(self, level: str = "turbine") -> Dict[str, Dict]:
        """
        List available measurement points for turbines at the site.

        Currently, point schemas are resolved for the first turbine found.

        Args:
            level: Object level to query. Only ``"turbine"`` is supported.

        Returns:
            Dict[str, Dict]: Mapping from turbine object ID to point schema
            definitions. Empty dictionary if no turbines are found.

        Example:
            ```python
            from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

            bz = BazefieldApi(operator="ox2", farm_code="MLK")
            sites = bz.list_points()
            ```
        """
        turbines = self.list_turbines(self.site_id)
        try:
            turbine_id = turbines["data"][self.site_id]["mdmobjects"]["200"][0][
                "objectId"
            ]
        except (KeyError, IndexError):
            logger.warning(
                "Unable to list points | no turbines found | site_id=%s",
                self.site_id,
            )
            return {}

        return {
            turbine_id: self._get(
                "objects/getschemas",
                params={"objectIds": turbine_id},
            )
        }

    def _resolve_site_id(self) -> str:
        response = self.list_sites()

        for entry in response.get("data", []):
            attrs = entry.get("attributes", {})
            if (
                attrs.get("objectTypeId") == "102"
                and attrs.get("objectCategoryId") == "21"
                and attrs.get("shortName", "").upper() == self.farm_code.upper()
            ):
                return entry["objectId"]

        logger.warning(
            "Bazefield site not found | farm_code=%s | sites_returned=%d",
            self.farm_code,
            len(response.get("data", [])),
        )
        raise RuntimeError(f"Site objectId not found for {self.farm_code}")

    def _resolve_turbine_mpid_map(self) -> dict[str, str]:
        mapping = {}

        turbines = self.list_turbines(self.site_id)

        for site_block in turbines.get("data", {}).values():
            for t in site_block.get("mdmobjects", {}).get("200", []):
                attrs = t.get("attributes", {})
                short_name = attrs.get("shortName")
                object_id = t.get("objectId")

                if short_name in self.mpid_map:
                    mapping[object_id] = self.mpid_map[short_name]
                elif "all" in self.mpid_map:
                    mapping[object_id] = self.mpid_map["all"]
                else:
                    logger.warning(
                        "Missing MPID mapping | turbine_object_id=%s | short_name=%s",
                        object_id,
                        short_name,
                    )

        if not mapping:
            logger.warning(
                "No turbine MPIDs resolved | farm_code=%s",
                self.farm_code,
            )
            raise RuntimeError(f"No turbine MPIDs resolved for farm {self.farm_code}")

        return mapping

    def get_historical_data(
        self,
        data_point: str,
        start_time_lb_utc: pd.Timestamp,
        stop_time_lb_Utc: pd.Timestamp,
        interval: str = "15m",
        aggregate: int = 3,
        aggregation_type: str = "sum",
    ) -> pd.DataFrame:
        """Fetch historical turbine-level time-series data.

        Data is fetched per turbine and aggregated across turbines according
        to the configured MPID mapping.

        Args:
            data_point: Bazefield point identifier. The data to fetch from the turbines. Example: ActivePower
            start_time_lb_utc: Inclusive start timestamp (UTC).
            stop_time_lb_Utc: Inclusive end timestamp (UTC).
            interval: Time resolution (e.g. ``"15m"``, ``"1h"``, ``"1d"``).
            aggregate: Bazefield aggregation flag.
            aggregation_type: Aggregation method (``"sum"`` or ``"average"``). This matter for the type of data being fetched. For wind speed or temperature use ``"average"``, for energy production use ``"sum"``.

        Returns:
            pandas.DataFrame: Aggregated time-series data. May be empty.

        Raises:
            ValueError: If an unsupported aggregation type is provided.

        Example:
            ```python
            from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

            bz = BazefieldApi(operator="ox2", farm_code="MLK")

            start_time = get_start_first_quarter_stockholm_day_utc("past", 19)
            end_time = get_start_first_quarter_stockholm_day_utc("past", 15)

            df = bz.get_historical_data_per_mpid(
                data_point="CALC-AverageWindSpeed",
                start_time_lb_utc=start_time,
                stop_time_lb_Utc=end_time,
                aggregation_type="average",
            )
            ```
        """

        from_time = (
            pd.to_datetime(start_time_lb_utc, utc=True) + pd.Timedelta(seconds=900)
        ).strftime("%Y-%m-%dT%H:%M:%SZ")

        to_time = (
            pd.to_datetime(stop_time_lb_Utc, utc=True) + pd.Timedelta(seconds=900)
        ).strftime("%Y-%m-%dT%H:%M:%SZ")

        mpid_frames = {}

        for turbine_object_id, mpid in self.turbine_object_to_mpid.items():
            payload = self._get(
                "objects/timeseries/aggregated",
                params={
                    "objectIds": turbine_object_id,
                    "points": data_point,
                    "aggregates": aggregate,
                    "from": from_time,
                    "to": to_time,
                    "interval": interval,
                },
            )

            df = self._to_dataframe(payload, mpid, interval)

            if df.empty:
                logger.warning(
                    "No data returned | turbine_object_id=%s | mpid=%s | point=%s | from=%s | to=%s",
                    turbine_object_id,
                    mpid,
                    data_point,
                    from_time,
                    to_time,
                )
                continue

            mpid_frames.setdefault(mpid, []).append(df)

        combined_frames = []

        for mpid, frames in mpid_frames.items():
            combined_df = pd.concat(frames, ignore_index=True)

            if aggregation_type == "sum":
                agg_df = combined_df.groupby(
                    [
                        "start_time_lb_utc",
                        "stop_time_lb_utc",
                        "variable_id",
                        "resolution_seconds",
                    ],
                    as_index=False,
                )["variable_value"].sum()
            elif aggregation_type == "average":
                agg_df = combined_df.groupby(
                    [
                        "start_time_lb_utc",
                        "stop_time_lb_utc",
                        "variable_id",
                        "resolution_seconds",
                    ],
                    as_index=False,
                )["variable_value"].mean()
            else:
                raise ValueError(f"Unsupported aggregation_type: {aggregation_type}")

            combined_frames.append(agg_df)

        if not combined_frames:
            logger.warning(
                "No historical data produced | point=%s | from=%s | to=%s",
                data_point,
                from_time,
                to_time,
            )
            return pd.DataFrame()

        return pd.concat(combined_frames, ignore_index=True)

    @staticmethod
    def _to_dataframe(response: dict, mpid: str, interval: str) -> pd.DataFrame:
        rows = []

        if interval.endswith("m"):
            resolution_seconds = int(interval[:-1]) * 60
        elif interval.endswith("h"):
            resolution_seconds = int(interval[:-1]) * 3600
        elif interval.endswith("d"):
            resolution_seconds = int(interval[:-1]) * 86400
        else:
            raise ValueError(f"Unsupported interval: {interval}")

        interval_ms = resolution_seconds * 1000

        for obj in response.get("objects", {}).values():
            for point_entries in obj.get("points", {}).values():
                for entry in point_entries:
                    for ts in entry.get("timeSeries", []):
                        stop_ms = ts.get("t")
                        if stop_ms is None:
                            continue

                        value = (
                            ts.get("v")
                            if "v" in ts
                            else (
                                ts.get("value", {}).get("v")
                                if isinstance(ts.get("value"), dict)
                                else ts.get("value")
                            )
                        )

                        if value is None:
                            continue

                        rows.append(
                            {
                                "start_time_lb_utc": pd.to_datetime(
                                    stop_ms - interval_ms, unit="ms", utc=True
                                ),
                                "stop_time_lb_utc": pd.to_datetime(
                                    stop_ms, unit="ms", utc=True
                                ),
                                "variable_id": mpid,
                                "variable_value": value,
                                "resolution_seconds": resolution_seconds,
                            }
                        )

        if not rows:
            logger.warning(
                "Parsed zero rows | mpid=%s | interval=%s | response_keys=%s",
                mpid,
                interval,
                list(response.keys()),
            )

        return pd.DataFrame(rows)

__init__(operator, farm_code)

Initialize a Bazefield API client for a specific operator and wind farm.

During initialization the client:

• Loads configuration from disk
• Retrieves API credentials from Azure Key Vault
• Creates an authenticated HTTP session
• Resolves the site object ID
• Resolves turbine-to-MPID mappings

Parameters:

Name	Type	Description	Default
operator	str	Operator name.	required
farm_code	str	Wind farm code as defined in the Bazefield configuration.	required

Raises:

Type	Description
KeyError	If the operator or farm code is missing from the configuration.
RuntimeError	If the site or turbine MPIDs cannot be resolved.

Example

from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

bz = BazefieldApi(operator="ox2", farm_code="MLK")

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def __init__(self, operator: str, farm_code: str):
    """
    Initialize a Bazefield API client for a specific operator and wind farm.

    During initialization the client:

    - Loads configuration from disk
    - Retrieves API credentials from Azure Key Vault
    - Creates an authenticated HTTP session
    - Resolves the site object ID
    - Resolves turbine-to-MPID mappings

    Args:
        operator: Operator name.
        farm_code: Wind farm code as defined in the Bazefield configuration.

    Raises:
        KeyError: If the operator or farm code is missing from the configuration.
        RuntimeError: If the site or turbine MPIDs cannot be resolved.

    Example:
        ```python
        from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

        bz = BazefieldApi(operator="ox2", farm_code="MLK")
        ```
    """
    self.operator = operator
    self.farm_code = farm_code

    config = load_bazefield_config()
    self.cfg = config[operator][farm_code]
    self.mpid_map = self.cfg["wtg_number_mpid_map"]

    api_key = get_secret(self.cfg["secret"])
    self.base_url, self.verify_ssl = get_operator_connection(operator)

    self.session = requests.Session()
    self.session.headers.update(
        {
            "Authorization": f"Bearer {api_key}",
            "Accept": "application/json",
        }
    )

    self.site_id = self._resolve_site_id()
    self.turbine_object_to_mpid = self._resolve_turbine_mpid_map()

get_historical_data(data_point, start_time_lb_utc, stop_time_lb_Utc, interval='15m', aggregate=3, aggregation_type='sum')

Fetch historical turbine-level time-series data.

Data is fetched per turbine and aggregated across turbines according to the configured MPID mapping.

Parameters:

Name	Type	Description	Default
data_point	str	Bazefield point identifier. The data to fetch from the turbines. Example: ActivePower	required
start_time_lb_utc	Timestamp	Inclusive start timestamp (UTC).	required
stop_time_lb_Utc	Timestamp	Inclusive end timestamp (UTC).	required
interval	str	Time resolution (e.g. "15m", "1h", "1d").	'15m'
aggregate	int	Bazefield aggregation flag.	3
aggregation_type	str	Aggregation method ("sum" or "average"). This matter for the type of data being fetched. For wind speed or temperature use "average", for energy production use "sum".	'sum'

Returns:

Type	Description
DataFrame	pandas.DataFrame: Aggregated time-series data. May be empty.

Raises:

Type	Description
ValueError	If an unsupported aggregation type is provided.

Example

from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

bz = BazefieldApi(operator="ox2", farm_code="MLK")

start_time = get_start_first_quarter_stockholm_day_utc("past", 19)
end_time = get_start_first_quarter_stockholm_day_utc("past", 15)

df = bz.get_historical_data_per_mpid(
    data_point="CALC-AverageWindSpeed",
    start_time_lb_utc=start_time,
    stop_time_lb_Utc=end_time,
    aggregation_type="average",
)

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def get_historical_data(
    self,
    data_point: str,
    start_time_lb_utc: pd.Timestamp,
    stop_time_lb_Utc: pd.Timestamp,
    interval: str = "15m",
    aggregate: int = 3,
    aggregation_type: str = "sum",
) -> pd.DataFrame:
    """Fetch historical turbine-level time-series data.

    Data is fetched per turbine and aggregated across turbines according
    to the configured MPID mapping.

    Args:
        data_point: Bazefield point identifier. The data to fetch from the turbines. Example: ActivePower
        start_time_lb_utc: Inclusive start timestamp (UTC).
        stop_time_lb_Utc: Inclusive end timestamp (UTC).
        interval: Time resolution (e.g. ``"15m"``, ``"1h"``, ``"1d"``).
        aggregate: Bazefield aggregation flag.
        aggregation_type: Aggregation method (``"sum"`` or ``"average"``). This matter for the type of data being fetched. For wind speed or temperature use ``"average"``, for energy production use ``"sum"``.

    Returns:
        pandas.DataFrame: Aggregated time-series data. May be empty.

    Raises:
        ValueError: If an unsupported aggregation type is provided.

    Example:
        ```python
        from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

        bz = BazefieldApi(operator="ox2", farm_code="MLK")

        start_time = get_start_first_quarter_stockholm_day_utc("past", 19)
        end_time = get_start_first_quarter_stockholm_day_utc("past", 15)

        df = bz.get_historical_data_per_mpid(
            data_point="CALC-AverageWindSpeed",
            start_time_lb_utc=start_time,
            stop_time_lb_Utc=end_time,
            aggregation_type="average",
        )
        ```
    """

    from_time = (
        pd.to_datetime(start_time_lb_utc, utc=True) + pd.Timedelta(seconds=900)
    ).strftime("%Y-%m-%dT%H:%M:%SZ")

    to_time = (
        pd.to_datetime(stop_time_lb_Utc, utc=True) + pd.Timedelta(seconds=900)
    ).strftime("%Y-%m-%dT%H:%M:%SZ")

    mpid_frames = {}

    for turbine_object_id, mpid in self.turbine_object_to_mpid.items():
        payload = self._get(
            "objects/timeseries/aggregated",
            params={
                "objectIds": turbine_object_id,
                "points": data_point,
                "aggregates": aggregate,
                "from": from_time,
                "to": to_time,
                "interval": interval,
            },
        )

        df = self._to_dataframe(payload, mpid, interval)

        if df.empty:
            logger.warning(
                "No data returned | turbine_object_id=%s | mpid=%s | point=%s | from=%s | to=%s",
                turbine_object_id,
                mpid,
                data_point,
                from_time,
                to_time,
            )
            continue

        mpid_frames.setdefault(mpid, []).append(df)

    combined_frames = []

    for mpid, frames in mpid_frames.items():
        combined_df = pd.concat(frames, ignore_index=True)

        if aggregation_type == "sum":
            agg_df = combined_df.groupby(
                [
                    "start_time_lb_utc",
                    "stop_time_lb_utc",
                    "variable_id",
                    "resolution_seconds",
                ],
                as_index=False,
            )["variable_value"].sum()
        elif aggregation_type == "average":
            agg_df = combined_df.groupby(
                [
                    "start_time_lb_utc",
                    "stop_time_lb_utc",
                    "variable_id",
                    "resolution_seconds",
                ],
                as_index=False,
            )["variable_value"].mean()
        else:
            raise ValueError(f"Unsupported aggregation_type: {aggregation_type}")

        combined_frames.append(agg_df)

    if not combined_frames:
        logger.warning(
            "No historical data produced | point=%s | from=%s | to=%s",
            data_point,
            from_time,
            to_time,
        )
        return pd.DataFrame()

    return pd.concat(combined_frames, ignore_index=True)

list_points(level='turbine')

List available measurement points for turbines at the site.

Currently, point schemas are resolved for the first turbine found.

Parameters:

Name	Type	Description	Default
level	str	Object level to query. Only "turbine" is supported.	'turbine'

Returns:

Type	Description
Dict[str, Dict]	Dict[str, Dict]: Mapping from turbine object ID to point schema
Dict[str, Dict]	definitions. Empty dictionary if no turbines are found.

Example

from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

bz = BazefieldApi(operator="ox2", farm_code="MLK")
sites = bz.list_points()

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def list_points(self, level: str = "turbine") -> Dict[str, Dict]:
    """
    List available measurement points for turbines at the site.

    Currently, point schemas are resolved for the first turbine found.

    Args:
        level: Object level to query. Only ``"turbine"`` is supported.

    Returns:
        Dict[str, Dict]: Mapping from turbine object ID to point schema
        definitions. Empty dictionary if no turbines are found.

    Example:
        ```python
        from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

        bz = BazefieldApi(operator="ox2", farm_code="MLK")
        sites = bz.list_points()
        ```
    """
    turbines = self.list_turbines(self.site_id)
    try:
        turbine_id = turbines["data"][self.site_id]["mdmobjects"]["200"][0][
            "objectId"
        ]
    except (KeyError, IndexError):
        logger.warning(
            "Unable to list points | no turbines found | site_id=%s",
            self.site_id,
        )
        return {}

    return {
        turbine_id: self._get(
            "objects/getschemas",
            params={"objectIds": turbine_id},
        )
    }

list_sites()

List available sites for the configured operator.

Returns:

Name	Type	Description
Dict	Dict	Raw API response containing site objects.

Example

from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

bz = BazefieldApi(operator="ox2", farm_code="MLK")
sites = bz.list_sites()

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def list_sites(self) -> Dict:
    """List available sites for the configured operator.

    Returns:
        Dict: Raw API response containing site objects.

    Example:
        ```python
        from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

        bz = BazefieldApi(operator="ox2", farm_code="MLK")
        sites = bz.list_sites()
        ```
    """
    return self._get(
        "objects/structure",
        params={"type": 102, "category": 21},
    )

list_turbines(site_object_id)

List turbines related to a specific site.

Parameters:

Name	Type	Description	Default
site_object_id	str	Bazefield site object ID.	required

Returns:

Name	Type	Description
Dict	Dict	Raw API response containing turbine objects.

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def list_turbines(self, site_object_id: str) -> Dict:
    """
    List turbines related to a specific site.

    Args:
        site_object_id (str):
            Bazefield site object ID.

    Returns:
        Dict:
            Raw API response containing turbine objects.
    """
    return self._get(
        "objects/related",
        params={"objectIds": site_object_id},
    )