Skip to content

ts_shape.loader.metadata.metadata_db_loader ¤

Classes:

  • DatapointDB –

    Class for accessing datapoints via a database.

DatapointDB ¤

DatapointDB(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)

Class for accessing datapoints via a database.

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

Methods:

Source code in src/ts_shape/loader/metadata/metadata_db_loader.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
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.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()

display_dataframe ¤

display_dataframe(device_name: str = None, aggregate: bool = False) -> None

Display metadata as a DataFrame for a specific device or all devices.

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

Source code in src/ts_shape/loader/metadata/metadata_db_loader.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def display_dataframe(self, device_name: str = None, aggregate: bool = False) -> None:
    """
    Display metadata as a DataFrame for a specific device or all devices.

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

get_all_metadata ¤

get_all_metadata() -> Dict[str, List[Dict[str, str]]]

Return a dictionary of metadata for each device.

Source code in src/ts_shape/loader/metadata/metadata_db_loader.py 
83
84
85
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()}

get_all_uuids ¤

get_all_uuids() -> Dict[str, List[str]]

Return a dictionary of UUIDs for each device.

Source code in src/ts_shape/loader/metadata/metadata_db_loader.py 
79
80
81
def get_all_uuids(self) -> Dict[str, List[str]]:
    """Return a dictionary of UUIDs for each device."""
    return self.device_uuids