greenbyte_api

This is a wrapper to communicate with the GreenByte API.

GreenByteApi

Client for the GreenByte Cloud 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.

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[int]	Flat list of all device ids loaded from the configuration.

Notes

Device ids and site ids come from the configuration; no API call is needed to resolve them. All time-series methods return :class:pandas.DataFrame objects.

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

class GreenByteApi:
    """Client for the GreenByte Cloud 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`.

    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 device ids loaded from the configuration.

    Notes:
        Device ids and site ids come from the configuration; no API call is
        needed to resolve them. 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 GreenByte API client for an operator.

        Args:
            operator: Operator identifier (top-level key in the configuration).
            mpid: Optional asset identifier (MPID). When provided, only the
                device 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.greenbyte_utils.GreenByteApi import GreenByteApi

            gb = GreenByteApi("eurus")
            gb_single = GreenByteApi("eurus", mpid="P01KAMA00")
            ```
        """
        if not logger:
            logger = get_logger(
                team="physical_operations", application_name="GreenByteApi"
            )
        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[int] = [
            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 GreenByteApi 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[int, str], Dict[str, str]]:
        """Build device→MPID and MPID→site-name 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, site_names).
        """
        device_to_mpid: Dict[int, str] = {}
        site_names: Dict[str, str] = {}
        for asset_mpid, asset in assets.items():
            if mpid is not None and asset_mpid != mpid:
                continue
            if not isinstance(asset, dict):
                continue
            site_names[asset_mpid] = asset.get("site_name") or asset_mpid
            for did in asset.get("device_ids") or []:
                if isinstance(did, int):
                    device_to_mpid[did] = asset_mpid
        return device_to_mpid, site_names

    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, site_names = 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"https://{subdomain}.greenbyte.cloud/api/2.2/",
            "headers": {
                "X-Api-Key": api_key,
                "Accept": "application/json",
                "User-Agent": "gb-multi-park/0.1",
            },
            "device_to_mpid": device_to_mpid,
            "site_names": site_names,
        }

    # ---------------- HTTP ----------------
    def get_data_endpoint(
        self,
        endpoint: str,
        base_url: str,
        headers: Dict[str, str],
        **params: Any,
    ) -> Any:
        """Perform a GET request against the GreenByte API.

        Args:
            endpoint: Path relative to the base API URL (e.g. ``"data"``).
            base_url: Base URL of the key the request targets.
            headers: HTTP headers (including the API key) for that key.
            **params: Optional query parameters.

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

        Raises:
            requests.HTTPError: If the request fails (non-2xx).
            requests.Timeout: If the request exceeds the 30 s timeout.
        """
        url = base_url + endpoint
        try:
            r = requests.get(
                url, headers=headers, params=(params or None), timeout=(10, 30)
            )
            r.raise_for_status()
        except requests.HTTPError as exc:
            self._logger.error(f"HTTP error on GET {url}: {exc}")
            raise
        return r.json()

    def post_data_endpoint(
        self,
        endpoint: str,
        payload: Dict[str, Any],
        base_url: str,
        headers: Dict[str, str],
    ) -> Any:
        """Perform a POST request against the GreenByte API.

        Args:
            endpoint: Path relative to the base API URL (e.g. ``"datasignals.json"``).
            payload: JSON-serializable request body.
            base_url: Base URL of the key the request targets.
            headers: HTTP headers (including the API key) for that key.

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

        Raises:
            requests.HTTPError: If the request fails (non-2xx).
            requests.Timeout: If the request exceeds the 30 s timeout.
        """
        url = base_url + endpoint
        try:
            r = requests.post(url, headers=headers, json=payload, timeout=30)
            r.raise_for_status()
        except requests.HTTPError as exc:
            self._logger.error(f"HTTP error on POST {url}: {exc}")
            raise
        return r.json()

    # ---------------- Resolution ----------------
    @staticmethod
    def _extract_device_id(item: Dict[str, Any]) -> Optional[int]:
        """Return the device id carried by a data/realtime response item.

        Handles several response shapes observed in the wild:
        - ``{"device": {"deviceId": N}}`` / ``{"device": {"id": N}}``
        - ``{"deviceId": N}``
        - ``{"aggregateId": N}`` (per-device aggregate response)
        - ``{"deviceIds": [N]}`` (single-element list)
        """
        device = item.get("device")
        if isinstance(device, dict):
            did = device.get("deviceId")
            if did is None:
                did = device.get("id")
            if isinstance(did, int):
                return did
        for key in ("deviceId", "aggregateId"):
            did = item.get(key)
            if isinstance(did, int):
                return did
        device_ids = item.get("deviceIds")
        if isinstance(device_ids, list) and len(device_ids) == 1:
            did = device_ids[0]
            if isinstance(did, int):
                return did
        return None

    def _pick_signal_id(self, key: Dict[str, Any], signal_title: str) -> Optional[int]:
        """Return the data-signal id for ``signal_title`` for a key's devices."""
        device_ids = sorted(key["device_to_mpid"])
        if not device_ids:
            return None
        signals = self.post_data_endpoint(
            "datasignals.json",
            {"deviceIds": [device_ids[0]]},
            key["base_url"],
            key["headers"],
        )
        signal_id = pick_signal_id_by_title(
            signals, device_ids[0], signal_title=signal_title
        )
        return signal_id if isinstance(signal_id, int) else None

    # ---------------- Onboarding helpers ----------------
    def get_sites(self, page: int = 1, page_size: int = 2000) -> List[Dict[str, Any]]:
        """List every site reachable across the operator's keys.

        This queries the GreenByte API directly (rather than the configuration)
        so that site ids can be discovered when onboarding new assets. Each
        returned dictionary is annotated with the ``key_secret`` and
        ``subdomain`` it originates from, which disambiguates sites when an
        operator spans several key-vault keys/subdomains.

        Args:
            page: Page number for pagination.
            page_size: Number of items per page (default 2000).

        Returns:
            List[Dict[str, Any]]: Site dictionaries (API payload plus
            ``key_secret`` and ``subdomain``). Empty list if none found.

        Example:
            ```python
            from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

            gb = GreenByteApi("eurus")
            sites = gb.get_sites()
            for s in sites:
                print(s["subdomain"], s.get("siteId"), s.get("title"))
            ```
        """
        sites: List[Dict[str, Any]] = []
        for key in self._keys:
            data = self.get_data_endpoint(
                "sites",
                key["base_url"],
                key["headers"],
                page=page,
                pageSize=page_size,
            )
            if not isinstance(data, list):
                continue
            for site in data:
                if isinstance(site, dict):
                    sites.append(
                        {
                            **site,
                            "key_secret": key["key_secret"],
                            "subdomain": key["subdomain"],
                        }
                    )
        return sites

    def get_devices(self, page: int = 1, page_size: int = 2000) -> List[Dict[str, Any]]:
        """List every device reachable across the operator's keys.

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

        Args:
            page: Page number for pagination.
            page_size: Number of items per page (default 2000).

        Returns:
            List[Dict[str, Any]]: Device dictionaries (API payload plus
            ``key_secret`` and ``subdomain``). Empty list if none found.

        Example:
            ```python
            from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

            gb = GreenByteApi("eurus")
            devices = gb.get_devices()
            print(len(devices))
            ```
        """
        devices: List[Dict[str, Any]] = []
        for key in self._keys:
            data = self.get_data_endpoint(
                "devices",
                key["base_url"],
                key["headers"],
                page=page,
                pageSize=page_size,
            )
            if not isinstance(data, list):
                continue
            for device in data:
                if isinstance(device, dict):
                    devices.append(
                        {
                            **device,
                            "key_secret": key["key_secret"],
                            "subdomain": key["subdomain"],
                        }
                    )
        return devices

    def get_device_ids_for_site(
        self,
        site_id: int,
        devices: Optional[List[Dict[str, Any]]] = None,
        subdomain: Optional[str] = None,
    ) -> List[int]:
        """Return all device ids associated with a site.

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

        Args:
            site_id: Site identifier to filter devices by.
            devices: Optional list of pre-fetched devices (as returned by
                :meth:`get_devices`). Fetched automatically when omitted.
            subdomain: Optional subdomain to further restrict matches, useful
                when site ids overlap across an operator's keys.

        Returns:
            List[int]: Integer device ids for the site.

        Example:
            ```python
            from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

            gb = GreenByteApi("eurus")
            device_ids = gb.get_device_ids_for_site(13)
            print(device_ids)
            ```
        """
        if devices is None:
            devices = self.get_devices(page=1, page_size=2000)
        out: List[int] = []
        for d in devices:
            if not isinstance(d, dict):
                continue
            if (d.get("site") or {}).get("siteId") != site_id:
                continue
            if subdomain is not None and d.get("subdomain") != subdomain:
                continue
            did = d.get("deviceId")
            if did is None:
                did = d.get("id")
            if isinstance(did, int):
                out.append(did)
        return out

    # ---------------- Fixed-resolution timeseries via GET /data ----------------
    def fetch_data(  # noqa: C901
        self,
        start_time: datetime,
        stop_time: datetime,
        signal_title: str = "power",
        resolution_seconds: int = 900,
        aggregate: str = "device",
        calculation: str = "sum",
        fill_strategy: str = "skip",
    ) -> pd.DataFrame:
        """Fetch fixed-resolution timeseries data (``GET /data``).

        Device ids are combined into a single request per key; per-device results
        are mapped back to their MPID, and results across keys are concatenated.

        Args:
            start_time: Inclusive start timestamp (UTC if naive).
            stop_time: Exclusive end timestamp.
            signal_title: Signal title to query (default ``"power"``).
            resolution_seconds: Data resolution in seconds (must exist in :data:`RESOLUTION_MAP`).
            aggregate: Aggregation mode passed to the API (default ``"device"``).
            calculation: Aggregation function (e.g., ``"sum"``).
            fill_strategy: Handling of non-numeric/null values:
                - ``"skip"`` (default) drops points.
                - ``"zero"`` substitutes ``0.0``.

        Returns:
            pandas.DataFrame: Normalized timeseries with columns
            ``start_time_lb_utc``, ``stop_time_lb_utc``, ``variable_id``,
            ``variable_value``, ``variable_unit``. May be empty.

        Raises:
            ValueError: If the time range is invalid or the resolution is unsupported.
        """
        if start_time >= stop_time:
            raise ValueError("start_time must be earlier than stop_time")
        if resolution_seconds not in RESOLUTION_MAP:
            raise ValueError(f"Unsupported resolution_seconds={resolution_seconds}")
        resolution = RESOLUTION_MAP[resolution_seconds]
        ts_start = to_rfc3339_utc(start_time)
        ts_end = to_rfc3339_utc(stop_time)

        frames: List[pd.DataFrame] = []
        for key in self._keys:
            device_to_mpid = key["device_to_mpid"]
            device_ids = sorted(device_to_mpid)
            self._logger.debug(
                f"Fetching data for key '{key['key_secret']}' "
                f"sites {sorted(set(key['site_names'].values()))}"
            )

            data_signal_id = self._pick_signal_id(key, signal_title)
            if data_signal_id is None:
                self._logger.warning(
                    f"No '{signal_title}' signal found for key "
                    f"'{key['key_secret']}'"
                )
                continue

            params = {
                "deviceIds": ",".join(map(str, device_ids)),
                "dataSignalIds": str(data_signal_id),
                "timestampStart": ts_start,
                "timestampEnd": ts_end,
                "useUtc": "true",
                "resolution": resolution,
                "aggregate": aggregate,
                "aggregateLevel": 0,
                "calculation": calculation,
            }

            try:
                data_items = self.get_data_endpoint(
                    "data", key["base_url"], key["headers"], **params
                )
            except Exception:
                continue
            if not isinstance(data_items, list):
                continue

            rows: List[Dict[str, Any]] = []
            for it in data_items:
                if not isinstance(it, dict):
                    continue
                device_id = self._extract_device_id(it)
                mpid = device_to_mpid.get(device_id) if device_id is not None else None
                if mpid is None:
                    continue
                ds = it.get("dataSignal") or {}
                unit = ds.get("unit") or "kW"
                data_map = it.get("data") or {}
                if not isinstance(data_map, dict):
                    continue
                for ts, val in sorted(data_map.items()):
                    if not is_number(val):
                        if fill_strategy == "zero":
                            val = 0.0
                        else:
                            continue
                    start = pd.to_datetime(ts, utc=True)
                    stop = start + pd.to_timedelta(resolution_seconds, unit="s")
                    rows.append(
                        {
                            "start_time_lb_utc": fmt_no_T(start),
                            "stop_time_lb_utc": fmt_no_T(stop),
                            "variable_value": float(val),
                            "variable_id": mpid,
                            "variable_unit": unit,
                        }
                    )

            if rows:
                frames.append(pd.DataFrame(rows))

        return self._combine_frames(frames, aggregation_method=calculation or "sum")

    def fetch_realtime_data(  # noqa: C901
        self,
        signal_title: Optional[str] = None,
        aggregate: Optional[str] = None,
        calculation: Optional[str] = None,
        fill_strategy: Optional[str] = None,
    ) -> pd.DataFrame:
        """Fetch instantaneous realtime points (``GET /realtimedata``).

        Device ids are combined into a single request per key; per-device results
        are mapped back to their MPID, and results across keys are concatenated.

        Args:
            signal_title: Signal title to query. Defaults to ``"power"``.
            aggregate: Aggregation mode passed to the API. Defaults to ``"device"``.
            calculation: Aggregation function (e.g., ``"sum"``).
            fill_strategy: Handling of non-numeric/null values:
                - ``"skip"`` (default) drops points.
                - ``"zero"`` substitutes ``0.0``.

        Returns:
            pandas.DataFrame: Same column contract as :meth:`fetch_data`; points are
            instantaneous (``start_time_lb_utc == stop_time_lb_utc``). May be empty.
        """
        now_utc = get_utc_now_custom_precision()
        stale_limit = timedelta(minutes=10)

        frames: List[pd.DataFrame] = []
        for key in self._keys:
            device_to_mpid = key["device_to_mpid"]
            site_names = key["site_names"]
            device_ids = sorted(device_to_mpid)
            self._logger.debug(
                f"Fetching realtime data for key '{key['key_secret']}' "
                f"sites {sorted(set(site_names.values()))}"
            )

            data_signal_id = self._pick_signal_id(key, signal_title or "power")
            if data_signal_id is None:
                self._logger.warning(
                    f"No '{signal_title or 'power'}' signal found for key "
                    f"'{key['key_secret']}'"
                )
                continue

            params = {
                "deviceIds": ",".join(map(str, device_ids)),
                "dataSignalIds": str(data_signal_id),
                "aggregate": aggregate or "device",
                "calculation": calculation or "sum",
            }

            try:
                rt_items = self.get_data_endpoint(
                    "realtimedata", key["base_url"], key["headers"], **params
                )
            except Exception:
                continue
            if not isinstance(rt_items, list):
                continue

            # Accumulate per MPID, ignoring sub-minute timestamp differences.
            # latest_ts tracks the most recent valid reading time for each MPID
            # so it can be used as the representative timestamp in the output.
            accumulator: Dict[str, float] = {}  # mpid -> summed value
            units: Dict[str, str] = {}  # mpid -> unit
            latest_ts: Dict[str, pd.Timestamp] = {}  # mpid -> most recent ts

            for item in rt_items:
                if not isinstance(item, dict):
                    continue
                device_id = self._extract_device_id(item)
                mpid = device_to_mpid.get(device_id) if device_id is not None else None
                if mpid is None:
                    continue
                ds = item.get("dataSignal") or {}
                unit = ds.get("unit") or units.get(mpid) or "kW"
                units[mpid] = unit
                data_map = item.get("data") or {}
                if not isinstance(data_map, dict):
                    continue
                for ts, val in data_map.items():
                    reading_time = pd.to_datetime(ts, utc=True)
                    age = now_utc - reading_time.to_pydatetime()
                    if age > stale_limit:
                        site_name = site_names.get(mpid, mpid)
                        self._logger.warning(
                            f"Stale realtime reading skipped — operator "
                            f"'{self.operator}', device {device_id} "
                            f"({site_name}): timestamp {ts} is "
                            f"{int(age.total_seconds() / 60)} min old"
                        )
                        continue
                    if not is_number(val):
                        if (fill_strategy or "skip") == "zero":
                            val = 0.0
                        else:
                            continue
                    accumulator[mpid] = accumulator.get(mpid, 0.0) + float(val)
                    if mpid not in latest_ts or reading_time > latest_ts[mpid]:
                        latest_ts[mpid] = reading_time

            rows: List[Dict[str, Any]] = []
            for mpid, val in sorted(accumulator.items()):
                ts_str = fmt_no_T(latest_ts[mpid])
                rows.append(
                    {
                        "start_time_lb_utc": ts_str,
                        "stop_time_lb_utc": ts_str,
                        "variable_value": float(val),
                        "variable_id": mpid,
                        "variable_unit": units.get(mpid, "kW"),
                    }
                )

            if rows:
                frames.append(pd.DataFrame(rows))

        return self._combine_frames(frames, aggregation_method=calculation or "sum")

    @staticmethod
    def _combine_frames(
        frames: List[pd.DataFrame], aggregation_method: str
    ) -> pd.DataFrame:
        """Concatenate per-key frames and aggregate duplicate rows."""
        if not frames:
            return make_empty_df()
        if aggregation_method not in ("sum", "average", "min", "max"):
            raise ValueError(f"Unsupported aggregation_method: {aggregation_method}")
        # The API only accepts average as parameter but pandas expects mean for aggregation
        if aggregation_method == "average":
            aggregation_method = "mean"
        df = (
            pd.concat(frames, ignore_index=True)
            .groupby(
                [
                    "variable_id",
                    "start_time_lb_utc",
                    "stop_time_lb_utc",
                    "variable_unit",
                ],
                as_index=False,
            )
            .agg(variable_value=("variable_value", aggregation_method))
        )
        return df[DF_COLUMNS]

    def list_signals_df_for_asset(self) -> pd.DataFrame:  # noqa: C901
        """List all available data signals for the operator's assets.

        Returns:
            pandas.DataFrame: Signal metadata with columns ``signal_id``,
            ``title``, ``name``, ``unit``. May be empty.
        """
        empty_cols = ["signal_id", "title", "name", "unit"]

        rows: List[Dict[str, Any]] = []
        seen = set()
        for key in self._keys:
            device_ids = sorted(key["device_to_mpid"])
            if not device_ids:
                continue
            signals = self.post_data_endpoint(
                "datasignals.json",
                {"deviceIds": device_ids},
                key["base_url"],
                key["headers"],
            )

            flat: List[Dict[str, Any]] = []
            if isinstance(signals, list):
                flat = signals
            elif isinstance(signals, dict):
                for sig_list in signals.values():
                    if isinstance(sig_list, list):
                        flat.extend(sig_list)

            for s in flat:
                if not isinstance(s, dict):
                    continue
                sid = s.get("id") if s.get("id") is not None else s.get("dataSignalId")
                title = s.get("title")
                name = s.get("name")
                unit = s.get("unit")
                dedup_key = (sid, (title or name or "").strip().lower())
                if dedup_key in seen:
                    continue
                seen.add(dedup_key)
                rows.append(
                    {"signal_id": sid, "title": title, "name": name, "unit": unit}
                )

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

        df = pd.DataFrame(rows)
        sort_cols = [c for c in ["title", "name", "signal_id"] if c in df.columns]
        return df.sort_values(sort_cols).reset_index(drop=True)

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

Initialize a GreenByte 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 device 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.greenbyte_utils.GreenByteApi import GreenByteApi

gb = GreenByteApi("eurus")
gb_single = GreenByteApi("eurus", mpid="P01KAMA00")

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

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

    Args:
        operator: Operator identifier (top-level key in the configuration).
        mpid: Optional asset identifier (MPID). When provided, only the
            device 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.greenbyte_utils.GreenByteApi import GreenByteApi

        gb = GreenByteApi("eurus")
        gb_single = GreenByteApi("eurus", mpid="P01KAMA00")
        ```
    """
    if not logger:
        logger = get_logger(
            team="physical_operations", application_name="GreenByteApi"
        )
    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[int] = [
        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 GreenByteApi for operator '{operator}' with "
        f"{len(self.device_ids)} device id(s) across {len(self._keys)} key(s)"
    )

fetch_data(start_time, stop_time, signal_title='power', resolution_seconds=900, aggregate='device', calculation='sum', fill_strategy='skip')

Fetch fixed-resolution timeseries data (GET /data).

Device ids are combined into a single request per key; per-device results are mapped back to their MPID, and results across keys are concatenated.

Parameters:

Name	Type	Description	Default
start_time	datetime	Inclusive start timestamp (UTC if naive).	required
stop_time	datetime	Exclusive end timestamp.	required
signal_title	str	Signal title to query (default "power").	'power'
resolution_seconds	int	Data resolution in seconds (must exist in :data:RESOLUTION_MAP).	900
aggregate	str	Aggregation mode passed to the API (default "device").	'device'
calculation	str	Aggregation function (e.g., "sum").	'sum'
fill_strategy	str	Handling of non-numeric/null values: - "skip" (default) drops points. - "zero" substitutes 0.0.	'skip'

Returns:

Type	Description
DataFrame	pandas.DataFrame: Normalized timeseries with columns
DataFrame	start_time_lb_utc, stop_time_lb_utc, variable_id,
DataFrame	variable_value, variable_unit. May be empty.

Raises:

Type	Description
ValueError	If the time range is invalid or the resolution is unsupported.

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def fetch_data(  # noqa: C901
    self,
    start_time: datetime,
    stop_time: datetime,
    signal_title: str = "power",
    resolution_seconds: int = 900,
    aggregate: str = "device",
    calculation: str = "sum",
    fill_strategy: str = "skip",
) -> pd.DataFrame:
    """Fetch fixed-resolution timeseries data (``GET /data``).

    Device ids are combined into a single request per key; per-device results
    are mapped back to their MPID, and results across keys are concatenated.

    Args:
        start_time: Inclusive start timestamp (UTC if naive).
        stop_time: Exclusive end timestamp.
        signal_title: Signal title to query (default ``"power"``).
        resolution_seconds: Data resolution in seconds (must exist in :data:`RESOLUTION_MAP`).
        aggregate: Aggregation mode passed to the API (default ``"device"``).
        calculation: Aggregation function (e.g., ``"sum"``).
        fill_strategy: Handling of non-numeric/null values:
            - ``"skip"`` (default) drops points.
            - ``"zero"`` substitutes ``0.0``.

    Returns:
        pandas.DataFrame: Normalized timeseries with columns
        ``start_time_lb_utc``, ``stop_time_lb_utc``, ``variable_id``,
        ``variable_value``, ``variable_unit``. May be empty.

    Raises:
        ValueError: If the time range is invalid or the resolution is unsupported.
    """
    if start_time >= stop_time:
        raise ValueError("start_time must be earlier than stop_time")
    if resolution_seconds not in RESOLUTION_MAP:
        raise ValueError(f"Unsupported resolution_seconds={resolution_seconds}")
    resolution = RESOLUTION_MAP[resolution_seconds]
    ts_start = to_rfc3339_utc(start_time)
    ts_end = to_rfc3339_utc(stop_time)

    frames: List[pd.DataFrame] = []
    for key in self._keys:
        device_to_mpid = key["device_to_mpid"]
        device_ids = sorted(device_to_mpid)
        self._logger.debug(
            f"Fetching data for key '{key['key_secret']}' "
            f"sites {sorted(set(key['site_names'].values()))}"
        )

        data_signal_id = self._pick_signal_id(key, signal_title)
        if data_signal_id is None:
            self._logger.warning(
                f"No '{signal_title}' signal found for key "
                f"'{key['key_secret']}'"
            )
            continue

        params = {
            "deviceIds": ",".join(map(str, device_ids)),
            "dataSignalIds": str(data_signal_id),
            "timestampStart": ts_start,
            "timestampEnd": ts_end,
            "useUtc": "true",
            "resolution": resolution,
            "aggregate": aggregate,
            "aggregateLevel": 0,
            "calculation": calculation,
        }

        try:
            data_items = self.get_data_endpoint(
                "data", key["base_url"], key["headers"], **params
            )
        except Exception:
            continue
        if not isinstance(data_items, list):
            continue

        rows: List[Dict[str, Any]] = []
        for it in data_items:
            if not isinstance(it, dict):
                continue
            device_id = self._extract_device_id(it)
            mpid = device_to_mpid.get(device_id) if device_id is not None else None
            if mpid is None:
                continue
            ds = it.get("dataSignal") or {}
            unit = ds.get("unit") or "kW"
            data_map = it.get("data") or {}
            if not isinstance(data_map, dict):
                continue
            for ts, val in sorted(data_map.items()):
                if not is_number(val):
                    if fill_strategy == "zero":
                        val = 0.0
                    else:
                        continue
                start = pd.to_datetime(ts, utc=True)
                stop = start + pd.to_timedelta(resolution_seconds, unit="s")
                rows.append(
                    {
                        "start_time_lb_utc": fmt_no_T(start),
                        "stop_time_lb_utc": fmt_no_T(stop),
                        "variable_value": float(val),
                        "variable_id": mpid,
                        "variable_unit": unit,
                    }
                )

        if rows:
            frames.append(pd.DataFrame(rows))

    return self._combine_frames(frames, aggregation_method=calculation or "sum")

fetch_realtime_data(signal_title=None, aggregate=None, calculation=None, fill_strategy=None)

Fetch instantaneous realtime points (GET /realtimedata).

Device ids are combined into a single request per key; per-device results are mapped back to their MPID, and results across keys are concatenated.

Parameters:

Name	Type	Description	Default
signal_title	Optional[str]	Signal title to query. Defaults to "power".	None
aggregate	Optional[str]	Aggregation mode passed to the API. Defaults to "device".	None
calculation	Optional[str]	Aggregation function (e.g., "sum").	None
fill_strategy	Optional[str]	Handling of non-numeric/null values: - "skip" (default) drops points. - "zero" substitutes 0.0.	None

Returns:

Type	Description
DataFrame	pandas.DataFrame: Same column contract as :meth:fetch_data; points are
DataFrame	instantaneous (start_time_lb_utc == stop_time_lb_utc). May be empty.

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def fetch_realtime_data(  # noqa: C901
    self,
    signal_title: Optional[str] = None,
    aggregate: Optional[str] = None,
    calculation: Optional[str] = None,
    fill_strategy: Optional[str] = None,
) -> pd.DataFrame:
    """Fetch instantaneous realtime points (``GET /realtimedata``).

    Device ids are combined into a single request per key; per-device results
    are mapped back to their MPID, and results across keys are concatenated.

    Args:
        signal_title: Signal title to query. Defaults to ``"power"``.
        aggregate: Aggregation mode passed to the API. Defaults to ``"device"``.
        calculation: Aggregation function (e.g., ``"sum"``).
        fill_strategy: Handling of non-numeric/null values:
            - ``"skip"`` (default) drops points.
            - ``"zero"`` substitutes ``0.0``.

    Returns:
        pandas.DataFrame: Same column contract as :meth:`fetch_data`; points are
        instantaneous (``start_time_lb_utc == stop_time_lb_utc``). May be empty.
    """
    now_utc = get_utc_now_custom_precision()
    stale_limit = timedelta(minutes=10)

    frames: List[pd.DataFrame] = []
    for key in self._keys:
        device_to_mpid = key["device_to_mpid"]
        site_names = key["site_names"]
        device_ids = sorted(device_to_mpid)
        self._logger.debug(
            f"Fetching realtime data for key '{key['key_secret']}' "
            f"sites {sorted(set(site_names.values()))}"
        )

        data_signal_id = self._pick_signal_id(key, signal_title or "power")
        if data_signal_id is None:
            self._logger.warning(
                f"No '{signal_title or 'power'}' signal found for key "
                f"'{key['key_secret']}'"
            )
            continue

        params = {
            "deviceIds": ",".join(map(str, device_ids)),
            "dataSignalIds": str(data_signal_id),
            "aggregate": aggregate or "device",
            "calculation": calculation or "sum",
        }

        try:
            rt_items = self.get_data_endpoint(
                "realtimedata", key["base_url"], key["headers"], **params
            )
        except Exception:
            continue
        if not isinstance(rt_items, list):
            continue

        # Accumulate per MPID, ignoring sub-minute timestamp differences.
        # latest_ts tracks the most recent valid reading time for each MPID
        # so it can be used as the representative timestamp in the output.
        accumulator: Dict[str, float] = {}  # mpid -> summed value
        units: Dict[str, str] = {}  # mpid -> unit
        latest_ts: Dict[str, pd.Timestamp] = {}  # mpid -> most recent ts

        for item in rt_items:
            if not isinstance(item, dict):
                continue
            device_id = self._extract_device_id(item)
            mpid = device_to_mpid.get(device_id) if device_id is not None else None
            if mpid is None:
                continue
            ds = item.get("dataSignal") or {}
            unit = ds.get("unit") or units.get(mpid) or "kW"
            units[mpid] = unit
            data_map = item.get("data") or {}
            if not isinstance(data_map, dict):
                continue
            for ts, val in data_map.items():
                reading_time = pd.to_datetime(ts, utc=True)
                age = now_utc - reading_time.to_pydatetime()
                if age > stale_limit:
                    site_name = site_names.get(mpid, mpid)
                    self._logger.warning(
                        f"Stale realtime reading skipped — operator "
                        f"'{self.operator}', device {device_id} "
                        f"({site_name}): timestamp {ts} is "
                        f"{int(age.total_seconds() / 60)} min old"
                    )
                    continue
                if not is_number(val):
                    if (fill_strategy or "skip") == "zero":
                        val = 0.0
                    else:
                        continue
                accumulator[mpid] = accumulator.get(mpid, 0.0) + float(val)
                if mpid not in latest_ts or reading_time > latest_ts[mpid]:
                    latest_ts[mpid] = reading_time

        rows: List[Dict[str, Any]] = []
        for mpid, val in sorted(accumulator.items()):
            ts_str = fmt_no_T(latest_ts[mpid])
            rows.append(
                {
                    "start_time_lb_utc": ts_str,
                    "stop_time_lb_utc": ts_str,
                    "variable_value": float(val),
                    "variable_id": mpid,
                    "variable_unit": units.get(mpid, "kW"),
                }
            )

        if rows:
            frames.append(pd.DataFrame(rows))

    return self._combine_frames(frames, aggregation_method=calculation or "sum")

get_data_endpoint(endpoint, base_url, headers, **params)

Perform a GET request against the GreenByte API.

Parameters:

Name	Type	Description	Default
endpoint	str	Path relative to the base API URL (e.g. "data").	required
base_url	str	Base URL of the key the request targets.	required
headers	Dict[str, str]	HTTP headers (including the API key) for that key.	required
**params	Any	Optional query parameters.	{}

Returns:

Name	Type	Description
Any	Any	Parsed JSON response, typically a dict or list.

Raises:

Type	Description
HTTPError	If the request fails (non-2xx).
Timeout	If the request exceeds the 30 s timeout.

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def get_data_endpoint(
    self,
    endpoint: str,
    base_url: str,
    headers: Dict[str, str],
    **params: Any,
) -> Any:
    """Perform a GET request against the GreenByte API.

    Args:
        endpoint: Path relative to the base API URL (e.g. ``"data"``).
        base_url: Base URL of the key the request targets.
        headers: HTTP headers (including the API key) for that key.
        **params: Optional query parameters.

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

    Raises:
        requests.HTTPError: If the request fails (non-2xx).
        requests.Timeout: If the request exceeds the 30 s timeout.
    """
    url = base_url + endpoint
    try:
        r = requests.get(
            url, headers=headers, params=(params or None), timeout=(10, 30)
        )
        r.raise_for_status()
    except requests.HTTPError as exc:
        self._logger.error(f"HTTP error on GET {url}: {exc}")
        raise
    return r.json()

get_device_ids_for_site(site_id, devices=None, subdomain=None)

Return all device ids associated with a site.

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

Parameters:

Name	Type	Description	Default
site_id	int	Site identifier to filter devices by.	required
devices	Optional[List[Dict[str, Any]]]	Optional list of pre-fetched devices (as returned by :meth:get_devices). Fetched automatically when omitted.	None
subdomain	Optional[str]	Optional subdomain to further restrict matches, useful when site ids overlap across an operator's keys.	None

Returns:

Type	Description
List[int]	List[int]: Integer device ids for the site.

Example

from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

gb = GreenByteApi("eurus")
device_ids = gb.get_device_ids_for_site(13)
print(device_ids)

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def get_device_ids_for_site(
    self,
    site_id: int,
    devices: Optional[List[Dict[str, Any]]] = None,
    subdomain: Optional[str] = None,
) -> List[int]:
    """Return all device ids associated with a site.

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

    Args:
        site_id: Site identifier to filter devices by.
        devices: Optional list of pre-fetched devices (as returned by
            :meth:`get_devices`). Fetched automatically when omitted.
        subdomain: Optional subdomain to further restrict matches, useful
            when site ids overlap across an operator's keys.

    Returns:
        List[int]: Integer device ids for the site.

    Example:
        ```python
        from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

        gb = GreenByteApi("eurus")
        device_ids = gb.get_device_ids_for_site(13)
        print(device_ids)
        ```
    """
    if devices is None:
        devices = self.get_devices(page=1, page_size=2000)
    out: List[int] = []
    for d in devices:
        if not isinstance(d, dict):
            continue
        if (d.get("site") or {}).get("siteId") != site_id:
            continue
        if subdomain is not None and d.get("subdomain") != subdomain:
            continue
        did = d.get("deviceId")
        if did is None:
            did = d.get("id")
        if isinstance(did, int):
            out.append(did)
    return out

get_devices(page=1, page_size=2000)

List every device reachable across the operator's keys.

This queries the GreenByte API directly (rather than the configuration) so that device ids can be discovered when onboarding new assets. Each returned dictionary is annotated with the key_secret and subdomain it originates from.

Parameters:

Name	Type	Description	Default
page	int	Page number for pagination.	1
page_size	int	Number of items per page (default 2000).	2000

Returns:

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

Example

from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

gb = GreenByteApi("eurus")
devices = gb.get_devices()
print(len(devices))

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def get_devices(self, page: int = 1, page_size: int = 2000) -> List[Dict[str, Any]]:
    """List every device reachable across the operator's keys.

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

    Args:
        page: Page number for pagination.
        page_size: Number of items per page (default 2000).

    Returns:
        List[Dict[str, Any]]: Device dictionaries (API payload plus
        ``key_secret`` and ``subdomain``). Empty list if none found.

    Example:
        ```python
        from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

        gb = GreenByteApi("eurus")
        devices = gb.get_devices()
        print(len(devices))
        ```
    """
    devices: List[Dict[str, Any]] = []
    for key in self._keys:
        data = self.get_data_endpoint(
            "devices",
            key["base_url"],
            key["headers"],
            page=page,
            pageSize=page_size,
        )
        if not isinstance(data, list):
            continue
        for device in data:
            if isinstance(device, dict):
                devices.append(
                    {
                        **device,
                        "key_secret": key["key_secret"],
                        "subdomain": key["subdomain"],
                    }
                )
    return devices

get_sites(page=1, page_size=2000)

List every site reachable across the operator's keys.

This queries the GreenByte API directly (rather than the configuration) so that site ids can be discovered when onboarding new assets. Each returned dictionary is annotated with the key_secret and subdomain it originates from, which disambiguates sites when an operator spans several key-vault keys/subdomains.

Parameters:

Name	Type	Description	Default
page	int	Page number for pagination.	1
page_size	int	Number of items per page (default 2000).	2000

Returns:

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

Example

from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

gb = GreenByteApi("eurus")
sites = gb.get_sites()
for s in sites:
    print(s["subdomain"], s.get("siteId"), s.get("title"))

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def get_sites(self, page: int = 1, page_size: int = 2000) -> List[Dict[str, Any]]:
    """List every site reachable across the operator's keys.

    This queries the GreenByte API directly (rather than the configuration)
    so that site ids can be discovered when onboarding new assets. Each
    returned dictionary is annotated with the ``key_secret`` and
    ``subdomain`` it originates from, which disambiguates sites when an
    operator spans several key-vault keys/subdomains.

    Args:
        page: Page number for pagination.
        page_size: Number of items per page (default 2000).

    Returns:
        List[Dict[str, Any]]: Site dictionaries (API payload plus
        ``key_secret`` and ``subdomain``). Empty list if none found.

    Example:
        ```python
        from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

        gb = GreenByteApi("eurus")
        sites = gb.get_sites()
        for s in sites:
            print(s["subdomain"], s.get("siteId"), s.get("title"))
        ```
    """
    sites: List[Dict[str, Any]] = []
    for key in self._keys:
        data = self.get_data_endpoint(
            "sites",
            key["base_url"],
            key["headers"],
            page=page,
            pageSize=page_size,
        )
        if not isinstance(data, list):
            continue
        for site in data:
            if isinstance(site, dict):
                sites.append(
                    {
                        **site,
                        "key_secret": key["key_secret"],
                        "subdomain": key["subdomain"],
                    }
                )
    return sites

list_signals_df_for_asset()

List all available data signals for the operator's assets.

Returns:

Type	Description
DataFrame	pandas.DataFrame: Signal metadata with columns signal_id,
DataFrame	title, name, unit. May be empty.

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def list_signals_df_for_asset(self) -> pd.DataFrame:  # noqa: C901
    """List all available data signals for the operator's assets.

    Returns:
        pandas.DataFrame: Signal metadata with columns ``signal_id``,
        ``title``, ``name``, ``unit``. May be empty.
    """
    empty_cols = ["signal_id", "title", "name", "unit"]

    rows: List[Dict[str, Any]] = []
    seen = set()
    for key in self._keys:
        device_ids = sorted(key["device_to_mpid"])
        if not device_ids:
            continue
        signals = self.post_data_endpoint(
            "datasignals.json",
            {"deviceIds": device_ids},
            key["base_url"],
            key["headers"],
        )

        flat: List[Dict[str, Any]] = []
        if isinstance(signals, list):
            flat = signals
        elif isinstance(signals, dict):
            for sig_list in signals.values():
                if isinstance(sig_list, list):
                    flat.extend(sig_list)

        for s in flat:
            if not isinstance(s, dict):
                continue
            sid = s.get("id") if s.get("id") is not None else s.get("dataSignalId")
            title = s.get("title")
            name = s.get("name")
            unit = s.get("unit")
            dedup_key = (sid, (title or name or "").strip().lower())
            if dedup_key in seen:
                continue
            seen.add(dedup_key)
            rows.append(
                {"signal_id": sid, "title": title, "name": name, "unit": unit}
            )

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

    df = pd.DataFrame(rows)
    sort_cols = [c for c in ["title", "name", "signal_id"] if c in df.columns]
    return df.sort_values(sort_cols).reset_index(drop=True)

post_data_endpoint(endpoint, payload, base_url, headers)

Perform a POST request against the GreenByte API.

Parameters:

Name	Type	Description	Default
endpoint	str	Path relative to the base API URL (e.g. "datasignals.json").	required
payload	Dict[str, Any]	JSON-serializable request body.	required
base_url	str	Base URL of the key the request targets.	required
headers	Dict[str, str]	HTTP headers (including the API key) for that key.	required

Returns:

Name	Type	Description
Any	Any	Parsed JSON response, typically a dict or list.

Raises:

Type	Description
HTTPError	If the request fails (non-2xx).
Timeout	If the request exceeds the 30 s timeout.

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def post_data_endpoint(
    self,
    endpoint: str,
    payload: Dict[str, Any],
    base_url: str,
    headers: Dict[str, str],
) -> Any:
    """Perform a POST request against the GreenByte API.

    Args:
        endpoint: Path relative to the base API URL (e.g. ``"datasignals.json"``).
        payload: JSON-serializable request body.
        base_url: Base URL of the key the request targets.
        headers: HTTP headers (including the API key) for that key.

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

    Raises:
        requests.HTTPError: If the request fails (non-2xx).
        requests.Timeout: If the request exceeds the 30 s timeout.
    """
    url = base_url + endpoint
    try:
        r = requests.post(url, headers=headers, json=payload, timeout=30)
        r.raise_for_status()
    except requests.HTTPError as exc:
        self._logger.error(f"HTTP error on POST {url}: {exc}")
        raise
    return r.json()