bazefield_api

This is a wrapper to communicate with the BazeField API.

BazefieldApi

Client for the Bazefield REST API bound to an operator.

The client loads every asset configured for an operator. An operator may expose several key-vault keys (each with its own subdomain and API key); requests are issued per key in a loop and the results are combined into a single :class:pandas.DataFrame. Data is aggregated per MPID (the key strings in each key's assets mapping).

Attributes:

Name	Type	Description
operator		Operator identifier (top-level key in the configuration).
mpid		Optional asset identifier restricting the client to a single asset.
device_ids	List[str]	Flat list of all turbine object ids loaded from the config.

Notes

Site ids, turbine (device) ids and MPID mappings come from the configuration; no API call is needed to resolve them on instantiation. 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 an operator.

    The client loads every asset configured for an operator. An operator may
    expose several key-vault keys (each with its own subdomain and API key);
    requests are issued per key in a loop and the results are combined into a
    single :class:`pandas.DataFrame`. Data is aggregated per MPID (the key
    strings in each key's ``assets`` mapping).

    Attributes:
        operator: Operator identifier (top-level key in the configuration).
        mpid: Optional asset identifier restricting the client to a single asset.
        device_ids: Flat list of all turbine object ids loaded from the config.

    Notes:
        Site ids, turbine (device) ids and MPID mappings come from the
        configuration; no API call is needed to resolve them on instantiation.
        All time-series methods return :class:`pandas.DataFrame` objects.
    """

    def __init__(
        self,
        operator: str,
        mpid: Optional[str] = None,
        logger: Optional[Logger] = None,
    ) -> None:
        """Initialize a Bazefield API client for an operator.

        Args:
            operator: Operator identifier (top-level key in the configuration).
            mpid: Optional asset identifier (MPID). When provided, only the
                turbine object ids configured for that asset are loaded.
            logger: Optional loguru logger. A default is created when omitted.

        Raises:
            ValueError: If the operator is unknown, a key is missing a subdomain,
                or no device ids are configured for the requested selection.

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

            bz = BazefieldApi("ox2")
            bz_single = BazefieldApi("ox2", mpid="P01MLK000")
            ```
        """
        if not logger:
            logger = get_logger(
                team="physical_operations", application_name="BazefieldApi"
            )
        self._logger = logger
        self.operator = operator
        self.mpid = mpid

        operator_cfg = config.get(operator)
        if not isinstance(operator_cfg, dict):
            self._logger.error(f"Unknown operator: {operator}")
            raise ValueError(f"Unknown operator: {operator}")

        self._keys: List[Dict[str, Any]] = []
        for key_secret, key_cfg in operator_cfg.items():
            if not isinstance(key_cfg, dict):
                continue
            ctx = self._build_key_context(key_secret, key_cfg, mpid)
            if ctx is not None:
                self._keys.append(ctx)

        self.device_ids: List[str] = [
            did for key in self._keys for did in key["device_to_mpid"]
        ]

        if not self.device_ids:
            msg = f"No device ids configured for operator '{operator}'" + (
                f" and mpid '{mpid}'" if mpid is not None else ""
            )
            self._logger.error(msg)
            raise ValueError(msg)

        self._logger.debug(
            f"Initialized BazefieldApi for operator '{operator}' with "
            f"{len(self.device_ids)} device id(s) across {len(self._keys)} key(s)"
        )

    # ---------------- Init helpers ----------------
    @staticmethod
    def _build_device_map(
        assets: Dict[str, Any],
        mpid: Optional[str],
    ) -> Tuple[Dict[str, str], Dict[str, Dict[str, Any]]]:
        """Build device→MPID and MPID→metadata maps from an asset config dict.

        Args:
            assets: Mapping of MPID → asset config dict.
            mpid: When set, only assets matching this MPID are included.

        Returns:
            Tuple of (device_to_mpid, asset_meta), where ``asset_meta`` maps each
            MPID to its ``site_name``, ``site_id`` and ``object_key``.
        """
        device_to_mpid: Dict[str, str] = {}
        asset_meta: Dict[str, Dict[str, Any]] = {}
        for asset_mpid, asset in assets.items():
            if mpid is not None and asset_mpid != mpid:
                continue
            if not isinstance(asset, dict):
                continue
            asset_meta[asset_mpid] = {
                "site_name": asset.get("site_name") or asset_mpid,
                "site_id": asset.get("site_id"),
                "object_key": asset.get("object_key"),
            }
            for did in asset.get("device_ids") or []:
                device_to_mpid[str(did)] = asset_mpid
        return device_to_mpid, asset_meta

    def _build_key_context(
        self,
        key_secret: str,
        key_cfg: Dict[str, Any],
        mpid: Optional[str],
    ) -> Optional[Dict[str, Any]]:
        """Build a key context dict from a single key configuration entry.

        Args:
            key_secret: Key vault secret name (passed to :func:`get_secret`).
            key_cfg: Key configuration dict containing ``subdomain`` and ``assets``.
            mpid: Optional MPID filter; when given, only the matching asset is loaded.

        Returns:
            Key context dict, or ``None`` if no devices match the filter.

        Raises:
            ValueError: If ``subdomain`` is missing or the retrieved API key is not
                a string.
        """
        subdomain = key_cfg.get("subdomain") or ""
        if not subdomain:
            self._logger.error(
                f"Missing subdomain for key '{key_secret}' of operator "
                f"'{self.operator}'"
            )
            raise ValueError(f"Missing subdomain for key: {key_secret}")

        assets = key_cfg.get("assets") or {}
        device_to_mpid, asset_meta = self._build_device_map(assets, mpid)
        if not device_to_mpid:
            return None

        api_key = get_secret(key_secret)
        if not isinstance(api_key, str):
            self._logger.error(f"api_key must be a string for key: {key_secret}")
            raise ValueError(f"api_key must be a string for key: {key_secret}")

        return {
            "key_secret": key_secret,
            "subdomain": subdomain,
            "base_url": f"{subdomain.rstrip('/')}/BazeField.Services/api",
            "verify_ssl": True,
            "headers": {
                "Authorization": f"Bearer {api_key}",
                "Accept": "application/json",
            },
            "device_to_mpid": device_to_mpid,
            "asset_meta": asset_meta,
        }

    # ---------------- HTTP ----------------
    def _get(
        self,
        path: str,
        base_url: str,
        headers: Dict[str, str],
        verify_ssl: bool = True,
        params: Optional[Dict[str, Any]] = None,
    ) -> Any:
        """Perform a GET request against the Bazefield API.

        Args:
            path: Path relative to the base API URL (e.g. ``"objects/structure"``).
            base_url: Base URL of the key the request targets.
            headers: HTTP headers (including the bearer token) for that key.
            verify_ssl: Whether to verify SSL certificates.
            params: Optional query parameters.

        Returns:
            Any: Parsed JSON response, typically a ``dict``.

        Raises:
            requests.HTTPError: If the request fails (non-2xx).
            requests.Timeout: If the request exceeds the 30 s timeout.
        """
        url = f"{base_url}/{path.lstrip('/')}"
        try:
            r = requests.get(
                url,
                headers=headers,
                params=params,
                timeout=(10, 30),
                verify=verify_ssl,
            )
            r.raise_for_status()
        except requests.HTTPError as exc:
            status = getattr(exc.response, "status_code", "?")
            body = getattr(exc.response, "text", "")[:500]
            self._logger.error(
                f"HTTP error on GET {url}: status={status} response={body}"
            )
            raise
        return r.json()

    # ---------------- Time-series ----------------
    @staticmethod
    def _interval_to_seconds(interval: str) -> int:
        """Convert an interval string (e.g. ``"15m"``, ``"1h"``, ``"1d"``) to seconds."""
        if interval.endswith("m"):
            return int(interval[:-1]) * 60
        if interval.endswith("h"):
            return int(interval[:-1]) * 3600
        if interval.endswith("d"):
            return int(interval[:-1]) * 86400
        raise ValueError(f"Unsupported interval: {interval}")

    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 time-series data aggregated per MPID.

        All turbine object ids exposed by the same key are fetched in a single
        request; results are mapped back to their MPID and aggregated. When an
        operator spans several keys, each key is queried in turn and the frames
        are concatenated.

        Args:
            data_point: Bazefield point identifier (e.g. ``"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: How to aggregate across turbines of the same MPID.
                Either ``"sum"`` (e.g. energy production) or ``"average"``
                (e.g. wind speed, temperature).

        Returns:
            pandas.DataFrame: Aggregated time-series with columns
            ``start_time_lb_utc``, ``stop_time_lb_utc``, ``variable_id``,
            ``variable_value``, ``resolution_seconds``. May be empty.

        Raises:
            ValueError: If an unsupported ``aggregation_type`` or ``interval`` is
                provided.

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

            bz = BazefieldApi("ox2")
            df = bz.get_historical_data(
                data_point="CALC-AverageWindSpeed",
                start_time_lb_utc="2026-06-30T00:00:00Z",
                stop_time_lb_utc="2026-06-30T06:00:00Z",
                aggregation_type="average",
            )
            ```
        """
        if aggregation_type not in ("sum", "average"):
            raise ValueError(f"Unsupported aggregation_type: {aggregation_type}")
        # Validate the interval early (raises ValueError on bad input).
        self._interval_to_seconds(interval)

        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")

        frames: List[pd.DataFrame] = []
        for key in self._keys:
            device_to_mpid = key["device_to_mpid"]
            object_ids = list(device_to_mpid)
            self._logger.debug(
                f"Fetching '{data_point}' for key '{key['key_secret']}' "
                f"({len(object_ids)} device(s))"
            )

            try:
                payload = self._get(
                    "objects/timeseries/aggregated",
                    key["base_url"],
                    key["headers"],
                    key["verify_ssl"],
                    params={
                        "objectIds": ",".join(object_ids),
                        "points": data_point,
                        "aggregates": aggregate,
                        "from": from_time,
                        "to": to_time,
                        "interval": interval,
                    },
                )
            except requests.HTTPError:
                continue

            df = self._to_dataframe(payload, device_to_mpid, interval)
            if df.empty:
                self._logger.warning(
                    f"No data returned | key={key['key_secret']} "
                    f"point={data_point} from={from_time} to={to_time}"
                )
                continue
            frames.append(df)

        return self._combine_frames(frames, aggregation_type)

    @staticmethod
    def _combine_frames(
        frames: List[pd.DataFrame], aggregation_type: str
    ) -> pd.DataFrame:
        """Concatenate per-key frames and aggregate duplicate rows per MPID."""
        valid = [f for f in frames if f is not None and not f.empty]
        if not valid:
            return pd.DataFrame(columns=DF_COLUMNS)
        method = "mean" if aggregation_type == "average" else "sum"
        df = pd.concat(valid, ignore_index=True)
        agg = df.groupby(
            [
                "start_time_lb_utc",
                "stop_time_lb_utc",
                "variable_id",
                "resolution_seconds",
            ],
            as_index=False,
        ).agg(variable_value=("variable_value", method))
        return agg[DF_COLUMNS]

    def _to_dataframe(
        self,
        response: Dict[str, Any],
        device_to_mpid: Dict[str, str],
        interval: str,
    ) -> pd.DataFrame:
        """Parse a timeseries response into a normalized DataFrame.

        Each object in the response is mapped to its MPID via ``device_to_mpid``;
        objects that do not belong to the key are ignored.
        """
        resolution_seconds = self._interval_to_seconds(interval)
        interval_ms = resolution_seconds * 1000

        objects = response.get("objects", {})
        if not isinstance(objects, dict):
            return pd.DataFrame(columns=DF_COLUMNS)

        rows: List[Dict[str, Any]] = []
        for object_id, obj in objects.items():
            mpid = device_to_mpid.get(str(object_id))
            if mpid is None or not isinstance(obj, dict):
                continue
            for ts in self._iter_timeseries(obj):
                row = self._build_row(ts, mpid, interval_ms, resolution_seconds)
                if row is not None:
                    rows.append(row)

        if not rows:
            return pd.DataFrame(columns=DF_COLUMNS)
        return pd.DataFrame(rows)

    @staticmethod
    def _iter_timeseries(obj: Dict[str, Any]) -> Iterator[Dict[str, Any]]:
        """Yield each timeSeries entry contained in an object's ``points``."""
        for point_entries in obj.get("points", {}).values():
            for entry in point_entries:
                for ts in entry.get("timeSeries", []):
                    yield ts

    @staticmethod
    def _extract_value(ts: Dict[str, Any]) -> Any:
        """Extract the numeric value from a timeSeries entry.

        Handles both flat (``{"v": ...}``) and nested
        (``{"value": {"v": ...}}`` / ``{"value": ...}``) shapes.
        """
        if "v" in ts:
            return ts.get("v")
        value = ts.get("value")
        if isinstance(value, dict):
            return value.get("v")
        return value

    @staticmethod
    def _build_row(
        ts: Dict[str, Any],
        mpid: str,
        interval_ms: int,
        resolution_seconds: int,
    ) -> Optional[Dict[str, Any]]:
        """Build a single normalized row from a timeSeries entry.

        Returns ``None`` when the entry has no timestamp or no value.
        """
        stop_ms = ts.get("t")
        if stop_ms is None:
            return None
        value = BazefieldApi._extract_value(ts)
        if value is None:
            return None
        return {
            "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,
        }

    # ---------------- Onboarding helpers ----------------
    def list_sites(self) -> List[Dict[str, Any]]:
        """List every site reachable across the operator's keys.

        Queries the Bazefield API directly (rather than the configuration) so
        that site object ids can be discovered when onboarding new assets. Each
        returned dictionary is annotated with the ``key_secret`` and
        ``subdomain`` it originates from.

        Returns:
            List[Dict[str, Any]]: Site entries (API payload plus ``key_secret``
            and ``subdomain``). Empty list if none found.
        """
        sites: List[Dict[str, Any]] = []
        for key in self._keys:
            data = self._get(
                "objects/structure",
                key["base_url"],
                key["headers"],
                key["verify_ssl"],
                params={"type": 102, "category": 21},
            )
            for entry in (data or {}).get("data", []):
                if isinstance(entry, dict):
                    sites.append(
                        {
                            **entry,
                            "key_secret": key["key_secret"],
                            "subdomain": key["subdomain"],
                        }
                    )
        return sites

    def list_turbines(
        self, site_object_id: str, key_secret: Optional[str] = None
    ) -> List[Dict[str, Any]]:
        """List turbine objects related to a site across the operator's keys.

        Args:
            site_object_id: Bazefield site object id.
            key_secret: Optional key vault secret name to restrict the lookup to
                a single key (useful when site ids overlap across keys).

        Returns:
            List[Dict[str, Any]]: Turbine entries (API payload plus ``key_secret``
            and ``subdomain``). Empty list if none found.
        """
        turbines: List[Dict[str, Any]] = []
        for key in self._keys:
            if key_secret is not None and key["key_secret"] != key_secret:
                continue
            data = self._get(
                "objects/related",
                key["base_url"],
                key["headers"],
                key["verify_ssl"],
                params={"objectIds": site_object_id},
            )
            for site_block in (data or {}).get("data", {}).values():
                for t in site_block.get("mdmobjects", {}).get("200", []):
                    if isinstance(t, dict):
                        turbines.append(
                            {
                                **t,
                                "key_secret": key["key_secret"],
                                "subdomain": key["subdomain"],
                            }
                        )
        return turbines

    def resolve_site_id(self, object_key: str) -> Optional[str]:
        """Resolve the Bazefield site object id for an ``object_key`` (short name).

        Intended for onboarding: discover the ``site_id`` to store in the config.

        Args:
            object_key: Site short name (e.g. ``"MLK"``).

        Returns:
            Optional[str]: The matching site object id, or ``None`` if not found.
        """
        for site in self.list_sites():
            attrs = site.get("attributes", {})
            if (
                attrs.get("objectTypeId") == "102"
                and attrs.get("objectCategoryId") == "21"
                and attrs.get("shortName", "").upper() == object_key.upper()
            ):
                return site.get("objectId")
        self._logger.warning(
            f"Site not found for object_key '{object_key}' (operator "
            f"'{self.operator}')"
        )
        return None

    def resolve_device_ids_for_site(self, site_object_id: str) -> List[str]:
        """Return all turbine object ids for a site.

        Intended for onboarding: pair a site id discovered via
        :meth:`resolve_site_id` with this method to obtain the ``device_ids``
        list for the new asset's configuration entry.

        Args:
            site_object_id: Bazefield site object id.

        Returns:
            List[str]: Turbine object ids for the site.
        """
        return [
            t["objectId"]
            for t in self.list_turbines(site_object_id)
            if t.get("objectId")
        ]

    def list_points(self, turbine_object_id: Optional[str] = None) -> Dict[str, Any]:
        """List available measurement points for a turbine.

        Args:
            turbine_object_id: Turbine object id to query. Defaults to the first
                configured device id when omitted.

        Returns:
            Dict[str, Any]: Point schema definitions, or an empty dict if the
            turbine cannot be located.
        """
        if turbine_object_id is None:
            if not self.device_ids:
                return {}
            turbine_object_id = self.device_ids[0]

        for key in self._keys:
            if turbine_object_id in key["device_to_mpid"]:
                return self._get(
                    "objects/getschemas",
                    key["base_url"],
                    key["headers"],
                    key["verify_ssl"],
                    params={"objectIds": turbine_object_id},
                )

        self._logger.warning(
            f"Turbine object id '{turbine_object_id}' not found in configuration "
            f"for operator '{self.operator}'"
        )
        return {}

__init__(operator, mpid=None, logger=None)

Initialize a Bazefield API client for an operator.

Parameters:

Name	Type	Description	Default
operator	str	Operator identifier (top-level key in the configuration).	required
mpid	Optional[str]	Optional asset identifier (MPID). When provided, only the turbine object ids configured for that asset are loaded.	None
logger	Optional[Logger]	Optional loguru logger. A default is created when omitted.	None

Raises:

Type	Description
ValueError	If the operator is unknown, a key is missing a subdomain, or no device ids are configured for the requested selection.

Example

from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

bz = BazefieldApi("ox2")
bz_single = BazefieldApi("ox2", mpid="P01MLK000")

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def __init__(
    self,
    operator: str,
    mpid: Optional[str] = None,
    logger: Optional[Logger] = None,
) -> None:
    """Initialize a Bazefield API client for an operator.

    Args:
        operator: Operator identifier (top-level key in the configuration).
        mpid: Optional asset identifier (MPID). When provided, only the
            turbine object ids configured for that asset are loaded.
        logger: Optional loguru logger. A default is created when omitted.

    Raises:
        ValueError: If the operator is unknown, a key is missing a subdomain,
            or no device ids are configured for the requested selection.

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

        bz = BazefieldApi("ox2")
        bz_single = BazefieldApi("ox2", mpid="P01MLK000")
        ```
    """
    if not logger:
        logger = get_logger(
            team="physical_operations", application_name="BazefieldApi"
        )
    self._logger = logger
    self.operator = operator
    self.mpid = mpid

    operator_cfg = config.get(operator)
    if not isinstance(operator_cfg, dict):
        self._logger.error(f"Unknown operator: {operator}")
        raise ValueError(f"Unknown operator: {operator}")

    self._keys: List[Dict[str, Any]] = []
    for key_secret, key_cfg in operator_cfg.items():
        if not isinstance(key_cfg, dict):
            continue
        ctx = self._build_key_context(key_secret, key_cfg, mpid)
        if ctx is not None:
            self._keys.append(ctx)

    self.device_ids: List[str] = [
        did for key in self._keys for did in key["device_to_mpid"]
    ]

    if not self.device_ids:
        msg = f"No device ids configured for operator '{operator}'" + (
            f" and mpid '{mpid}'" if mpid is not None else ""
        )
        self._logger.error(msg)
        raise ValueError(msg)

    self._logger.debug(
        f"Initialized BazefieldApi for operator '{operator}' with "
        f"{len(self.device_ids)} device id(s) across {len(self._keys)} key(s)"
    )

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

Fetch historical time-series data aggregated per MPID.

All turbine object ids exposed by the same key are fetched in a single request; results are mapped back to their MPID and aggregated. When an operator spans several keys, each key is queried in turn and the frames are concatenated.

Parameters:

Name	Type	Description	Default
data_point	str	Bazefield point identifier (e.g. "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	How to aggregate across turbines of the same MPID. Either "sum" (e.g. energy production) or "average" (e.g. wind speed, temperature).	'sum'

Returns:

Type	Description
DataFrame	pandas.DataFrame: Aggregated time-series with columns
DataFrame	start_time_lb_utc, stop_time_lb_utc, variable_id,
DataFrame	variable_value, resolution_seconds. May be empty.

Raises:

Type	Description
ValueError	If an unsupported aggregation_type or interval is provided.

Example

from physical_operations_utils.bazefield_utils.BazeFieldAPI import BazefieldApi

bz = BazefieldApi("ox2")
df = bz.get_historical_data(
    data_point="CALC-AverageWindSpeed",
    start_time_lb_utc="2026-06-30T00:00:00Z",
    stop_time_lb_utc="2026-06-30T06:00:00Z",
    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 time-series data aggregated per MPID.

    All turbine object ids exposed by the same key are fetched in a single
    request; results are mapped back to their MPID and aggregated. When an
    operator spans several keys, each key is queried in turn and the frames
    are concatenated.

    Args:
        data_point: Bazefield point identifier (e.g. ``"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: How to aggregate across turbines of the same MPID.
            Either ``"sum"`` (e.g. energy production) or ``"average"``
            (e.g. wind speed, temperature).

    Returns:
        pandas.DataFrame: Aggregated time-series with columns
        ``start_time_lb_utc``, ``stop_time_lb_utc``, ``variable_id``,
        ``variable_value``, ``resolution_seconds``. May be empty.

    Raises:
        ValueError: If an unsupported ``aggregation_type`` or ``interval`` is
            provided.

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

        bz = BazefieldApi("ox2")
        df = bz.get_historical_data(
            data_point="CALC-AverageWindSpeed",
            start_time_lb_utc="2026-06-30T00:00:00Z",
            stop_time_lb_utc="2026-06-30T06:00:00Z",
            aggregation_type="average",
        )
        ```
    """
    if aggregation_type not in ("sum", "average"):
        raise ValueError(f"Unsupported aggregation_type: {aggregation_type}")
    # Validate the interval early (raises ValueError on bad input).
    self._interval_to_seconds(interval)

    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")

    frames: List[pd.DataFrame] = []
    for key in self._keys:
        device_to_mpid = key["device_to_mpid"]
        object_ids = list(device_to_mpid)
        self._logger.debug(
            f"Fetching '{data_point}' for key '{key['key_secret']}' "
            f"({len(object_ids)} device(s))"
        )

        try:
            payload = self._get(
                "objects/timeseries/aggregated",
                key["base_url"],
                key["headers"],
                key["verify_ssl"],
                params={
                    "objectIds": ",".join(object_ids),
                    "points": data_point,
                    "aggregates": aggregate,
                    "from": from_time,
                    "to": to_time,
                    "interval": interval,
                },
            )
        except requests.HTTPError:
            continue

        df = self._to_dataframe(payload, device_to_mpid, interval)
        if df.empty:
            self._logger.warning(
                f"No data returned | key={key['key_secret']} "
                f"point={data_point} from={from_time} to={to_time}"
            )
            continue
        frames.append(df)

    return self._combine_frames(frames, aggregation_type)

list_points(turbine_object_id=None)

List available measurement points for a turbine.

Parameters:

Name	Type	Description	Default
turbine_object_id	Optional[str]	Turbine object id to query. Defaults to the first configured device id when omitted.	None

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Point schema definitions, or an empty dict if the
Dict[str, Any]	turbine cannot be located.

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def list_points(self, turbine_object_id: Optional[str] = None) -> Dict[str, Any]:
    """List available measurement points for a turbine.

    Args:
        turbine_object_id: Turbine object id to query. Defaults to the first
            configured device id when omitted.

    Returns:
        Dict[str, Any]: Point schema definitions, or an empty dict if the
        turbine cannot be located.
    """
    if turbine_object_id is None:
        if not self.device_ids:
            return {}
        turbine_object_id = self.device_ids[0]

    for key in self._keys:
        if turbine_object_id in key["device_to_mpid"]:
            return self._get(
                "objects/getschemas",
                key["base_url"],
                key["headers"],
                key["verify_ssl"],
                params={"objectIds": turbine_object_id},
            )

    self._logger.warning(
        f"Turbine object id '{turbine_object_id}' not found in configuration "
        f"for operator '{self.operator}'"
    )
    return {}

list_sites()

List every site reachable across the operator's keys.

Queries the Bazefield API directly (rather than the configuration) so that site object ids can be discovered when onboarding new assets. Each returned dictionary is annotated with the key_secret and subdomain it originates from.

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: Site entries (API payload plus key_secret
List[Dict[str, Any]]	and subdomain). Empty list if none found.

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def list_sites(self) -> List[Dict[str, Any]]:
    """List every site reachable across the operator's keys.

    Queries the Bazefield API directly (rather than the configuration) so
    that site object ids can be discovered when onboarding new assets. Each
    returned dictionary is annotated with the ``key_secret`` and
    ``subdomain`` it originates from.

    Returns:
        List[Dict[str, Any]]: Site entries (API payload plus ``key_secret``
        and ``subdomain``). Empty list if none found.
    """
    sites: List[Dict[str, Any]] = []
    for key in self._keys:
        data = self._get(
            "objects/structure",
            key["base_url"],
            key["headers"],
            key["verify_ssl"],
            params={"type": 102, "category": 21},
        )
        for entry in (data or {}).get("data", []):
            if isinstance(entry, dict):
                sites.append(
                    {
                        **entry,
                        "key_secret": key["key_secret"],
                        "subdomain": key["subdomain"],
                    }
                )
    return sites

list_turbines(site_object_id, key_secret=None)

List turbine objects related to a site across the operator's keys.

Parameters:

Name	Type	Description	Default
site_object_id	str	Bazefield site object id.	required
key_secret	Optional[str]	Optional key vault secret name to restrict the lookup to a single key (useful when site ids overlap across keys).	None

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: Turbine entries (API payload plus key_secret
List[Dict[str, Any]]	and subdomain). Empty list if none found.

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def list_turbines(
    self, site_object_id: str, key_secret: Optional[str] = None
) -> List[Dict[str, Any]]:
    """List turbine objects related to a site across the operator's keys.

    Args:
        site_object_id: Bazefield site object id.
        key_secret: Optional key vault secret name to restrict the lookup to
            a single key (useful when site ids overlap across keys).

    Returns:
        List[Dict[str, Any]]: Turbine entries (API payload plus ``key_secret``
        and ``subdomain``). Empty list if none found.
    """
    turbines: List[Dict[str, Any]] = []
    for key in self._keys:
        if key_secret is not None and key["key_secret"] != key_secret:
            continue
        data = self._get(
            "objects/related",
            key["base_url"],
            key["headers"],
            key["verify_ssl"],
            params={"objectIds": site_object_id},
        )
        for site_block in (data or {}).get("data", {}).values():
            for t in site_block.get("mdmobjects", {}).get("200", []):
                if isinstance(t, dict):
                    turbines.append(
                        {
                            **t,
                            "key_secret": key["key_secret"],
                            "subdomain": key["subdomain"],
                        }
                    )
    return turbines

resolve_device_ids_for_site(site_object_id)

Return all turbine object ids for a site.

Intended for onboarding: pair a site id discovered via :meth:resolve_site_id with this method to obtain the device_ids list for the new asset's configuration entry.

Parameters:

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

Returns:

Type	Description
List[str]	List[str]: Turbine object ids for the site.

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def resolve_device_ids_for_site(self, site_object_id: str) -> List[str]:
    """Return all turbine object ids for a site.

    Intended for onboarding: pair a site id discovered via
    :meth:`resolve_site_id` with this method to obtain the ``device_ids``
    list for the new asset's configuration entry.

    Args:
        site_object_id: Bazefield site object id.

    Returns:
        List[str]: Turbine object ids for the site.
    """
    return [
        t["objectId"]
        for t in self.list_turbines(site_object_id)
        if t.get("objectId")
    ]

resolve_site_id(object_key)

Resolve the Bazefield site object id for an object_key (short name).

Intended for onboarding: discover the site_id to store in the config.

Parameters:

Name	Type	Description	Default
object_key	str	Site short name (e.g. "MLK").	required

Returns:

Type	Description
Optional[str]	Optional[str]: The matching site object id, or None if not found.

Source code in physical_operations_utils/bazefield_utils/BazeFieldAPI.py

def resolve_site_id(self, object_key: str) -> Optional[str]:
    """Resolve the Bazefield site object id for an ``object_key`` (short name).

    Intended for onboarding: discover the ``site_id`` to store in the config.

    Args:
        object_key: Site short name (e.g. ``"MLK"``).

    Returns:
        Optional[str]: The matching site object id, or ``None`` if not found.
    """
    for site in self.list_sites():
        attrs = site.get("attributes", {})
        if (
            attrs.get("objectTypeId") == "102"
            and attrs.get("objectCategoryId") == "21"
            and attrs.get("shortName", "").upper() == object_key.upper()
        ):
            return site.get("objectId")
    self._logger.warning(
        f"Site not found for object_key '{object_key}' (operator "
        f"'{self.operator}')"
    )
    return None