Altergo API

Client Objects

class Client()

A client to interact with the Altergo API, with customizable endpoints for both factory and IoT services.

Arguments:

• apiKey str - The API key used for authentication with the Altergo API.
• companyName str, optional - The name of the company used to auto-generate default URLs for factory and IoT APIs. If provided, URLs will default to:
  • Factory API: https://companyName.altergo.io
  • IoT API: https://iot.companyName.altergo.io
• **kwargs - Additional keyword arguments to manually override API URLs.
  • iot_api (str, optional): A custom URL for the IoT API.
  • factory_api (str, optional): A custom URL for the Factory API.

Raises:

• ValueError - If neither companyName nor iot_api and factory_api are provided.

Attributes:

• _altergo_client AltergoClient - A client instance that communicates with the Altergo API, initialized with the provided API key and base URLs.

  Usage: You can either provide a companyName to auto-generate the API URLs or pass custom URLs directly through the iot_api and factory_api keyword arguments.

Examples:

client = Client(apiKey="your_api_key", companyName="your_company")
# OR
client = Client(apiKey="your_api_key", iot_api="https://custom.iot.api", factory_api="https://custom.factory.api")

createActivity

def createActivity(newActivity: Activity) -> Union[Activity, None]

Add a new Acitvity to one or more entity.

Arguments:

• newActivity Activity - Activity object to be created.

Returns:

• Activity - Created activity object. with attributes filled by the server.

getActivityById

def getActivityById(
        activityId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Activity, None]

Get an activity by its id.

Arguments:

• activityId str - Id of the activity to be retrieved.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Activity - Retrieved activity object.

getActivities

def getActivities(filterBy: Optional[Dict[Any, Any]] = None,
                  populate: Optional[Dict[Any, Any]] = None,
                  projection: Optional[Dict[Any, Any]] = None,
                  skipTo: int = 0,
                  limitBy: int = 100) -> List[Activity]

Get a list of activities.

Arguments:

• filterBy Optional[Dict[Any, Any]] - Filter the activities by the given criteria. Defaults to None. Example: {"id": "5f7b9b0b0e9b4b0001b0b0b0", "name": "Sample activity"}
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip to the given index. Defaults to 0.
• limitBy int - Limit the number of activities to be retrieved. Defaults to 100.

Returns:

• List[Activity] - List of retrieved activities.

updateActivity

def updateActivity(updatedActivity: Activity) -> Union[Activity, None]

Update an existing activity.

Arguments:

• updatedActivity Activity - Activity object to be updated.

Returns:

• Activity - Updated activity object.

deleteActivity

def deleteActivity(existingActivity: Activity) -> Union[Dict[Any, Any], None]

Delete an existing activity.

Arguments:

• existingActivity Activity - Activity object to be deleted.

Returns:

Dict[Any, Any]: Response from the server. [ErrorCode, Message]

createAsset

def createAsset(newAsset: Asset,
                startDate: Optional[str] = None) -> Union[Asset, None]

Create a new asset.

Arguments:

• newAsset Asset - Asset object to be created.
• startDate Date - Date when Asset is to be created.

Returns:

• Asset - Created asset object. with attributes filled by the server.

getOrCreateAsset

def getOrCreateAsset(
        serialNumber: str,
        blueprintName: str,
        startDateString: Optional[str] = None) -> Union[Asset, None]

Get or create an asset by its serial number.

Arguments:

• serialNumber str - Serial number of the asset.
• startDate Date - Date when Asset is to be created.

Returns:

• Asset - Retrieved or created asset object.

createAssetsFromBlueprintInterfaceAndAssemble

def createAssetsFromBlueprintInterfaceAndAssemble(
        asset: Asset,
        interface: Interface,
        compatibleBlueprint: Blueprint,
        quantity: Optional[int] = None,
        snLabel: Optional[str] = None,
        snSeriesStart: int = 1,
        startDateString: Optional[str] = datetime.now().strftime(
            "%Y-%m-%dT%H:%M:%SZ"),
        assemblyDateString: Optional[str] = None,
        zeroPadding: Optional[int] = 3) -> List[Asset]

Creates child assets from a given asset using a compatible blueprint and assembles them onto the parent asset.

Arguments:

• asset Asset - The parent asset to which child assets will be assembled.
• interface Interface - The interface that connects the asset with child assets, containing possible blueprints.
• compatibleBlueprint Blueprint - The blueprint that defines the specifications for creating compatible child assets.
• quantity Optional[int], optional - The number of child assets to create. If not provided, the method defaults to filling available slots on the interface.
• snLabel Optional[str], optional - A prefix for the serial number of the child assets. If not provided, a default prefix based on the asset and interface names will be generated.
• snSeriesStart int, optional - The starting number for the serial number series of the child assets. Defaults to 1.
• startDateString Optional[str], optional - The start date for the child assets, formatted as an ISO 8601 string (default: current date and time).
• assemblyDateString Optional[str], optional - The assembly date for the child assets. Defaults to startDateString if not provided.
• zeroPadding Optional[int], optional - The number of digits for zero-padding the serial number suffix. Defaults to 3.

Returns:

• List[Asset] - A list of created child assets, as well as the updated parent asset with assembled children.

Raises:

• ValueError - If no compatible blueprint is found in the provided interface.
• Warning - If no child assets are created.

Examples:

```python
asset, childAssets = createAssetsFromBlueprintInterfaceAndAssemble(
    asset=parentAsset, 
    interface=assetInterface, 
    compatibleBlueprint=selectedBlueprint, 
    quantity=5, 
    snLabel="SN123", 
    snSeriesStart=100
)
```

getAssetById

def getAssetById(
        assetId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Asset, None]

Get an asset by its id.

Arguments:

• assetId str - Id of the asset to be retrieved.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Asset - Retrieved asset object.

getAssetBySerialNumber

def getAssetBySerialNumber(
        serialNumber: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Asset, None]

Get an asset by its serial number.

Arguments:

• serialNumber str - Serial number of the asset to be retrieved.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Asset - Retrieved asset object.

getAssets

def getAssets(filterBy: Optional[Dict[Any, Any]] = None,
              populate: Optional[Dict[Any, Any]] = None,
              projection: Optional[Dict[Any, Any]] = None,
              skipTo: int = 0,
              limitBy: int = 100) -> List[Asset]

Get a list of assets.

Arguments:

• filterBy Optional[Dict[Any, Any]] - Filter the assets by the given criteria. Defaults to None. Example: {"id": "5f7b9b0b0e9b4b0001b0b0b0", "serialNumber": "SN-0001"}
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip to the given index. Defaults to 0.
• limitBy int - Limit the number of assets to be retrieved. Defaults to 100.

Returns:

• List[Asset] - List of retrieved assets.

updateAsset

def updateAsset(updatedAsset: Asset) -> Union[Asset, None]

Update an existing asset.

Arguments:

• updatedAsset Asset - Asset object to be updated.

Returns:

• Asset - Updated asset object.

updateAssetState

def updateAssetState(existingAsset: Asset,
                     newState: AssetState) -> Union[Asset, None]

Update the state of an existing asset.

Arguments:

• existingAsset Asset - Asset object to be updated.

Returns:

• Asset - Updated asset object.

editAssetState

def editAssetState(existingAsset: Asset,
                   newState: AssetState) -> Union[Asset, None]

Update an existing state of an asset.

Arguments:

• existingAsset Asset - Asset object to be updated.

Returns:

• Asset - Updated asset object.

deleteAsset

def deleteAsset(existingAsset: Asset) -> Union[Dict[Any, Any], None]

Delete an existing asset.

Arguments:

• existingAsset Asset - Asset object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

deleteAssetAndChildren

def deleteAssetAndChildren(existingAsset: Asset)

Recursively delete an asset and its children.

Arguments:

• asset Asset - Asset object to be deleted.

Returns:

None

assembleAsset

def assembleAsset(parentAsset: Asset,
                  childrenAssetsAndInterfaces: List[Dict[Asset, Interface]],
                  assemblyDate: Optional[str] = None)

Assemble a parent asset with its children assets via their interfaces. TODO: complete documentation for this method.

Arguments:

• parentAsset Asset - Parent asset object.
• childrenAssetsAndInterfaces List[Dict[Asset, Interface]] - List of children assets and their interfaces.

Returns:

Dict[Any, Any]: Response from the server.

disassembleAsset

def disassembleAsset(parentAsset: Asset,
                     childrenAssets: List[Asset],
                     assemblyDate: Optional[str] = None) -> Union[Asset, None]

Disassemble a parent asset from its children assets.

Arguments:

• parentAsset Asset - Parent asset object.
• childrenAssets List[Asset] - List of children assets.

Returns:

• Asset - Updated asset object.

createBlueprint

def createBlueprint(newBlueprint: Blueprint) -> Union[Blueprint, None]

Create a new blueprint.

Arguments:

• newBlueprint Blueprint - Blueprint object to be created.

Returns:

• Blueprint - Created blueprint object.

cloneBlueprint

def cloneBlueprint(blueprintToClone: Blueprint) -> Blueprint

Clone a blueprint.

Arguments:

• blueprintToClone Blueprint - Blueprint object to clone.

Returns:

• Blueprint - Cloned blueprint object.

createBatteryBlueprint

def createBatteryBlueprint(name, cellChemistry, avgCapacity, cellNbSeries,
                           cellNbParallel, energy, currentDischarge,
                           currentCharge, minOperatingTemperature,
                           maxOperatingTemperature, minOperatingSoc,
                           maxOperatingSoc, endOfLifeCapacity, minVoltage,
                           nominalVoltage, maxVoltage) -> Blueprint

Creates a battery blueprint object, pre-configured with key parameters for battery performance and compatible with battery report generation.

Arguments:

• name str - The name of the battery blueprint.
• cellChemistry str - The cell chemistry used in the battery (e.g., Lithium-ion, Nickel-metal hydride).
• avgCapacity float - The average capacity of the battery in Ah (Amp-hours).
• cellNbSeries int - The number of cells connected in series.
• cellNbParallel int - The number of cells connected in parallel.
• energy float - The total energy capacity of the battery in Wh (Watt-hours).
• currentDischarge float - The nominal discharge current of the battery in Amperes.
• currentCharge float - The nominal charge current of the battery in Amperes.
• minOperatingTemperature float - The minimum operating temperature of the battery in degrees Celsius.
• maxOperatingTemperature float - The maximum operating temperature of the battery in degrees Celsius.
• minOperatingSoc float - The minimum state of charge (SOC) the battery can operate at (percentage).
• maxOperatingSoc float - The maximum state of charge (SOC) the battery can operate at (percentage).
• endOfLifeCapacity float - The end-of-life capacity threshold of the battery (as a percentage of original capacity).
• minVoltage float - The minimum operating voltage of the battery in Volts.
• nominalVoltage float - The nominal operating voltage of the battery in Volts.
• maxVoltage float - The maximum operating voltage of the battery in Volts.

Returns:

• Blueprint - The created battery blueprint object with all specified parameters set.

Examples:

```python
battery_bp = createBatteryBlueprint(
    name="BatteryModel123",
    cellChemistry="Lithium-ion",
    avgCapacity=3000,
    cellNbSeries=3,
    cellNbParallel=4,
    energy=48.0,
    currentDischarge=10.0,
    currentCharge=5.0,
    minOperatingTemperature=-20,
    maxOperatingTemperature=60,
    minOperatingSoc=20,
    maxOperatingSoc=90,
    endOfLifeCapacity=80,
    minVoltage=3.0,
    nominalVoltage=3.7,
    maxVoltage=4.2
)
```

getBlueprintById

def getBlueprintById(
        blueprintId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Blueprint, None]

Get a blueprint by its id.

Arguments:

• blueprintId str - Id of the blueprint.

Returns:

• Blueprint - Blueprint object.

getBlueprints

def getBlueprints(filterBy: Optional[Dict[Any, Any]] = None,
                  populate: Optional[Dict[Any, Any]] = None,
                  projection: Optional[Dict[Any, Any]] = None,
                  skipTo: int = 0,
                  limitBy: int = 100) -> List[Blueprint]

Get a list of blueprints.

Arguments:

• filterBy Optional[Dict[Any, Any]] - Filter the list of blueprints. e.g. {"name": "my_blueprint"}
• skipTo int - Skip to a specific index in the list.
• limitBy int - Limit the number of blueprints to be returned.

Returns:

• List[Blueprint] - List of blueprint objects.

updateBlueprint

def updateBlueprint(updatedBlueprint: Blueprint) -> Union[Blueprint, None]

Update a blueprint.

Arguments:

• updatedBlueprint Blueprint - Blueprint object to be updated.

Returns:

• Blueprint - Updated blueprint object.

uploadBlueprintImage

def uploadBlueprintImage(existingBlueprint: Blueprint, imageFilePath: str)

Upload a blueprint image. limitation: The image file size should be less than 1MB. The image file type should be either PNG or JPG.

Arguments:

• existingBlueprint Blueprint - Blueprint object.
• imageFilePath str - Path to the image file.

Returns:

None

deleteBlueprint

def deleteBlueprint(
        existingBlueprint: Blueprint) -> Union[Dict[Any, Any], None]

Delete a blueprint.

Arguments:

• existingBlueprint Blueprint - Blueprint object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

createCategory

def createCategory(newCategory: Category) -> Union[Category, None]

Create a new category.

Arguments:

• newCategory Category - Category object to be created.

Returns:

• Category - Created category object.

getCategoryById

def getCategoryById(
        categoryId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Category, None]

Get a category by id.

Arguments:

• categoryId str - Category id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Category - Category object.

getCategories

def getCategories(filterBy: Optional[Dict[Any, Any]] = None,
                  populate: Optional[Dict[Any, Any]] = None,
                  projection: Optional[Dict[Any, Any]] = None,
                  skipTo: int = 0,
                  limitBy: int = 100) -> List[Category]

Get a list of categories.

Arguments:

• filterBy Dict[Any, Any] - Filter the list of categories. eg: {"name": "test"}
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip the number of categories to be returned.
• limitBy int - Limit the number of categories to be returned.

Returns:

• List[Category] - List of category objects.

updateCategory

def updateCategory(updatedCategory: Category) -> Union[Category, None]

Update a category.

Arguments:

• updatedCategory Category - Category object to be updated.

Returns:

• Category - Updated category object.

deleteCategory

def deleteCategory(existingCategory: Category) -> Union[Dict[Any, Any], None]

Delete a category.

Arguments:

• existingCategory Category - Category object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

createCompany

def createCompany(newCompany: Company) -> Union[Company, None]

Create a new company.

Arguments:

• newCompany Company - Company object to be created.

Returns:

• Company - Created company object.

getCompanyById

def getCompanyById(companyId: str) -> Union[Company, None]

Get a company by id.

Arguments:

• companyId str - Company id.

Returns:

• Company - Company object.

createDashboard

def createDashboard(newDashboard: Dashboard) -> Union[Dashboard, None]

Create a new dashboard.

Arguments:

• newDashboard Dashboard - Dashboard object to be created.

Returns:

• Dashboard - Created dashboard object.

getDashboardById

def getDashboardById(
        dashboardId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Dashboard, None]

Get a dashboard by id.

Arguments:

• dashboardId str - Dashboard id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Dashboard - Dashboard object.

getDashboards

def getDashboards(filterBy: Optional[Dict[Any, Any]] = None,
                  populate: Optional[Dict[Any, Any]] = None,
                  projection: Optional[Dict[Any, Any]] = None,
                  skipTo: int = 0,
                  limitBy: int = 100) -> List[Dashboard]

Get a list of dashboards.

Arguments:

• filterBy Dict[Any, Any] - Filter the list of dashboards. eg: {"name": "test"}
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip the number of dashboards to be returned.
• limitBy int - Limit the number of dashboards to be returned.

Returns:

• List[Dashboard] - List of dashboard objects.

updateDashboard

def updateDashboard(updatedDashboard: Dashboard) -> Union[Dashboard, None]

Update a dashboard.

Arguments:

• updatedDashboard Dashboard - Dashboard object to be updated.

Returns:

• Dashboard - Updated dashboard object.

deleteDashboard

def deleteDashboard(
        existingDashboard: Dashboard) -> Union[Dict[Any, Any], None]

Delete a dashboard.

Arguments:

• existingDashboard Dashboard - Dashboard object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

createParameter

def createParameter(newParameter: Parameter) -> Union[Parameter, None]

Create a new parameter.

Arguments:

• newParameter Parameter - Parameter object to be created.

Returns:

• Parameter - Created parameter object.

getParameterById

def getParameterById(
        parameterId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Parameter, None]

Get a blueprint parameter by id.

Arguments:

• parameterId str - Parameter id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Parameter - Parameter object.

getParameters

def getParameters(filterBy: Optional[Dict[Any, Any]] = None,
                  populate: Optional[Dict[Any, Any]] = None,
                  projection: Optional[Dict[Any, Any]] = None,
                  skipTo: int = 0,
                  limitBy: int = 100) -> List[Parameter]

Get a list of blueprint parameters.

Arguments:

• filterBy Dict[Any, Any] - Filter the list of parameters. eg: {"name": "test"}
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip the number of parameters to be returned.
• limitBy int - Limit the number of parameters to be returned.

Returns:

• List[Parameter] - List of parameter objects.

updateParameter

def updateParameter(updatedParameter: Parameter) -> Union[Parameter, None]

Update a blueprint parameter.

Arguments:

• updatedParameter Parameter - Parameter object to be updated.

Returns:

• Parameter - Updated parameter object.

deleteParameter

def deleteParameter(
        existingParameter: Parameter) -> Union[Dict[Any, Any], None]

Delete a blueprint parameter.

Arguments:

• existingParameter Parameter - Parameter object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

createParameterNamePrefix

def createParameterNamePrefix(
    newParameterNamePrefix: ParameterNamePrefix
) -> Union[ParameterNamePrefix, None]

Create a new blueprint parameter name prefix.

Arguments:

• newParameterNamePrefix ParameterNamePrefix - Parameter name prefix object to be created.

Returns:

• ParameterNamePrefix - Created parameter name prefix object.

getParameterNamePrefixById

def getParameterNamePrefixById(
        parameterNamePrefixId: str) -> Union[ParameterNamePrefix, None]

Get a blueprint parameter name prefix by id.

Arguments:

• parameterNamePrefixId str - Parameter name prefix id.

Returns:

• ParameterNamePrefix - Parameter name prefix object.

getParameterNamePrefixes

def getParameterNamePrefixes(filterBy: Optional[Dict[Any, Any]] = None,
                             skipTo: int = 0,
                             limitBy: int = 100) -> List[ParameterNamePrefix]

Get a list of blueprint parameter name prefixes.

Arguments:

• filterBy Dict[Any, Any] - Filter the list of parameter name prefixes. eg: {"name": "test"}
• skipTo int - Skip the number of parameter name prefixes to be returned.
• limitBy int - Limit the number of parameter name prefixes to be returned.

Returns:

• List[ParameterNamePrefix] - List of parameter name prefix objects.

updateParameterNamePrefix

def updateParameterNamePrefix(
    updatedParameterNamePrefix: ParameterNamePrefix
) -> Union[ParameterNamePrefix, None]

Update a blueprint parameter name prefix.

Arguments:

• updatedParameterNamePrefix ParameterNamePrefix - Parameter name prefix object to be updated.

Returns:

• ParameterNamePrefix - Updated parameter name prefix object.

deleteParameterNamePrefix

def deleteParameterNamePrefix(
    existingParameterNamePrefix: ParameterNamePrefix
) -> Union[Dict[Any, Any], None]

Delete a blueprint parameter name prefix.

Arguments:

• existingParameterNamePrefix ParameterNamePrefix - Parameter name prefix object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

createPermission

def createPermission(newPermission: Permission) -> Union[Permission, None]

Create a new user permission.

Arguments:

• newPermission Permission - Permission object to be created.

Returns:

• Permission - Created permission object.

getPermissionById

def getPermissionById(permissionId: str) -> Union[Permission, None]

Get a user permission by id.

Arguments:

• permissionId str - Permission id.

Returns:

• Permission - Permission object.

getPermissions

def getPermissions(filterBy: Optional[Dict[Any, Any]] = None,
                   skipTo: int = 0,
                   limitBy: int = 100) -> List[Permission]

Get a list of user permissions.

Arguments:

• filterBy Dict[Any, Any] - Filter the list of permissions. eg: {"name": "test"}
• skipTo int - Skip the number of permissions to be returned.
• limitBy int - Limit the number of permissions to be returned.

Returns:

• List[Permission] - List of permission objects.

updatePermission

def updatePermission(updatedPermission: Permission) -> Union[Permission, None]

Update a user permission.

Arguments:

• updatedPermission Permission - Permission object to be updated.

Returns:

• Permission - Updated permission object.

deletePermission

def deletePermission(
        existingPermission: Permission) -> Union[Dict[Any, Any], None]

Delete a user permission.

Arguments:

• existingPermission Permission - Permission object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

createProject

def createProject(newProject: Project) -> Union[Project, None]

Create a new dashboard project.

Arguments:

• newProject Project - Project object to be created.

Returns:

• Project - Created project object.

getProjectById

def getProjectById(
        projectId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Project, None]

Get a dashboard project by id.

Arguments:

• projectId str - Project id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Project - Project object.

getProjects

def getProjects(filterBy: Optional[Dict[Any, Any]] = None,
                populate: Optional[Dict[Any, Any]] = None,
                projection: Optional[Dict[Any, Any]] = None,
                skipTo: int = 0,
                limitBy: int = 100) -> List[Project]

Get a list of dashboard projects.

Arguments:

• filterBy Dict[Any, Any] - Filter the list of projects. eg: {"name": "test"}
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip the number of projects to be returned.
• limitBy int - Limit the number of projects to be returned.

Returns:

• List[Project] - List of project objects.

updateProject

def updateProject(updatedProject: Project) -> Union[Project, None]

Update a dashboard project.

Arguments:

• updatedProject Project - Project object to be updated.

Returns:

• Project - Updated project object.

deleteProject

def deleteProject(existingProject: Project) -> Union[Dict[Any, Any], None]

Delete a dashboard project.

Arguments:

• existingProject Project - Project object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

createSensor

def createSensor(newSensor: Sensor) -> Union[Sensor, None]

Create a new blueprint sensor.

Arguments:

• newSensor Sensor - Sensor object to be created.

Returns:

• Sensor - Created sensor object.

getSensorById

def getSensorById(
        sensorId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Sensor, None]

Get a new blueprint sensor by id.

Arguments:

• sensorId str - Sensor id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Sensor - Sensor object.

getSensors

def getSensors(filterBy: Optional[Dict[Any, Any]] = None,
               populate: Optional[Dict[Any, Any]] = None,
               projection: Optional[Dict[Any, Any]] = None,
               skipTo: int = 0,
               limitBy: int = 100) -> List[Sensor]

Get a list of blueprint sensors.

Arguments:

• filterBy Dict[Any, Any] - Filter the list of sensors. eg: {"name": "test"}
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip the number of sensors to be returned.
• limitBy int - Limit the number of sensors to be returned.

Returns:

• List[Sensor] - List of sensor objects.

updateSensor

def updateSensor(updatedSensor: Sensor) -> Union[Sensor, None]

Update a blueprint sensor.

Arguments:

• updatedSensor Sensor - Sensor object to be updated.

Returns:

• Sensor - Updated sensor object.

deleteSensor

def deleteSensor(existingSensor: Sensor) -> Union[Dict[Any, Any], None]

Delete a blueprint sensor.

Arguments:

• existingSensor Sensor - Sensor object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

createSiPrefix

def createSiPrefix(newSiPrefix: SiPrefix) -> Union[SiPrefix, None]

Create a new parameter SI (International System of Units)prefix.

Arguments:

• newSiPrefix SiPrefix - SiPrefix object to be created.

Returns:

• SiPrefix - Created SI (International System of Units)prefix object.

getSiPrefixById

def getSiPrefixById(siPrefixId: str) -> Union[SiPrefix, None]

Get a new parameter SI(International System of Units) prefix by id.

Arguments:

• siPrefixId str - SiPrefix id.

Returns:

• SiPrefix - SiPrefix object.

getSiPrefixes

def getSiPrefixes(filterBy: Optional[Dict[Any, Any]] = None,
                  skipTo: int = 0,
                  limitBy: int = 100) -> List[SiPrefix]

Get a list of parameter SI (International System of Units) prefixes.

Arguments:

• filterBy Dict[Any, Any] - Filter the list of SiPrefixes. eg: {"name": "test"}
• skipTo int - Skip the number of SiPrefixes to be returned.
• limitBy int - Limit the number of SiPrefixes to be returned.

Returns:

• List[SiPrefix] - List of SiPrefix objects.

updateSiPrefix

def updateSiPrefix(updatedSiPrefix: SiPrefix) -> Union[SiPrefix, None]

Update a parameter SI (International System of Units) prefix. such as "kilo", "mega", "giga", "tera","deca", "hecto", "deci", "centi", "milli", "micro", "nano"

Arguments:

• updatedSiPrefix SiPrefix - SiPrefix object to be updated.

Returns:

• SiPrefix - Updated SiPrefix object.

deleteSiPrefix

def deleteSiPrefix(existingSiPrefix: SiPrefix) -> Union[Dict[Any, Any], None]

Delete a parameter SI (International System of Units) prefix. such as "kilo", "mega", "giga", "tera","deca", "hecto", "deci", "centi", "milli", "micro", "nano"

Arguments:

• existingSiPrefix SiPrefix - SiPrefix object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

createSiUnit

def createSiUnit(newSiUnit: SiUnit) -> Union[SiUnit, None]

Create a new parameter SI (International System of Units) unit.

Arguments:

• newSiUnit SiUnit - SiUnit object to be created.

Returns:

• SiUnit - Created SI (International System of Units) unit object.

getSiUnitById

def getSiUnitById(
        siUnitId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[SiUnit, None]

Get a new parameter SI (International System of Units) unit by id.

Arguments:

• siUnitId str - SiUnit id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• SiUnit - SiUnit object.

getSiUnits

def getSiUnits(filterBy: Optional[Dict[Any, Any]] = None,
               populate: Optional[Dict[Any, Any]] = None,
               projection: Optional[Dict[Any, Any]] = None,
               skipTo: int = 0,
               limitBy: int = 100) -> List[SiUnit]

Get a list of parameter SI (International System of Units) units.

Arguments:

• filterBy Dict[Any, Any] - Filter the list of SiUnits. eg: {"name": "test"}
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip the number of SiUnits to be returned.
• limitBy int - Limit the number of SiUnits to be returned.

Returns:

• List[SiUnit] - List of SiUnit objects.

updateSiUnit

def updateSiUnit(updatedSiUnit: SiUnit) -> Union[SiUnit, None]

Update a parameter SI (International System of Units) unit.

Arguments:

• updatedSiUnit SiUnit - SiUnit object to be updated.

Returns:

• SiUnit - Updated SiUnit object.

deleteSiUnit

def deleteSiUnit(existingSiUnit: SiUnit) -> Union[Dict[Any, Any], None]

Delete a parameter SI (International System of Units) unit.

Arguments:

• existingSiUnit SiUnit - SiUnit object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

createTag

def createTag(newTag: Tag) -> Union[Tag, None]

Create a new tag.

Arguments:

• newTag Tag - Tag object to be created.

Returns:

• Tag - Created Tag object.

getTagById

def getTagById(
        tagId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Tag, None]

Get a tag by id.

Arguments:

• tagId str - Tag id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Tag - Tag object.

getOrCreateTag

def getOrCreateTag(tagName: str, color: str) -> Union[Tag, None]

Retrieves an existing tag by name or creates a new one if it does not exist.

Arguments:

• tagName str - The name of the tag to retrieve or create.
• color str - The color to assign to the tag if a new tag is created.

Returns:

Union[Tag, None]: The existing or newly created Tag object. Returns None if no tag could be created or found.

Behavior:

• If a tag with the specified tagName already exists, it is returned.
• If no exact match for the tag name is found, a new tag is created with the specified tagName and color, and then returned.

Examples:

```python
tag = getOrCreateTag(tagName="Important", color="red")

If the tag "Important" already exists, it will be retrieved. 
If not, a new tag will be created with the name "Important" and the color "red".
```

getSingleTagByName

def getSingleTagByName(
        tagName: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Tag, None]

Get a tag by name.

Arguments:

• tagName str - Tag name.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Tag - Tag object.

getMultipleTagsWithSimilarName

def getMultipleTagsWithSimilarName(
        substring: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Tag, None]

Get a tag by name.

Arguments:

• substring str - substring of tag name.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Tag - Tag object.

getTags

def getTags(filterBy: Optional[Dict[Any, Any]] = None,
            populate: Optional[Dict[Any, Any]] = None,
            projection: Optional[Dict[Any, Any]] = None,
            skipTo: int = 0,
            limitBy: int = 100) -> List[Tag]

Get a list of tags.

Arguments:

• filterBy Dict[Any, Any] - Filter the list of Tags. eg: {"name": "test"}
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip the number of Tags to be returned.
• limitBy int - Limit the number of Tags to be returned.

Returns:

• List[Tag] - List of Tag objects.

updateTag

def updateTag(updatedTag: Tag) -> Union[Tag, None]

Update a tag.

Arguments:

• updatedTag Tag - Tag object to be updated.

Returns:

• Tag - Updated Tag object.

deleteTag

def deleteTag(existingTag: Tag) -> Union[Dict[Any, Any], None]

Delete a tag.

Arguments:

• existingTag Tag - Tag object to be deleted.

Returns:

Dict[Any, Any]: Response from the server.

linkBlueprintByTags

def linkBlueprintByTags(tags: List[Tag], existingBlueprint: Blueprint)

Link a blueprint to a list of tags.

Arguments:

• tags List[Tag] - List of Tag objects.
• existingBlueprint Blueprint - Blueprint object.

Returns:

Dict[Any, Any]: Response from the server.

linkTagToAssets

def linkTagToAssets(existingTag: Tag, assets: List[Asset])

Link a tag to a list of assets.

Arguments:

• existingTag Tag - Tag object.
• assets List[Asset] - List of Asset objects.

Returns:

Dict[Any, Any]: Response from the server.

unlinkTagFromAssets

def unlinkTagFromAssets(existingTag: Tag,
                        assets: List[Asset]) -> Union[Dict[Any, Any], None]

Unlink a tag from a list of assets.

Arguments:

• existingTag Tag - Tag object.
• assets List[Asset] - List of Asset objects.

Returns:

Dict[Any, Any]: Response from the server.

getTaskById

def getTaskById(
        taskId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Task, None]

Get a task by its id.

Arguments:

• taskId str - Task id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• Task - Task object.

getTasks

def getTasks(filterBy: Optional[Dict[Any, Any]] = None,
             populate: Optional[Dict[Any, Any]] = None,
             projection: Optional[Dict[Any, Any]] = None,
             skipTo: int = 0,
             limitBy: int = 100) -> List[Task]

Get a list of tasks.

Arguments:

• filterBy Optional[Dict[Any, Any]] - Filter by. Defaults to None.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip to. Defaults to 0.
• limitBy int - Limit by. Defaults to 100.

Returns:

• List[Task] - List of Task objects.

updateTask

def updateTask(existingTask: Task) -> Union[Task, None]

Update a task.

Arguments:

• existingTask Task - Task object.

Returns:

• Task - Task object.

createUser

def createUser(newUser: User) -> Union[User, None]

Create a new user.

Arguments:

• newUser User - User object.

Returns:

• User - User object.

getUserById

def getUserById(
        userId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[User, None]

Get a user by its id.

Arguments:

• userId str - User id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned

Returns:

• User - User object.

getUsers

def getUsers(filterBy: Optional[Dict[Any, Any]] = None,
             populate: Optional[Dict[Any, Any]] = None,
             projection: Optional[Dict[Any, Any]] = None,
             skipTo: int = 0,
             limitBy: int = 100) -> List[User]

Get a list of users.

Arguments:

• filterBy Optional[Dict[Any, Any]] - Filter by. Defaults to None. e.g {"username": "admin"}
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip to. Defaults to 0.
• limitBy int - Limit by. Defaults to 100.

Returns:

• List[User] - List of User objects.

updateUser

def updateUser(updatedUser: User) -> Union[User, None]

Update a user.

Arguments:

• updatedUser User - User object.

Returns:

• User - User object.

deleteUser

def deleteUser(existingUser: User) -> Union[Dict[Any, Any], None]

Delete a user.

Arguments:

• existingUser User - User object.

Returns:

Dict[Any, Any]: Response object.

createRole

def createRole(newRole: Role) -> Union[Role, None]

Create a new role for users.

Arguments:

• newRole Role - Role object.

Returns:

• Role - Role object.

getRoleById

def getRoleById(roleId: str) -> Union[Role, None]

Get a role by its id.

Arguments:

• roleId str - Role id.

Returns:

• Role - Role object.

getRoles

def getRoles(filterBy: Optional[Dict[Any, Any]] = None,
             skipTo: int = 0,
             limitBy: int = 100) -> List[Role]

Get a list of roles.

Arguments:

• filterBy Optional[Dict[Any, Any]] - Filter by. Defaults to None. e.g {"name": "admin"}
• skipTo int - Skip to. Defaults to 0.
• limitBy int - Limit by. Defaults to 100.

Returns:

• List[Role] - List of Role objects.

updateRole

def updateRole(updatedRole: Role) -> Union[Role, None]

Update a role.

Arguments:

• updatedRole Role - Role object.

Returns:

• Role - Role object.

deleteRole

def deleteRole(existingRole: Role) -> Union[Dict[Any, Any], None]

Delete a role.

Arguments:

• existingRole Role - Role object.

Returns:

Dict[Any, Any]: Response object.

waitForTaskToComplete

def waitForTaskToComplete(existingTask: Task, verbose=True) -> Task

Wait for a task to complete serverside.

Arguments:

• existingTask Task - Task object.

Returns:

• Task - Task object.

refactorDataframeToAsset

def refactorDataframeToAsset(
    df: pd.DataFrame,
    existingAsset: Asset,
    sensors: Optional[Dict[Any, Any]] = dict()) -> List[Any]

Refactor the columns of the DataFrame with the correct names of the sensors associated with the existingAsset.

Arguments:

• df pd.DataFrame - DataFrame that has to be refactored.
• existingAsset Asset - Asset for which the DataFrame must be refactored.
• sensors dict, optional - To-do. Defaults to {}.

Returns:

• list - Refactored list of sensor names.

eraseAllDataFromAsset

def eraseAllDataFromAsset(existingAsset: Asset, sync=True) -> Task

Erase all data from an existing asset.

Arguments:

• existingAsset Asset - Asset for which all data must be erased.

Returns:

• Task - Task that is responsible for the deletion of all data from the asset. You can use the task to check the detailed analytics of the deletion process.

eraseAllDataFromAssetBySerialNumber

def eraseAllDataFromAssetBySerialNumber(serialNumber: str) -> Task

Erase all data from an existing asset viat its serial number.

Arguments:

• serialNumber str - Serial number of the asset for which all data must be erased.

Returns:

• Task - Task that is responsible for the deletion of all data from the asset. You can use the task to check the detailed analytics of the deletion process.

eraseSensorDataFromAsset

def eraseSensorDataFromAsset(serialNumbers: List[str],
                             sensorCodes: List[Dict[str, str]], startDate: int,
                             endDate: int,
                             isCleanDelete: bool) -> Union[Task, None]

Erase data from an existing asset. The data to be erased can be filtered by sensor code and date range.

Arguments:

• serialNumbers list - List of serial numbers of the assets for which data must be erased.
• sensorCodes list - List of sensor codes of the sensors for which data must be erased.
• startDate int - Start date of the data to be erased.
• endDate int - End date of the data to be erased.
• isCleanDelete bool - To-do.

getAssetSensorData

def getAssetSensorData(assets: List[Asset],
                       sensors: List[str],
                       startDate: datetime,
                       endDate: datetime,
                       cache: bool = True,
                       verbose: bool = True)

Retrieves and processes sensor data for a list of assets within a specified time range, embedding the data as a pandas DataFrame into the 'df' attribute of each Asset object.

This method interacts with the Altergo system to fetch and organize sensor data for a list of assets. The data is structured into a pandas DataFrame, which is then stored in the df attribute of each asset. Optionally, the function supports caching to reuse previously retrieved data and provides verbose output for monitoring the data retrieval process.

Arguments:

• assets List[Asset] - A list of Asset objects for which sensor data is to be retrieved. All assets must be unique based on their serial_number.

• sensors List[str] - A list of sensor identifiers. Each sensor can be referenced by its name, code, or alias, optionally followed by a position, e.g., "Voltage|0" or "voltage|1". The format should either be "sensorName|position" or just "sensorName".

• startDate datetime - The start datetime for the data retrieval window.

• endDate datetime - The end datetime for the data retrieval window.

• cache bool, optional - If True (default), the function will use cached data if available. If False, it will force a fresh retrieval from Altergo.

• verbose bool, optional - If True (default), detailed output about the data retrieval and processing steps will be printed to the console.

  Output: This function does not return a value. Instead, it updates the df attribute of each Asset object in the assets list with a pandas DataFrame containing the sensor data. The DataFrame is indexed by time and contains one column per sensor.

Raises:

ValueError:

• If duplicate assets are detected in the assets list (based on their serial numbers).
• If duplicate sensors are detected in the sensors list.
• If a sensor format is incorrect (e.g., more than one position provided).
• If a specified sensor is not found in the asset's blueprint.
• If the downloaded data contains unexpected sensor IDs or positions that do not match the asset's blueprint.

Examples:

from datetime import datetime

# Define the start and end date for the data retrieval
start_date = datetime(2021, 1, 1)
end_date = datetime(2021, 2, 1)

# Define the assets and sensors to retrieve data for
assets = [asset1, asset2]
sensors = ["Temperature|0", "Voltage"]

# Retrieve the sensor data and embed it in each asset's 'df' attribute
getAssetSensorData(
    assets=assets,
    sensors=sensors,
    startDate=start_date,
    endDate=end_date,
    cache=True,
    verbose=True
)

Notes:

• The df attribute of each asset will contain a pandas DataFrame with the sensor data, where the index is the timestamp and each column represents a sensor.
• Sensor data is fetched either from cache (if available) or directly from the Altergo system.

sendSensorDataToAssets

def sendSensorDataToAssets(
        existingAssets: Union[Asset, List[Asset]],
        targetedSensorNames: List[str] = None,
        startDate: datetime = None,
        endDate: datetime = None,
        updateMethod: dataUpdateMethod = dataUpdateMethod.INSERT,
        cleanDelete: bool = False,
        quiet: bool = False,
        sync: bool = True,
        downSample: bool = True,
        ignoreInvalidSensors: bool = False)

Sends sensor data from the df attribute of one or more assets to the backend for processing.

Arguments:

• existingAssets Union[Asset, List[Asset]] - A single Asset object or a list of Asset objects. Each asset must have a Pandas DataFrame (df) containing the sensor data.
• targetedSensorNames List[str], optional - A list of sensor names to send. These sensor names should match the column names in the asset's df. If None, all sensors in the df will be sent. Defaults to None.
• startDate datetime, optional - The start date for the sensor data being sent. If not provided, the earliest date in the assets' DataFrames will be used. Defaults to None.
• endDate datetime, optional - The end date for the sensor data being sent. If not provided, the latest date in the assets' DataFrames will be used. Defaults to None.
• updateMethod dataUpdateMethod, optional - Specifies the method for updating the sensor data. Can be:
  • dataUpdateMethod.INSERT (default): Inserts new data.
  • dataUpdateMethod.APPEND: Appends data without overwriting existing data.
  • dataUpdateMethod.REPLACE: Replaces existing data within the specified date range.
• cleanDelete bool, optional - If True, performs a clean delete before sending new data (only relevant for REPLACE method). Defaults to False.
• quiet bool, optional - If True, suppresses print outputs. Defaults to False.
• sync bool, optional - If True, waits for the backend to finish processing the data before returning. This ensures data ingestion is complete. Defaults to True.
• downSample bool, optional - If True, downsamples data to remove periods of inactivity and reduce the amount of data sent. Defaults to True.
• ignoreInvalidSensors bool, optional - If True, skips over any invalid sensors found in the asset's DataFrame. Defaults to False.

Raises:

• ValueError - If no assets are provided or if the df attribute of any asset is None or not properly indexed by datetime.
• ValueError - If sensor position in df columns is invalid (e.g., not an integer or less than 0).

Returns:

If sync is True, waits for data ingestion to complete and returns the result. If sync is False, returns the task associated with the file upload, allowing for asynchronous processing.

Examples:

from datetime import datetime

# Define start and end dates
start_date = datetime(2021, 1, 1)
end_date = datetime(2021, 2, 1)

# Send data to multiple assets
sendSensorDataToAssets(
    existingAssets=[asset1, asset2], 
    targetedSensorNames=["Temperature", "Voltage"], 
    startDate=start_date, 
    endDate=end_date, 
    updateMethod=dataUpdateMethod.REPLACE, 
    sync=True
)

Notes:

Each asset's df attribute must be a valid Pandas DataFrame with a DatetimeIndex and columns corresponding to the sensor names. If targetedSensorNames is not provided, all sensors from the DataFrame will be sent.

quickSensorDataUpload

def quickSensorDataUpload(filepath: str = None,
                          dataframe: pd.DataFrame = None,
                          asset: Asset = None,
                          blueprint: Blueprint = None,
                          newBlueprintName: str = None,
                          newAssetSerialNumber: str = None,
                          targetedDatetimeColumn: str = None,
                          timezone: str = None,
                          selectedColumns: List[str] = None,
                          excludedColumns: List[str] = None,
                          sensorFuzzyMatchingRatio: int = 100,
                          skipIncompatibleColumn: bool = False,
                          skipUnmatchedSensors: bool = False,
                          ingestionMethod=dataUpdateMethod.INSERT) -> Any

Quickly uploads sensor data from a CSV file or DataFrame to the Altergo platform, creating a new asset and/or blueprint if necessary. A fuzzy matching algorithm is used to match the column names to sensors in the blueprint, allowing for flexible sensor ingestion.

Arguments:

• filepath str, optional - Path to the CSV file containing the sensor data. Either filepath or dataframe must be provided.
• dataframe pd.DataFrame, optional - A Pandas DataFrame containing the sensor data. Either filepath or dataframe must be provided.
• asset Asset, optional - An existing asset to which the data will be uploaded. If None, a new asset will be created.
• blueprint Blueprint, optional - An existing blueprint to match the sensors in the CSV file. If None, a new blueprint will be created.
• newBlueprintName str, optional - The name for the new blueprint to be created (if applicable). Defaults to None. If not provided, the CSV file name is used.
• newAssetSerialNumber str, optional - The serial number for the new asset to be created (if applicable). If None, the CSV file name is used. If an asset with the same serial number already exists, an error will be raised.
• targetedDatetimeColumn str, optional - The column name containing datetime values in the CSV or DataFrame. If None, the system will attempt to auto-detect the datetime column.
• timezone str, optional - The timezone of the datetime values in the CSV or DataFrame. Defaults to None.
• selectedColumns List[str], optional - A list of specific columns to include from the CSV or DataFrame. Defaults to None (all columns are included).
• excludedColumns List[str], optional - A list of columns to exclude from the ingestion. Defaults to None (no columns are excluded).
• sensorFuzzyMatchingRatio int, optional - A value between 0 and 100 that determines how strictly column names must match sensor names in the blueprint. Defaults to 100 (exact match).
• skipIncompatibleColumn bool, optional - If True, skips non-numeric columns during ingestion. Defaults to False.
• skipUnmatchedSensors bool, optional - If True, skips any sensors that could not be matched to a column. Defaults to False.
• ingestionMethod dataUpdateMethod, optional - The method for ingesting the data. Can be one of:
  • dataUpdateMethod.INSERT: Inserts new data (default).
  • dataUpdateMethod.APPEND: Appends data without overwriting existing data.
  • dataUpdateMethod.REPLACE: Replaces existing data with the new data.

Returns:

• dict - A dictionary containing:
  • "asset": The asset to which the data was ingested.
  • "blueprint": The blueprint used for ingestion.
  • "ingestionResult": The result object containing the ingestion task and report.

Raises:

• ValueError - If both filepath and dataframe are None or if both are provided.
• ValueError - If the file provided is not a CSV or does not exist.
• ValueError - If both blueprint and asset are provided.
• ValueError - If an asset with the same serial number already exists.
• ValueError - If the asset's df attribute is missing or improperly indexed.

Examples:

# Upload sensor data from a CSV file, creating a new asset and blueprint automatically
quickSensorDataUpload(
    filepath="path/to/sensor_data.csv",
    newBlueprintName="BatteryModel2023",
    newAssetSerialNumber="BAT12345",
    targetedDatetimeColumn="timestamp",
    selectedColumns=["Temperature", "Voltage"],
    sensorFuzzyMatchingRatio=90,
    skipIncompatibleColumn=True
)

Notes:

• Either a filepath or dataframe must be provided. If both are provided, an error will be raised.
• If asset is not provided, a new asset will be created using the newAssetSerialNumber.
• If blueprint is not provided, a new blueprint will be created based on the data in the CSV or DataFrame.
• The df attribute of the asset will be populated with the processed data after ingestion.

createBlueprintFromDataframe

def createBlueprintFromDataframe(dataframe: pd.DataFrame,
                                 newBlueprintName: str = None,
                                 sensorFuzzyMatchingRatio: int = 100,
                                 createSensorsIfNotFound: bool = True)

Quickly create a blueprint from a dataframe. A fuzzy matching algorithm will be used to match the column names to the sensors in the blueprint. You can configure the matching ratio using the sensorFuzzyMatchingRatio parameter.

Arguments:

• dataframe DataFrame - The name of the blueprint to be created.
• newBlueprintName str, optional - The name of the blueprint to be created. Defaults to None. If None, the name of the dataframe file will be used.
• sensorFuzzyMatchingRatio int, optional - The ratio used to match the sensors in the blueprint to the columns in the CSV file. Value between 0 and 100. Defaults to 100 (exact match). The higher the ratio, the more strict the matching will be.
• createSensorsIfNotFound bool, optional - Create sensors if not found. Defaults to True. If True, sensors that are not matched to a column will be created.

Returns:

• Blueprint - The newly created blueprint

createDataset

def createDataset(
        newDataset: Dataset,
        filePath: str,
        readMethod: str = FileHandleTypeConstants.READ
) -> Union[Dataset, None]

Creates a dataset and uploads a file associated with the dataset to the Altergo platform.

Arguments:

• newDataset Dataset - The Dataset object containing metadata such as name, description, activities, assets, blueprints, and tags.
• filePath str - The full path of the file to be uploaded along with the dataset. The file should contain relevant data for the dataset.
• readMethod str, optional - The file reading method, typically used for text or binary files. Defaults to FileHandleTypeConstants.READ (for reading text files). Use FileHandleTypeConstants.READBINARY for binary files (e.g., images, PDFs).

Returns:

Union[Dataset, None]: The created Dataset object if the operation is successful, otherwise None.

Raises:

• ValueError - If the file cannot be found or read using the specified filePath or readMethod.

• RuntimeWarning - If the dataset creation or file upload fails, a warning will be raised with details of the failure.

  Workflow:

  1. The newDataset object should contain attributes such as activities, assets, blueprints, and tags (each of which may be optional).
  2. The file specified by filePath is opened and read based on the given readMethod.
  3. The function collects IDs from the newDataset object for activities, assets, blueprints, and tags.
  4. These IDs, along with the dataset's name and description, are sent to the Altergo API.
  5. The file is uploaded alongside the dataset metadata as part of the API request.
  6. If the API call is successful, the newly created dataset is returned. If the call fails, a warning is raised.

Examples:

new_dataset = Dataset(name="Test Dataset", description="A dataset for testing.")
dataset = createDataset(
    newDataset=new_dataset, 
    filePath="/path/to/datafile.csv", 
    readMethod=FileHandleTypeConstants.READ
)

# If successful, the `dataset` object will contain all details about the newly created dataset.

Notes:

• The file path must be valid, and the file should be accessible for reading.
• Ensure that the newDataset object contains the necessary metadata (activities, assets, blueprints, or tags) before calling this method.

getDatasetById

def getDatasetById(
        datasetId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Dataset, None]

Gets a dataset by id.

Arguments:

• datasetId str - Dataset id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned return:
• Dataset - Dataset object.

getDatasets

def getDatasets(filterBy: Optional[Dict[Any, Any]] = None,
                populate: Optional[Dict[Any, Any]] = None,
                projection: Optional[Dict[Any, Any]] = None,
                skipTo: int = 0,
                limitBy: int = 100) -> List[Dataset]

Gets all datasets matching the filter criteria. The size of the return is limited by skipTo and limitBy.

Arguments:

• filterBy Optional[Dict[Any, Any]] - Filter criteria.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip to.
• limitBy int - Limit by. return:
• List[Dataset] - List of datasets.

updateDataset

def updateDataset(updatedDataset: Dataset,
                  filePath: str = None,
                  fileName: str = None) -> Union[Dataset, None]

Updates a dataset.

Arguments:

• updatedDataset Dataset - Dataset to be updated.
• filePath str - File path.
• fileName str - File name. return:
• Dataset - Updated dataset.

deleteDataset

def deleteDataset(existingDataset: Dataset) -> Union[Dict[Any, Any], None]

Deletes a dataset.

Arguments:

• existingDataset Dataset - Dataset to be deleted. return: Dict[Any, Any]: Response.

getDatasetContent

def getDatasetContent(existingDataset: Dataset,
                      encoding="utf8") -> Union[Dict[Any, Any], None]

Downloads the content of an existing dataset from the Altergo platform.

Arguments:

• existingDataset Dataset - The Dataset object whose content needs to be downloaded.
• encoding str, optional - The character encoding used to decode the downloaded content. Defaults to "utf8". If None, the raw bytes are returned instead of a decoded string.

Returns:

Union[str, bytes, None]:

• A string containing the dataset content if encoding is provided.
• Raw bytes if encoding is None.
• None if the download fails.

Raises:

• UserWarning - Raised if there is an issue with the download process, such as a failed API call or an invalid dataset ID.

  Workflow:

  1. The method first retrieves the download URL for the dataset by calling the Altergo API using the dataset's ID.
  2. Once the URL is obtained, the content is downloaded from the URL.
  3. If an encoding is provided, the content is decoded into a string using the specified encoding (default is UTF-8). If no encoding is provided, the raw content (bytes) is returned.
  4. If the download fails, the method raises a warning with details of the failure.

Examples:

# To download the content of a dataset as a string:
dataset_content = getDatasetContent(existingDataset=myDataset)

# To download the content as raw bytes (e.g., for binary files):
dataset_content_bytes = getDatasetContent(existingDataset=myDataset, encoding=None)

Notes:

• Ensure that the existingDataset object has a valid id before calling this method.
• The content can be returned as either a decoded string or raw bytes, depending on the encoding parameter.

createFunction

def createFunction(newFunction: Function, filePath: str, fileName: str)

Creates a new function in the program manager and uploads an associated file.

Arguments:

• newFunction Function - The Function object to be created, containing metadata such as name, source, and type.
• filePath str - The full file path of the function file to be uploaded.
• fileName str - The name of the file to be uploaded.

Returns:

• Function - The created Function object if the operation is successful.

Raises:

• RuntimeWarning - If the function creation or file upload fails, a warning is raised with details of the failure.

Examples:

new_function = Function(name="MyFunction", source="source_code", type="Python")
created_function = createFunction(newFunction=new_function, filePath="/path/to/function.py", fileName="function.py")

getFunctionById

def getFunctionById(
        functionId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Function, None]

Retrieves a function by its ID.

Arguments:

• functionId str - The unique ID of the function to retrieve.
• populate Optional[Dict[Any, Any]] - A dictionary specifying fields to be populated in the result. Defaults to None.
• projection Optional[Dict[Any, Any]] - A dictionary containing projection keys with a comma-separated string of fields to return. Defaults to None.

Returns:

• Function - The retrieved Function object if found, otherwise None.

Raises:

• UserWarning - If the function retrieval fails, a warning is raised with the failure details.

Examples:

function = getFunctionById(functionId="12345")

getFunctions

def getFunctions(filterBy: Optional[Dict[Any, Any]] = None,
                 populate: Optional[Dict[Any, Any]] = None,
                 projection: Optional[Dict[Any, Any]] = None,
                 skipTo: int = 0,
                 limitBy: int = 100) -> List[Function]

Retrieves all functions that match the specified filter criteria.

Arguments:

• filterBy Optional[Dict[Any, Any]] - A dictionary specifying filter criteria. Defaults to None.
• populate Optional[Dict[Any, Any]] - A dictionary specifying fields to be populated in the result. Defaults to None.
• projection Optional[Dict[Any, Any]] - A dictionary containing projection keys with a comma-separated string of fields to return. Defaults to None.
• skipTo int, optional - The number of records to skip before returning the results. Defaults to 0.
• limitBy int, optional - The maximum number of functions to return. Defaults to 100.

Returns:

• List[Function] - A list of Function objects that match the criteria.

Raises:

• UserWarning - If the function retrieval fails for any of the entries, a warning is raised.

Examples:

functions = getFunctions(filterBy={"type": "Python"}, skipTo=10, limit

updateFunction

def updateFunction(existingFunction: Function, filePath: str, fileName: str)

Updates an existing function and uploads an updated associated file.

Arguments:

• existingFunction Function - The Function object to be updated, containing updated metadata such as name, source, and type.
• filePath str - The full file path of the updated function file.
• fileName str - The name of the updated function file to be uploaded.

Returns:

• Function - The updated Function object if the operation is successful.

Raises:

• RuntimeWarning - If the function update or file upload fails, a warning is raised with details of the failure.

Examples:

updated_function = updateFunction(existingFunction=my_function, filePath="/path/to/updated_function.py", fileName="updated_function.py")

deleteFunction

def deleteFunction(existingFunction: Function) -> Union[Dict[Any, Any], None]

Deletes an existing function.

Arguments:

• existingFunction Function - The Function object to be deleted.

Returns:

Dict[Any, Any]: A dictionary containing the result of the delete operation if successful, otherwise None.

Raises:

• UserWarning - If the function deletion fails, a warning is raised with details of the failure.

Examples:

delete_result = deleteFunction(existingFunction=my_function)

createProgram

def createProgram(newProgram: Program) -> Union[Program, None]

Creates a new program in the system.

Arguments:

• newProgram Program - The Program object to be created, containing metadata such as name, description, and functions.

Returns:

• Program - The created Program object if the operation is successful, otherwise None.

Raises:

• UserWarning - If the program creation fails, a warning is raised with details of the failure.

Examples:

new_program = Program(name="My Program", description="A new program for testing.")
created_program = createProgram(newProgram=new_program)

getProgramById

def getProgramById(
        programId: str,
        populate: Optional[Dict[Any, Any]] = None,
        projection: Optional[Dict[Any, Any]] = None) -> Union[Program, None]

Gets a program by id.

Arguments:

• programId str - Program id.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned return:
• Program - Program object.

getPrograms

def getPrograms(filterBy: Optional[Dict[Any, Any]] = None,
                populate: Optional[Dict[Any, Any]] = None,
                projection: Optional[Dict[Any, Any]] = None,
                skipTo: int = 0,
                limitBy: int = 100) -> List[Program]

Gets all programs matching the filter criteria. The size of the return is limited by skipTo and limitBy.

Arguments:

• filterBy Optional[Dict[Any, Any]] - Filter criteria.
• populate Optional[Dict[Any, Any]] - Array containing the fields that need to be populated.
• projection Optional[Dict[Any, Any]] - dict containing projection key with value a comma seperated string containing the fields that needs to returned
• skipTo int - Skip to.
• limitBy int - Limit by. return:
• List[Program] - List of programs.

updateProgram

def updateProgram(updatedProgram: Program) -> Union[Program, None]

Updates an existing program in the system.

Arguments:

• updatedProgram Program - The Program object to be updated, containing the modified fields such as name, description, and functions.

Returns:

Union[Program, None]: The updated Program object if the operation is successful, otherwise None.

Raises:

• UserWarning - If the program update fails, a warning is raised with details of the failure.

Examples:

updated_program = updateProgram(updatedProgram=my_updated_program)

runProgram

def runProgram(programId: str, parameters: dict) -> dict

Runs a program by its ID with the given parameters.

Arguments:

• programId str - The unique ID of the program to be executed.
• parameters dict - A dictionary containing the parameters required for the program's execution.

Returns:

• dict - The result of the program execution, including any output or status information.

Raises:

• Exception - If the program execution fails, an exception is raised with the error message.

Examples:

result = runProgram(programId="12345", parameters={"param1": "value1", "param2": "value2"})

runProgramByIdAndGetOutputContent

def runProgramByIdAndGetOutputContent(programId: str,
                                      parameters: dict) -> dict

Runs a program by its ID, waits for the task to complete, and retrieves the output content.

Arguments:

• programId str - The unique ID of the program to be executed.
• parameters dict - A dictionary containing the parameters required for the program's execution.

Returns:

• dict - A dictionary containing:
  • "task": The task object associated with the program execution.
  • "outputFiles": A list of dictionaries containing the file name and content of the program output files.

Raises:

• Exception - If the program execution or task completion fails, an exception is raised with the error message.

Examples:

output = runProgramByIdAndGetOutputContent(programId="12345", parameters={"param1": "value1", "param2": "value2"})
for file in output["outputFiles"]:
    print(f"File: {file['fileName']}, Content: {file['content']}")

runProgramByNameAndGetOutputContent

def runProgramByNameAndGetOutputContent(programName: str,
                                        parameters: dict) -> dict

Runs a program by its name, waits for the task to complete, and retrieves the output content.

Arguments:

• programName str - The name of the program to be executed.
• parameters dict - A dictionary containing the parameters required for the program's execution.

Returns:

• dict - A dictionary containing the output content of the program's execution.

Raises:

• Exception - If the program is not found or if execution fails.

Examples:

output = runProgramByNameAndGetOutputContent(programName="MyProgram", parameters={"param1": "value1"})

getProgramOutputContent

def getProgramOutputContent(programId: str, programOutputId: str,
                            programOutputFileName: str) -> dict

Retrieves the content of a program's output file.

Arguments:

• programId str - The unique ID of the program.
• programOutputId str - The ID of the program's output.
• programOutputFileName str - The name of the output file to retrieve.

Returns:

• dict - The content of the program's output file.

Raises:

• Exception - If the file retrieval fails.

Examples:

output_content = getProgramOutputContent(programId="12345", programOutputId="67890", programOutputFileName="output.txt")

deleteProgram

def deleteProgram(existingProgram: Program) -> Union[Dict[Any, Any], None]

Deletes an existing program.

Arguments:

• existingProgram Program - The Program object to be deleted.

Returns:

Union[Dict[Any, Any], None]: A dictionary containing the result of the delete operation if successful, otherwise None.

Raises:

• UserWarning - If the program deletion fails, a warning is raised.

Examples:

delete_result = deleteProgram(existingProgram=my_program)

createPanel

def createPanel(newPanel: Panel) -> Union[Panel, None]

Creates a new panel in the system.

Arguments:

• newPanel Panel - The Panel object to be created, containing panel details such as name and description.

Returns:

Union[Panel, None]: The created Panel object if the operation is successful, otherwise None.

Raises:

• UserWarning - If the panel creation fails, a warning is raised.

Examples:

created_panel = createPanel(newPanel=my_new_panel)

getPanelById

def getPanelById(panelId: str) -> Union[Panel, None]

Retrieves a panel by its ID.

Arguments:

• panelId str - The unique ID of the panel to retrieve.

Returns:

Union[Panel, None]: The retrieved Panel object if found, otherwise None.

Raises:

• UserWarning - If the panel retrieval fails, a warning is raised.

Examples:

panel = getPanelById(panelId="12345")

getPanels

def getPanels(filterBy: Optional[Dict[Any, Any]] = None,
              skipTo: int = 0,
              limitBy: int = 100) -> List[Panel]

Retrieves all panels that match the specified filter criteria.

Arguments:

• filterBy Optional[Dict[Any, Any]] - A dictionary specifying filter criteria. Defaults to None.
• skipTo int, optional - The number of records to skip before returning the results. Defaults to 0.
• limitBy int, optional - The maximum number of panels to return. Defaults to 100.

Returns:

• List[Panel] - A list of Panel objects that match the criteria.

Raises:

• UserWarning - If the panel retrieval fails for any entry, a warning is raised.

Examples:

panels = getPanels(filterBy={"type": "Dashboard"}, skipTo=10, limitBy=20)

updatePanel

def updatePanel(updatedPanel: Panel) -> Union[Panel, None]

Updates an existing panel in the system.

Arguments:

• updatedPanel Panel - The Panel object to be updated, containing the modified fields such as name and description.

Returns:

Union[Panel, None]: The updated Panel object if the operation is successful, otherwise None.

Raises:

• UserWarning - If the panel update fails, a warning is raised.

Examples:

updated_panel = updatePanel(updatedPanel=my_updated_panel)

deletePanel

def deletePanel(existingPanel: Panel) -> Union[Dict[Any, Any], None]

Deletes an existing panel.

Arguments:

• existingPanel Panel - The Panel object to be deleted.

Returns:

Union[Dict[Any, Any], None]: A dictionary containing the result of the delete operation if successful, otherwise None.

Raises:

• UserWarning - If the panel deletion fails, a warning is raised.

Examples:

delete_result = deletePanel(existingPanel=my_panel)

createSpecificAlert

def createSpecificAlert(newAlert) -> Union[None, None]

Creates a new specific alert in the system.

Arguments:

• newAlert dict - A dictionary containing the details of the alert to be created.

Returns:

Union[None, None]: The result of the alert creation.

Raises:

• UserWarning - If the alert creation fails, a warning is raised.

Examples:

result = createSpecificAlert(newAlert={"name": "Temperature Alert", "condition": "temp > 50"})

regenerateAlerts

def regenerateAlerts(asset,
                     startDate: datetime,
                     endDate: datetime,
                     alertRules: List[str] = None,
                     sendNotification: bool = False) -> Union[None, None]

Regenerates alerts for a given asset within a specified date range.

Arguments:

• asset Asset - The asset for which alerts will be regenerated.
• startDate datetime - The start date of the regeneration period.
• endDate datetime - The end date of the regeneration period.
• alertRules List[str], optional - A list of alert rule IDs to apply. Defaults to None.
• sendNotification bool, optional - If True, sends a notification when alerts are regenerated. Defaults to False.

Returns:

Union[None, None]: The result of the alert regeneration process.

Raises:

• UserWarning - If the alert regeneration fails, a warning is raised.

Examples:

regenerateAlerts(asset=my_asset, startDate=start_time, endDate=end_time, alertRules=["rule1", "rule2"], sendNotification=True)

getAlerts

def getAlerts() -> Union[None, None]

Retrieves all alerts from the system.

Returns:

Union[None, None]: A dictionary containing all the alerts if successful, otherwise None.

Raises:

• UserWarning - If the alert retrieval fails, a warning is raised.

Examples:

alerts = getAlerts()

createAlertRule

def createAlertRule(alertRule) -> Union[None, None]

Creates a new alert rule in the system.

Arguments:

• alertRule dict - A dictionary containing the details of the alert rule to be created.

Returns:

Union[None, None]: The result of the alert rule creation.

Raises:

• UserWarning - If the alert rule creation fails, a warning is raised.

Examples:

createAlertRule(alertRule={"name": "Temperature Exceeded", "condition": "temp > 50"})

getAlertRules

def getAlertRules() -> Union[None, None]

Retrieves all alert rules from the system.

Returns:

Union[None, None]: A dictionary containing all the alert rules if successful, otherwise None.

Raises:

• UserWarning - If the alert rule retrieval fails, a warning is raised.

Examples:

alert_rules = getAlertRules()

deleteAlertRuleById

def deleteAlertRuleById(alertRuleId) -> Union[None, None]

Deletes an alert rule by its ID.

Arguments:

• alertRuleId str - The unique ID of the alert rule to be deleted.

Returns:

Union[None, None]: The result of the delete operation if successful, otherwise None.

Raises:

• UserWarning - If the alert rule deletion fails, a warning is raised.

Examples:

delete_result = deleteAlertRuleById(alertRuleId="12345")

deleteAlertById

def deleteAlertById(alertId) -> Union[None, None]

Deletes an alert by its ID.

Arguments:

• alertId str - The unique ID of the alert to be deleted.

Returns:

Union[None, None]: The result of the delete operation if successful, otherwise None.

Raises:

• UserWarning - If the alert deletion fails, a warning is raised.

Examples:

delete_result = deleteAlertById(alertId="54321")

deleteAllAlertsFromAsset

def deleteAllAlertsFromAsset(asset, startDate, endDate) -> Union[None, None]

Deletes all alerts from an asset within a specified date range.

Arguments:

• asset Asset - The asset from which alerts will be deleted.
• startDate datetime - The start date of the alert deletion range.
• endDate datetime - The end date of the alert deletion range.

Returns:

Union[None, None]: The result of the delete operation if successful, otherwise None.

Raises:

• UserWarning - If the alert deletion fails, a warning is raised.

Examples:

deleteAllAlertsFromAsset(asset=my_asset, startDate=start_time, endDate=end_time)

deleteModelFromBlueprint

def deleteModelFromBlueprint(blueprintId, modelId) -> Union[None, None]

Deletes a model from a blueprint by its ID.

Arguments:

• blueprintId str - The unique ID of the blueprint.
• modelId str - The unique ID of the model to be deleted.

Returns:

Union[None, None]: The result of the delete operation if successful, otherwise None.

Raises:

• UserWarning - If the model deletion fails, a warning is raised.

Examples:

deleteModelFromBlueprint(blueprintId="12345", modelId="67890")

updateDashboardPanelsDatasource

def updateDashboardPanelsDatasource(dashboard: Dashboard,
                                    update,
                                    panelNames=None) -> Union[None, None]

Updates the datasource for all panels in a dashboard.

Arguments:

• dashboard Dashboard - The dashboard whose panels' datasources will be updated.
• update dict - A dictionary containing the key-value pairs to update in the datasource.
• panelNames List[str], optional - A list of panel names to update. Defaults to None (all panels will be updated).

Returns:

Union[None, None]: The result of the update operation if successful, otherwise None.

Raises:

• ValueError - If any key in the update dictionary is not found in the panel's datasource.

Examples:

updateDashboardPanelsDatasource(dashboard=my_dashboard, update={"key1": "new_value})