add: metadata loader improved

Author: jakobgabriel

README.md

@@ -92,14 +92,4 @@ Please ensure to update tests as appropriate.
 
 ## License
 
-Distributed under the MIT License. See LICENSE for more information.
-
-
-## Development 
-
-- Generate new pdocs: `.\generate_docs.sh`
-- Install package locally: `pip install -e .`
-- Run tests locally with pytest: `pytest ./tests`
-
-- Build package for upload: `python setup.py sdist bdist_wheel`
-- Upload build package to pypi: `twine upload dist/* --verbose --skip-existing`
+Distributed under the MIT License. See LICENSE for more information.

dev_readme.md

@@ -0,0 +1,8 @@
+## Development 
+
+- Generate new pdocs: `.\generate_docs.sh`
+- Install package locally: `pip install -e .`
+- Run tests locally with pytest: `pytest ./tests`
+
+- Build package for upload: `python setup.py sdist bdist_wheel`
+- Upload build package to pypi: `twine upload dist/* --verbose --skip-existing`

requirements.txt

@@ -1,2 +1,3 @@
 pandas==2.1.0
-pdoc==14.6.1 # for docs
+pdoc==14.6.1 # for docs
+scipy==1.14.1

setup.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name = "timeseries-shaper",
-    version = "0.0.0.15",
+    version = "0.0.0.17",
     author = "Jakob Gabriel",
     author_email = "[email protected]",
     description = "timeseries-shaper filters, transforms and engineer your timeseries dataframe",

src/timeseries_shaper/loader/metadata/metadata_api_loader.py

@@ -6,66 +6,104 @@
 
 class DatapointAPI:
     """
-    Class for accessing datapoints via an API.
+    Class for accessing datapoints for multiple devices via an API.
     """
 
-    def __init__(self, device_name: str, base_url: str, api_token: str, output_path: str = "data_points.json"):
-        self.device_name = device_name
+    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):
+        """
+        Initialize the DatapointAPI class.
+
+        :param device_names: List of device names to retrieve metadata for.
+        :param base_url: Base URL of the API.
+        :param api_token: API token for authentication.
+        :param output_path: Directory to save the data points JSON files.
+        :param required_uuid_list: Mixed list of UUIDs to filter the metadata across devices (optional).
+        :param filter_enabled: Whether to filter metadata by "enabled == True" (default is True).
+        """
+        self.device_names = device_names
         self.base_url = base_url
         self.api_token = api_token
         self.output_path = output_path
-        self.uuids: List[str] = []
-        self.metadata: pd.DataFrame = pd.DataFrame([])
+        self.required_uuid_list = required_uuid_list or []  # Defaults to an empty list if None
+        self.filter_enabled = filter_enabled
+        self.device_metadata: Dict[str, pd.DataFrame] = {}  # Store metadata for each device
+        self.device_uuids: Dict[str, List[str]] = {}  # Store UUIDs for each device
         self._api_access()
 
     def _api_access(self) -> None:
-        """Connect to the API and retrieve metadata for the specified device."""
+        """Connect to the API and retrieve metadata for the specified devices."""
         headers = {
             "Content-Type": "application/json",
             "Authorization": f"Bearer {self.api_token}",
         }
 
-        metadata = []
-        devices_found = []
-
-        for datatron in requests.get(f"{self.base_url}", headers=headers).json():
-            for device in requests.get(f"{self.base_url}/{datatron['id']}/devices", headers=headers).json():
-                if device["name"] == self.device_name:
-                    datapoints = requests.get(
-                        f"{self.base_url}/{datatron['id']}/devices/{device['id']}/data_points",
-                        headers=headers,
-                    ).json()
-                    metadata += datapoints
-                    devices_found.append(device["name"])
+        for device_name in self.device_names:
+            metadata = []
+            devices_found = []
+
+            for datatron in requests.get(f"{self.base_url}", headers=headers).json():
+                for device in requests.get(f"{self.base_url}/{datatron['id']}/devices", headers=headers).json():
+                    if device["name"] == device_name:
+                        datapoints = requests.get(
+                            f"{self.base_url}/{datatron['id']}/devices/{device['id']}/data_points",
+                            headers=headers,
+                        ).json()
+                        metadata += datapoints
+                        devices_found.append(device["name"])
+                    if devices_found:
+                        break
                 if devices_found:
                     break
-            if devices_found:
-                break
-
-        self.metadata = pd.DataFrame(metadata)
-        if not self.metadata.empty:
-            self.metadata = self.metadata[self.metadata["enabled"] == True][["uuid", "label", "config"]]
-            data_points = self.metadata.to_dict(orient="records")
-            self._export_json(data_points)
-            self.uuids = [data["uuid"] for data in data_points]
-
-    def _export_json(self, data_points: List[Dict[str, str]]) -> None:
-        """Export data points to a JSON file."""
-        with open(self.output_path, 'w') as f:
+
+            # Process metadata for the current device
+            metadata_df = pd.DataFrame(metadata)
+            if not metadata_df.empty:
+                if self.filter_enabled:
+                    metadata_df = metadata_df[metadata_df["enabled"] == True]
+
+                metadata_df = metadata_df[["uuid", "label", "config"]]
+
+                # Filter metadata by required UUIDs, if any
+                if self.required_uuid_list:
+                    metadata_df = metadata_df[metadata_df["uuid"].isin(self.required_uuid_list)]
+
+                # Store processed metadata and UUIDs
+                self.device_metadata[device_name] = metadata_df
+                self.device_uuids[device_name] = metadata_df["uuid"].tolist()
+
+                # Export JSON file for this device
+                self._export_json(metadata_df.to_dict(orient="records"), device_name)
+
+    def _export_json(self, data_points: List[Dict[str, str]], device_name: str) -> None:
+        """Export data points to a JSON file for the specified device."""
+        file_name = f"{self.output_path}/{device_name.replace(' ', '_')}_data_points.json"
+        with open(file_name, 'w') as f:
             json.dump(data_points, f, indent=2)
 
-    def get_uuids(self) -> List[str]:
-        """Return the list of UUIDs."""
-        return self.uuids
+    def get_all_uuids(self) -> Dict[str, List[str]]:
+        """Return a dictionary of UUIDs for each device."""
+        return self.device_uuids
 
-    def get_full_config(self) -> List[Dict[str, str]]:
-        """Return the full configuration (uuid, label, config) as a list of dictionaries."""
-        return self.metadata.to_dict(orient="records")
+    def get_all_metadata(self) -> Dict[str, List[Dict[str, str]]]:
+        """Return a dictionary of metadata for each device."""
+        return {device: metadata.to_dict(orient="records") for device, metadata in self.device_metadata.items()}
 
-    def get_uuid_label_pairs(self) -> List[Dict[str, str]]:
-        """Return a list of uuid and label pairs."""
-        return self.metadata[['uuid', 'label']].to_dict(orient='records')
+    def display_dataframe(self, device_name: str = None) -> None:
+        """
+        Print the metadata DataFrame for a specific device or all devices.
 
-    def display_dataframe(self) -> None:
-        """Print the metadata DataFrame to visually inspect data points."""
-        print(self.metadata)
+        :param device_name: Name of the device to display metadata for (optional).
+                            If None, displays metadata for all devices.
+        """
+        if device_name:
+            # Display metadata for a specific device
+            if device_name in self.device_metadata:
+                print(f"Metadata for device: {device_name}")
+                print(self.device_metadata[device_name])
+            else:
+                print(f"No metadata found for device: {device_name}")
+        else:
+            # Display metadata for all devices
+            for device, metadata in self.device_metadata.items():
+                print(f"\nMetadata for device: {device}")
+                print(metadata)

src/timeseries_shaper/loader/metadata/metadata_db_loader.py

@@ -9,18 +9,31 @@ class DatapointDB:
     Class for accessing datapoints via a database.
     """
 
-    def __init__(self, device_name: str, db_user: str, db_pass: str, db_host: str, output_path: str = "data_points.json"):
-        self.device_name = device_name
+    def __init__(self, device_names: List[str], db_user: str, db_pass: str, db_host: str, output_path: str = "data", required_uuid_list: List[str] = None, filter_enabled: bool = True):
+        """
+        Initialize the DatapointDB class.
+
+        :param device_names: List of device names to retrieve metadata for.
+        :param db_user: Database user.
+        :param db_pass: Database password.
+        :param db_host: Database host.
+        :param output_path: Directory to save JSON files.
+        :param required_uuid_list: List of UUIDs to filter the metadata (optional).
+        :param filter_enabled: Whether to filter metadata by "enabled == True" and "archived == False" (default is True).
+        """
+        self.device_names = device_names
         self.db_user = db_user
         self.db_pass = db_pass
         self.db_host = db_host
         self.output_path = output_path
-        self.uuids: List[str] = []
-        self.metadata: pd.DataFrame = pd.DataFrame([])
+        self.required_uuid_list = required_uuid_list or []
+        self.filter_enabled = filter_enabled
+        self.device_metadata: Dict[str, pd.DataFrame] = {}  # Store metadata for each device
+        self.device_uuids: Dict[str, List[str]] = {}  # Store UUIDs for each device
         self._db_access()
 
     def _db_access(self) -> None:
-        """Connect to the database and retrieve metadata for the specified device."""
+        """Connect to the database and retrieve metadata for each device."""
         conn = psycopg2.connect(
             dbname="config_repository",
             user=self.db_user,
@@ -30,37 +43,65 @@ def _db_access(self) -> None:
         )
         cursor = conn.cursor()
 
-        cursor.execute(f"""
-            SELECT dp.uuid, dp.label, dp.config
-            FROM data_points dp
-            INNER JOIN devices dev ON dev.id = dp.device_id
-            WHERE dp.enabled = true AND dp.archived = false AND dev.name = %s
-        """, (self.device_name,))
+        for device_name in self.device_names:
+            query = """
+                SELECT dp.uuid, dp.label, dp.config
+                FROM data_points dp
+                INNER JOIN devices dev ON dev.id = dp.device_id
+                WHERE dev.name = %s
+            """
+            if self.filter_enabled:
+                query += " AND dp.enabled = true AND dp.archived = false"
 
-        data_points = [{"uuid": r[0], "label": r[1], "config": r[2]} for r in cursor.fetchall()]
-        conn.close()
+            cursor.execute(query, (device_name,))
+            data_points = [{"uuid": r[0], "label": r[1], "config": r[2]} for r in cursor.fetchall()]
+
+            # Convert to DataFrame and filter by required UUIDs if necessary
+            metadata_df = pd.DataFrame(data_points)
+            if not metadata_df.empty and self.required_uuid_list:
+                metadata_df = metadata_df[metadata_df["uuid"].isin(self.required_uuid_list)]
 
-        self.metadata = pd.DataFrame(data_points)
-        self._export_json(data_points)
-        self.uuids = [data["uuid"] for data in data_points]
+            # Store metadata and UUIDs for the device
+            self.device_metadata[device_name] = metadata_df
+            self.device_uuids[device_name] = metadata_df["uuid"].tolist()
+
+            # Export to JSON file
+            self._export_json(metadata_df.to_dict(orient="records"), device_name)
+
+        conn.close()
 
-    def _export_json(self, data_points: List[Dict[str, str]]) -> None:
-        """Export data points to a JSON file."""
-        with open(self.output_path, 'w') as f:
+    def _export_json(self, data_points: List[Dict[str, str]], device_name: str) -> None:
+        """Export data points to a JSON file for the specified device."""
+        file_name = f"{self.output_path}/{device_name.replace(' ', '_')}_data_points.json"
+        with open(file_name, 'w') as f:
             json.dump(data_points, f, indent=2)
 
-    def get_uuids(self) -> List[str]:
-        """Return the list of UUIDs."""
-        return self.uuids
+    def get_all_uuids(self) -> Dict[str, List[str]]:
+        """Return a dictionary of UUIDs for each device."""
+        return self.device_uuids
 
-    def get_full_config(self) -> List[Dict[str, str]]:
-        """Return the full configuration (uuid, label, config) as a list of dictionaries."""
-        return self.metadata.to_dict(orient="records")
+    def get_all_metadata(self) -> Dict[str, List[Dict[str, str]]]:
+        """Return a dictionary of metadata for each device."""
+        return {device: metadata.to_dict(orient="records") for device, metadata in self.device_metadata.items()}
 
-    def get_uuid_label_pairs(self) -> List[Dict[str, str]]:
-        """Return a list of uuid and label pairs."""
-        return self.metadata[['uuid', 'label']].to_dict(orient='records')
+    def display_dataframe(self, device_name: str = None, aggregate: bool = False) -> None:
+        """
+        Display metadata as a DataFrame for a specific device or all devices.
 
-    def display_dataframe(self) -> None:
-        """Print the metadata DataFrame to visually inspect data points."""
-        print(self.metadata)
+        :param device_name: Name of the device to display metadata for (optional).
+        :param aggregate: If True, combine metadata from all devices into a single DataFrame.
+        """
+        if aggregate:
+            combined_df = pd.concat(self.device_metadata.values(), keys=self.device_metadata.keys())
+            print("Aggregated metadata for all devices:")
+            print(combined_df)
+        elif device_name:
+            if device_name in self.device_metadata:
+                print(f"Metadata for device: {device_name}")
+                print(self.device_metadata[device_name])
+            else:
+                print(f"No metadata found for device: {device_name}")
+        else:
+            for device, metadata in self.device_metadata.items():
+                print(f"\nMetadata for device: {device}")
+                print(metadata)