greenbyte_api

This is a wrapper to communicate with the GreenByte API.

GreenByteApi

Client for the GreenByte Cloud API bound to a specific asset.

The client provides high-level methods for:

• Listing sites and devices.
• Discovering available data signals.
• Fetching fixed-resolution timeseries.
• Fetching realtime instantaneous values.

Attributes:

Name	Type	Description
asset_name		Configured asset identifier (key in greenbyte_config.config).
base_url		Base URL for the API, including subdomain.
cfg		Asset configuration dictionary.
headers		HTTP headers containing the API key and metadata.

Notes

Each instance is bound to a single asset configuration. 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 a specific asset.

    The client provides high-level methods for:

    - Listing sites and devices.
    - Discovering available data signals.
    - Fetching fixed-resolution timeseries.
    - Fetching realtime instantaneous values.

    Attributes:
        asset_name: Configured asset identifier (key in ``greenbyte_config.config``).
        base_url: Base URL for the API, including subdomain.
        cfg: Asset configuration dictionary.
        headers: HTTP headers containing the API key and metadata.

    Notes:
        Each instance is bound to a single asset configuration.
        All time-series methods return :class:`pandas.DataFrame` objects.
    """

    def __init__(self, asset_name: str):
        """Initialize a GreenByte API client for a specific asset.

        Args:
            asset_name: Asset identifier as defined in ``greenbyte_config.config``.

        Raises:
            ValueError: If the asset configuration is invalid or incomplete.

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

            gb = GreenByteApi("militorp")
            ```
        """
        self.asset_name = asset_name
        self.cfg = config.get(asset_name)
        if not isinstance(self.cfg, dict):
            raise ValueError(f"Unknown asset_name: {asset_name}")

        subdomain = self.cfg.get("subdomain") or ""
        if not subdomain:
            raise ValueError(f"Missing subdomain for asset: {asset_name}")

        api_key_secret = self.cfg.get("api_key_secret") or ""
        if not isinstance(api_key_secret, str):
            raise ValueError(f"api_key_secret must be a string for asset: {asset_name}")

        api_key = get_secret(api_key_secret)
        if not isinstance(api_key, str):
            raise ValueError(f"api_key must be a string for asset: {asset_name}")

        self.base_url = f"https://{subdomain}.greenbyte.cloud/api/2.2/"
        self.headers = {
            "X-Api-Key": api_key,
            "Accept": "application/json",
            "User-Agent": "gb-multi-park/0.1",
        }

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

        Args:
            endpoint: Path relative to the base API URL (e.g. ``"sites"``, ``"data"``).
            **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.

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

            gb = GreenByteApi("militorp")
            sites = gb.get_data_endpoint("sites", page=1, pageSize=100)
            ```
        """
        url = self.base_url + endpoint
        r = requests.get(url, headers=self.headers, params=(params or None), timeout=30)
        r.raise_for_status()
        return r.json()

    def post_data_endpoint(self, endpoint: str, payload: Dict[str, Any]) -> 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.

        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.
            ValueError: If the payload cannot be serialized.

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

            gb = GreenByteApi("militorp")
            signals = gb.post_data_endpoint("datasignals.json", {"deviceIds": [1, 2, 3]})
            ```
        """
        url = self.base_url + endpoint
        r = requests.post(url, headers=self.headers, json=payload, timeout=30)
        r.raise_for_status()
        return r.json()

    # ---------------- Public helpers ----------------
    def get_sites(self, page: int = 1, pageSize: int = 2000) -> List[Dict[str, Any]]:
        """List available sites for the configured asset.

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

        Returns:
            List[Dict[str, Any]]: Site dictionaries. Empty list if none found.

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

            gb = GreenByteApi("militorp")
            sites = gb.get_sites()
            print(len(sites))
            ```
        """
        data = self.get_data_endpoint("sites", page=page, pageSize=pageSize)
        return data if isinstance(data, list) else []

    def get_devices(self, page: int = 1, pageSize: int = 2000) -> List[Dict[str, Any]]:
        """List devices accessible to the configured asset.

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

        Returns:
            List[Dict[str, Any]]: Device dictionaries. Empty list if none found.

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

            gb = GreenByteApi("militorp")
            devices = gb.get_devices()
            print(len(devices))
            ```
        """
        data = self.get_data_endpoint("devices", page=page, pageSize=pageSize)
        return data if isinstance(data, list) else []

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

        Args:
            site_id: Site identifier to filter devices by.
            devices: Optional list of pre-fetched devices.

        Returns:
            List[int]: Integer device IDs.

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

            gb = GreenByteApi("militorp")
            device_ids = gb.get_device_ids_for_site(123)
            print(device_ids)
            ```
        """
        if devices is None:
            devices = self.get_devices(page=1, pageSize=2000)
        out = []
        for d in devices:
            if isinstance(d, dict) and (d.get("site") or {}).get("siteId") == site_id:
                did = d.get("deviceId") or 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.datetime,
        stop_time: datetime.datetime,
        signal_title: str = "power",
        resolution_seconds: int = 900,
        aggregate: str = "site",
        calculation: str = "sum",
        mpids: Optional[List[str]] = None,
        fill_strategy: str = "skip",
    ) -> pd.DataFrame:
        """Fetch fixed-resolution timeseries data (``GET /data``).

        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 (``"site"`` or ``"device"``).
            calculation: Aggregation function (e.g., ``"sum"``).
            mpids: Optional list of MPIDs to query. Defaults to configured MPIDs.
            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)

        sites = self.get_sites(page=1, pageSize=2000)
        devices = self.get_devices(page=1, pageSize=2000)
        if not sites or not devices:
            return make_empty_df()
        site_id = sites[0].get("siteId")
        all_device_ids = self.get_device_ids_for_site(site_id, devices=devices)
        if not all_device_ids:
            return make_empty_df()

        signals = self.post_data_endpoint(
            "datasignals.json", {"deviceIds": [all_device_ids[0]]}
        )
        data_signal_id = pick_signal_id_by_title(
            signals, all_device_ids[0], signal_title=signal_title
        )
        if not isinstance(data_signal_id, int):
            return make_empty_df()

        mpids = mpids or (self.cfg.get("mpids") or [])
        wtg_map = self.cfg.get("wtg_number_mpid_map") or {}

        rows: List[Dict[str, Any]] = []
        for mpid in mpids:
            if "all" in wtg_map:
                ids_for_mpid = all_device_ids[:]
            else:
                ids_for_mpid = []
                for k, v in wtg_map.items():
                    if v == mpid:
                        try:
                            ids_for_mpid.append(int(k))
                        except Exception:
                            pass
            if not ids_for_mpid:
                continue

            params = {
                "deviceIds": ",".join(map(str, ids_for_mpid)),
                "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", **params)
            except Exception:
                continue
            if not isinstance(data_items, list):
                continue

            unit = "kW"
            for it in data_items:
                ds = it.get("dataSignal") or {}
                unit = ds.get("unit") or unit
                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 not rows:
            return make_empty_df()

        df = (
            pd.DataFrame(rows)
            .groupby(
                [
                    "variable_id",
                    "start_time_lb_utc",
                    "stop_time_lb_utc",
                    "variable_unit",
                ],
                as_index=False,
            )
            .agg(variable_value=("variable_value", "sum"))
        )
        return df[DF_COLUMNS]

    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``).

        Args:
            signal_title: Signal title to query. Defaults to ``"power"``.
            aggregate: Aggregation mode (``"site"`` or ``"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.
        """
        sites = self.get_sites(page=1, pageSize=2000)
        devices = self.get_devices(page=1, pageSize=2000)
        if not sites or not devices:
            return make_empty_df()
        site_id = sites[0].get("siteId")
        all_device_ids = self.get_device_ids_for_site(site_id, devices=devices)
        if not all_device_ids:
            return make_empty_df()

        signals = self.post_data_endpoint(
            "datasignals.json", {"deviceIds": [all_device_ids[0]]}
        )
        data_signal_id = pick_signal_id_by_title(
            signals, all_device_ids[0], signal_title=signal_title or "power"
        )
        if not isinstance(data_signal_id, int):
            return make_empty_df()

        mpids = self.cfg.get("mpids") or []
        wtg_map = self.cfg.get("wtg_number_mpid_map") or {}

        rows: List[Dict[str, Any]] = []
        for mpid in mpids:
            if "all" in wtg_map:
                ids_for_mpid = all_device_ids[:]
            else:
                ids_for_mpid = []
                for k, v in wtg_map.items():
                    if v == mpid:
                        try:
                            ids_for_mpid.append(int(k))
                        except Exception:
                            pass
            if not ids_for_mpid:
                continue

            params = {
                "deviceIds": ",".join(map(str, ids_for_mpid)),
                "dataSignalIds": str(data_signal_id),
                "aggregate": aggregate,
                "calculation": calculation,
            }

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

            unit = "kW"
            ts_values: Dict[str, float] = {}
            for item in rt_items:
                ds = item.get("dataSignal") or {}
                unit = ds.get("unit") or unit
                data_map = item.get("data") or {}
                if not isinstance(data_map, dict):
                    continue
                for ts, val in data_map.items():
                    if not is_number(val):
                        if fill_strategy == "zero":
                            val = 0.0
                        else:
                            continue
                    ts_values[ts] = ts_values.get(ts, 0.0) + float(val)

            for ts, val in sorted(ts_values.items()):
                t = pd.to_datetime(ts, utc=True)
                rows.append(
                    {
                        "start_time_lb_utc": fmt_no_T(t),
                        "stop_time_lb_utc": fmt_no_T(t),
                        "variable_value": float(val),
                        "variable_id": mpid,
                        "variable_unit": unit,
                    }
                )

        if not rows:
            return make_empty_df()

        df = (
            pd.DataFrame(rows)
            .groupby(
                [
                    "variable_id",
                    "start_time_lb_utc",
                    "stop_time_lb_utc",
                    "variable_unit",
                ],
                as_index=False,
            )
            .agg(variable_value=("variable_value", "sum"))
        )
        return df[DF_COLUMNS]

    def list_signals_df_for_asset(self) -> pd.DataFrame:  # noqa: C901
        """List all available data signals for the configured asset.

        Returns:
            pandas.DataFrame: Signal metadata with columns ``signal_id``,
            ``title``, ``name``, ``unit``. May be empty.
        """
        sites = self.get_sites(page=1, pageSize=2000)
        devices = self.get_devices(page=1, pageSize=2000)
        if not sites or not devices:
            return pd.DataFrame(columns=["signal_id", "title", "name", "unit"])
        site_id = sites[0].get("siteId")
        device_ids = self.get_device_ids_for_site(site_id, devices=devices)
        if not device_ids:
            return pd.DataFrame(columns=["signal_id", "title", "name", "unit"])

        signals = self.post_data_endpoint("datasignals.json", {"deviceIds": device_ids})

        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)

        if not flat:
            return pd.DataFrame(columns=["signal_id", "title", "name", "unit"])

        rows: List[Dict[str, Any]] = []
        seen = set()
        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")
            key = (sid, (title or name or "").strip().lower())
            if key in seen:
                continue
            seen.add(key)
            rows.append({"signal_id": sid, "title": title, "name": name, "unit": unit})

        if not rows:
            return pd.DataFrame(columns=["signal_id", "title", "name", "unit"])

        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__(asset_name)

Initialize a GreenByte API client for a specific asset.

Parameters:

Name	Type	Description	Default
asset_name	str	Asset identifier as defined in greenbyte_config.config.	required

Raises:

Type	Description
ValueError	If the asset configuration is invalid or incomplete.

Example

from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

gb = GreenByteApi("militorp")

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def __init__(self, asset_name: str):
    """Initialize a GreenByte API client for a specific asset.

    Args:
        asset_name: Asset identifier as defined in ``greenbyte_config.config``.

    Raises:
        ValueError: If the asset configuration is invalid or incomplete.

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

        gb = GreenByteApi("militorp")
        ```
    """
    self.asset_name = asset_name
    self.cfg = config.get(asset_name)
    if not isinstance(self.cfg, dict):
        raise ValueError(f"Unknown asset_name: {asset_name}")

    subdomain = self.cfg.get("subdomain") or ""
    if not subdomain:
        raise ValueError(f"Missing subdomain for asset: {asset_name}")

    api_key_secret = self.cfg.get("api_key_secret") or ""
    if not isinstance(api_key_secret, str):
        raise ValueError(f"api_key_secret must be a string for asset: {asset_name}")

    api_key = get_secret(api_key_secret)
    if not isinstance(api_key, str):
        raise ValueError(f"api_key must be a string for asset: {asset_name}")

    self.base_url = f"https://{subdomain}.greenbyte.cloud/api/2.2/"
    self.headers = {
        "X-Api-Key": api_key,
        "Accept": "application/json",
        "User-Agent": "gb-multi-park/0.1",
    }

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

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

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 ("site" or "device").	'site'
calculation	str	Aggregation function (e.g., "sum").	'sum'
mpids	Optional[List[str]]	Optional list of MPIDs to query. Defaults to configured MPIDs.	None
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.datetime,
    stop_time: datetime.datetime,
    signal_title: str = "power",
    resolution_seconds: int = 900,
    aggregate: str = "site",
    calculation: str = "sum",
    mpids: Optional[List[str]] = None,
    fill_strategy: str = "skip",
) -> pd.DataFrame:
    """Fetch fixed-resolution timeseries data (``GET /data``).

    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 (``"site"`` or ``"device"``).
        calculation: Aggregation function (e.g., ``"sum"``).
        mpids: Optional list of MPIDs to query. Defaults to configured MPIDs.
        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)

    sites = self.get_sites(page=1, pageSize=2000)
    devices = self.get_devices(page=1, pageSize=2000)
    if not sites or not devices:
        return make_empty_df()
    site_id = sites[0].get("siteId")
    all_device_ids = self.get_device_ids_for_site(site_id, devices=devices)
    if not all_device_ids:
        return make_empty_df()

    signals = self.post_data_endpoint(
        "datasignals.json", {"deviceIds": [all_device_ids[0]]}
    )
    data_signal_id = pick_signal_id_by_title(
        signals, all_device_ids[0], signal_title=signal_title
    )
    if not isinstance(data_signal_id, int):
        return make_empty_df()

    mpids = mpids or (self.cfg.get("mpids") or [])
    wtg_map = self.cfg.get("wtg_number_mpid_map") or {}

    rows: List[Dict[str, Any]] = []
    for mpid in mpids:
        if "all" in wtg_map:
            ids_for_mpid = all_device_ids[:]
        else:
            ids_for_mpid = []
            for k, v in wtg_map.items():
                if v == mpid:
                    try:
                        ids_for_mpid.append(int(k))
                    except Exception:
                        pass
        if not ids_for_mpid:
            continue

        params = {
            "deviceIds": ",".join(map(str, ids_for_mpid)),
            "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", **params)
        except Exception:
            continue
        if not isinstance(data_items, list):
            continue

        unit = "kW"
        for it in data_items:
            ds = it.get("dataSignal") or {}
            unit = ds.get("unit") or unit
            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 not rows:
        return make_empty_df()

    df = (
        pd.DataFrame(rows)
        .groupby(
            [
                "variable_id",
                "start_time_lb_utc",
                "stop_time_lb_utc",
                "variable_unit",
            ],
            as_index=False,
        )
        .agg(variable_value=("variable_value", "sum"))
    )
    return df[DF_COLUMNS]

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

Fetch instantaneous realtime points (GET /realtimedata).

Parameters:

Name	Type	Description	Default
signal_title	Optional[str]	Signal title to query. Defaults to "power".	None
aggregate	Optional[str]	Aggregation mode ("site" or "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``).

    Args:
        signal_title: Signal title to query. Defaults to ``"power"``.
        aggregate: Aggregation mode (``"site"`` or ``"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.
    """
    sites = self.get_sites(page=1, pageSize=2000)
    devices = self.get_devices(page=1, pageSize=2000)
    if not sites or not devices:
        return make_empty_df()
    site_id = sites[0].get("siteId")
    all_device_ids = self.get_device_ids_for_site(site_id, devices=devices)
    if not all_device_ids:
        return make_empty_df()

    signals = self.post_data_endpoint(
        "datasignals.json", {"deviceIds": [all_device_ids[0]]}
    )
    data_signal_id = pick_signal_id_by_title(
        signals, all_device_ids[0], signal_title=signal_title or "power"
    )
    if not isinstance(data_signal_id, int):
        return make_empty_df()

    mpids = self.cfg.get("mpids") or []
    wtg_map = self.cfg.get("wtg_number_mpid_map") or {}

    rows: List[Dict[str, Any]] = []
    for mpid in mpids:
        if "all" in wtg_map:
            ids_for_mpid = all_device_ids[:]
        else:
            ids_for_mpid = []
            for k, v in wtg_map.items():
                if v == mpid:
                    try:
                        ids_for_mpid.append(int(k))
                    except Exception:
                        pass
        if not ids_for_mpid:
            continue

        params = {
            "deviceIds": ",".join(map(str, ids_for_mpid)),
            "dataSignalIds": str(data_signal_id),
            "aggregate": aggregate,
            "calculation": calculation,
        }

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

        unit = "kW"
        ts_values: Dict[str, float] = {}
        for item in rt_items:
            ds = item.get("dataSignal") or {}
            unit = ds.get("unit") or unit
            data_map = item.get("data") or {}
            if not isinstance(data_map, dict):
                continue
            for ts, val in data_map.items():
                if not is_number(val):
                    if fill_strategy == "zero":
                        val = 0.0
                    else:
                        continue
                ts_values[ts] = ts_values.get(ts, 0.0) + float(val)

        for ts, val in sorted(ts_values.items()):
            t = pd.to_datetime(ts, utc=True)
            rows.append(
                {
                    "start_time_lb_utc": fmt_no_T(t),
                    "stop_time_lb_utc": fmt_no_T(t),
                    "variable_value": float(val),
                    "variable_id": mpid,
                    "variable_unit": unit,
                }
            )

    if not rows:
        return make_empty_df()

    df = (
        pd.DataFrame(rows)
        .groupby(
            [
                "variable_id",
                "start_time_lb_utc",
                "stop_time_lb_utc",
                "variable_unit",
            ],
            as_index=False,
        )
        .agg(variable_value=("variable_value", "sum"))
    )
    return df[DF_COLUMNS]

get_data_endpoint(endpoint, **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. "sites", "data").	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.

Example

from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

gb = GreenByteApi("militorp")
sites = gb.get_data_endpoint("sites", page=1, pageSize=100)

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

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

    Args:
        endpoint: Path relative to the base API URL (e.g. ``"sites"``, ``"data"``).
        **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.

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

        gb = GreenByteApi("militorp")
        sites = gb.get_data_endpoint("sites", page=1, pageSize=100)
        ```
    """
    url = self.base_url + endpoint
    r = requests.get(url, headers=self.headers, params=(params or None), timeout=30)
    r.raise_for_status()
    return r.json()

get_device_ids_for_site(site_id, devices=None)

Return all device IDs associated with a site.

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.	None

Returns:

Type	Description
List[int]	List[int]: Integer device IDs.

Example

from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

gb = GreenByteApi("militorp")
device_ids = gb.get_device_ids_for_site(123)
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
) -> List[int]:
    """Return all device IDs associated with a site.

    Args:
        site_id: Site identifier to filter devices by.
        devices: Optional list of pre-fetched devices.

    Returns:
        List[int]: Integer device IDs.

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

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

get_devices(page=1, pageSize=2000)

List devices accessible to the configured asset.

Parameters:

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

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: Device dictionaries. Empty list if none found.

Example

from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

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

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def get_devices(self, page: int = 1, pageSize: int = 2000) -> List[Dict[str, Any]]:
    """List devices accessible to the configured asset.

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

    Returns:
        List[Dict[str, Any]]: Device dictionaries. Empty list if none found.

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

        gb = GreenByteApi("militorp")
        devices = gb.get_devices()
        print(len(devices))
        ```
    """
    data = self.get_data_endpoint("devices", page=page, pageSize=pageSize)
    return data if isinstance(data, list) else []

get_sites(page=1, pageSize=2000)

List available sites for the configured asset.

Parameters:

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

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: Site dictionaries. Empty list if none found.

Example

from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

gb = GreenByteApi("militorp")
sites = gb.get_sites()
print(len(sites))

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def get_sites(self, page: int = 1, pageSize: int = 2000) -> List[Dict[str, Any]]:
    """List available sites for the configured asset.

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

    Returns:
        List[Dict[str, Any]]: Site dictionaries. Empty list if none found.

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

        gb = GreenByteApi("militorp")
        sites = gb.get_sites()
        print(len(sites))
        ```
    """
    data = self.get_data_endpoint("sites", page=page, pageSize=pageSize)
    return data if isinstance(data, list) else []

list_signals_df_for_asset()

List all available data signals for the configured asset.

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 configured asset.

    Returns:
        pandas.DataFrame: Signal metadata with columns ``signal_id``,
        ``title``, ``name``, ``unit``. May be empty.
    """
    sites = self.get_sites(page=1, pageSize=2000)
    devices = self.get_devices(page=1, pageSize=2000)
    if not sites or not devices:
        return pd.DataFrame(columns=["signal_id", "title", "name", "unit"])
    site_id = sites[0].get("siteId")
    device_ids = self.get_device_ids_for_site(site_id, devices=devices)
    if not device_ids:
        return pd.DataFrame(columns=["signal_id", "title", "name", "unit"])

    signals = self.post_data_endpoint("datasignals.json", {"deviceIds": device_ids})

    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)

    if not flat:
        return pd.DataFrame(columns=["signal_id", "title", "name", "unit"])

    rows: List[Dict[str, Any]] = []
    seen = set()
    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")
        key = (sid, (title or name or "").strip().lower())
        if key in seen:
            continue
        seen.add(key)
        rows.append({"signal_id": sid, "title": title, "name": name, "unit": unit})

    if not rows:
        return pd.DataFrame(columns=["signal_id", "title", "name", "unit"])

    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)

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

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.
ValueError	If the payload cannot be serialized.

Example

from physical_operations_utils.greenbyte_utils.GreenByteApi import GreenByteApi

gb = GreenByteApi("militorp")
signals = gb.post_data_endpoint("datasignals.json", {"deviceIds": [1, 2, 3]})

Source code in physical_operations_utils/greenbyte_utils/GreenByteApi.py

def post_data_endpoint(self, endpoint: str, payload: Dict[str, Any]) -> 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.

    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.
        ValueError: If the payload cannot be serialized.

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

        gb = GreenByteApi("militorp")
        signals = gb.post_data_endpoint("datasignals.json", {"deviceIds": [1, 2, 3]})
        ```
    """
    url = self.base_url + endpoint
    r = requests.post(url, headers=self.headers, json=payload, timeout=30)
    r.raise_for_status()
    return r.json()