timeseries_shaper.loader.metadata.metadata_api_loader

  1import requests
  2import pandas as pd
  3import json
  4from typing import List, Dict
  5
  6
  7class DatapointAPI:
  8    """
  9    Class for accessing datapoints for multiple devices via an API.
 10    """
 11
 12    def __init__(self, device_names: List[str], base_url: str, api_token: str, output_path: str = "data", required_uuid_list: List[str] = None, filter_enabled: bool = True):
 13        """
 14        Initialize the DatapointAPI class.
 15
 16        :param device_names: List of device names to retrieve metadata for.
 17        :param base_url: Base URL of the API.
 18        :param api_token: API token for authentication.
 19        :param output_path: Directory to save the data points JSON files.
 20        :param required_uuid_list: Mixed list of UUIDs to filter the metadata across devices (optional).
 21        :param filter_enabled: Whether to filter metadata by "enabled == True" (default is True).
 22        """
 23        self.device_names = device_names
 24        self.base_url = base_url
 25        self.api_token = api_token
 26        self.output_path = output_path
 27        self.required_uuid_list = required_uuid_list or []  # Defaults to an empty list if None
 28        self.filter_enabled = filter_enabled
 29        self.device_metadata: Dict[str, pd.DataFrame] = {}  # Store metadata for each device
 30        self.device_uuids: Dict[str, List[str]] = {}  # Store UUIDs for each device
 31        self._api_access()
 32
 33    def _api_access(self) -> None:
 34        """Connect to the API and retrieve metadata for the specified devices."""
 35        headers = {
 36            "Content-Type": "application/json",
 37            "Authorization": f"Bearer {self.api_token}",
 38        }
 39
 40        for device_name in self.device_names:
 41            metadata = []
 42            devices_found = []
 43
 44            for datatron in requests.get(f"{self.base_url}", headers=headers).json():
 45                for device in requests.get(f"{self.base_url}/{datatron['id']}/devices", headers=headers).json():
 46                    if device["name"] == device_name:
 47                        datapoints = requests.get(
 48                            f"{self.base_url}/{datatron['id']}/devices/{device['id']}/data_points",
 49                            headers=headers,
 50                        ).json()
 51                        metadata += datapoints
 52                        devices_found.append(device["name"])
 53                    if devices_found:
 54                        break
 55                if devices_found:
 56                    break
 57
 58            # Process metadata for the current device
 59            metadata_df = pd.DataFrame(metadata)
 60            if not metadata_df.empty:
 61                if self.filter_enabled:
 62                    metadata_df = metadata_df[metadata_df["enabled"] == True]
 63
 64                metadata_df = metadata_df[["uuid", "label", "config"]]
 65
 66                # Filter metadata by required UUIDs, if any
 67                if self.required_uuid_list:
 68                    metadata_df = metadata_df[metadata_df["uuid"].isin(self.required_uuid_list)]
 69
 70                # Store processed metadata and UUIDs
 71                self.device_metadata[device_name] = metadata_df
 72                self.device_uuids[device_name] = metadata_df["uuid"].tolist()
 73
 74                # Export JSON file for this device
 75                self._export_json(metadata_df.to_dict(orient="records"), device_name)
 76
 77    def _export_json(self, data_points: List[Dict[str, str]], device_name: str) -> None:
 78        """Export data points to a JSON file for the specified device."""
 79        file_name = f"{self.output_path}/{device_name.replace(' ', '_')}_data_points.json"
 80        with open(file_name, 'w') as f:
 81            json.dump(data_points, f, indent=2)
 82
 83    def get_all_uuids(self) -> Dict[str, List[str]]:
 84        """Return a dictionary of UUIDs for each device."""
 85        return self.device_uuids
 86
 87    def get_all_metadata(self) -> Dict[str, List[Dict[str, str]]]:
 88        """Return a dictionary of metadata for each device."""
 89        return {device: metadata.to_dict(orient="records") for device, metadata in self.device_metadata.items()}
 90
 91    def display_dataframe(self, device_name: str = None) -> None:
 92        """
 93        Print the metadata DataFrame for a specific device or all devices.
 94
 95        :param device_name: Name of the device to display metadata for (optional).
 96                            If None, displays metadata for all devices.
 97        """
 98        if device_name:
 99            # Display metadata for a specific device
100            if device_name in self.device_metadata:
101                print(f"Metadata for device: {device_name}")
102                print(self.device_metadata[device_name])
103            else:
104                print(f"No metadata found for device: {device_name}")
105        else:
106            # Display metadata for all devices
107            for device, metadata in self.device_metadata.items():
108                print(f"\nMetadata for device: {device}")
109                print(metadata)