avista-digital-exchange-sdk 0.3.8

SDK to programmatically access the Avista Digital Exchange

Avista Digital Exchange SDK

This package allows you to access the Avista Digital Exchange and perform a subset of its features programmatically. All that you need is a personal authentication token that can be generated at energy.collaboratives.io.

PyPi listing

GitHub Repository

Note: This document contains HTML tags at each section header to support internal document references in the PyPi markdown viewer. View this readme_renderer issue for more information.

Table of Contents

Note: Internal document links are not working when viewing this document in PyPi because of this readme_renderer issue

• Avista Digital Exchange SDK
  • Table of Contents
  • Getting Started
  • AvistaDigitalExchange Functions
    • getUserInfo
    • dataStore
      • listDataStores
      • getDataStore
      • getDataStoreDirectory
      • getDataStoreFileMeta
      • downloadDataStoreFile
      • uploadFileToDataStore
      • deleteDataStoreFile
    • iot
      • getEndpoint
      • createEndpoint
      • createModel
      • listEndpointLastValues
      • queryByTimeRange
      • queryDataByTimeRange
      • publish
      • updateEndpointProperties
    • dataCapture
      • startCapture
      • stopCapture
      • publishData
      • subscribeToData
  • Types
    • User
    • Organization
    • DataStore
      • Methods
        • cd
        • ls
        • pwd
        • uploadFile
        • downloadFile
        • deleteFile
    • DataStoreDirectory
      • Methods
        • printContents
    • DataStoreFile
    • DigitalTwinModel
    • ModelProperty
    • ModelTelemetry
    • IotEndpoint
    • EndpointProperty
    • EndpointTelemetry
    • DxTypes.CaptureDataRecordInput
    • DxTypes.PublishCaptureDataResult
    • DxTypes.PublishCaptureDataSuccessfulRecord
    • DxTypes.PublishCaptureDataFailedRecord
    • DxTypes.StartCaptureResult
    • DxTypes.StopCaptureResult
  • Troubleshooting
  • Development
  • Deployment
  • Generating Python types and GraphQL Client from GraphQL Schema
  • Fix for py-graphql-mapper module
  • websocketclient / websocket-client submodule
  • Resources

Getting Started

1. Install the python package.

pip3 install avista-digital-exchange-sdk

2. Ensure you have the latest version.

pip3 install --upgrade avista-digital-exchange-sdk

3. Import the module in your python script.
4. Initialize the module with your authentication token. The authentication token can be a user authentication token, or an iot endpoint authentication token. If an endpoint token is used, only iot operations can be performed.

from avista_digital_exchange_sdk import AvistaDigitalExchange

digitalExchange = AvistaDigitalExchange("tokenvalue")

# Alternatively, instantiate with debug mode enabled
# digitalExchange = AvistaDigitalExchange("tokenvalue", True)

---

AvistaDigitalExchange Functions

getUserInfo

Retrieves your user information.

Parameters

None

Return Type

User

Example

user = digitalExchange.getUserInfo()

---

dataStore

listDataStores

Lists the data stores you own.

Parameters

None

Return Type

[DataStore]

Example

dataStores = digitalExchange.dataStores.listDataStores()

getDataStore

Gets a data store object. The object can be interacted with as a command line utility to navigate the data store. See DataStore.

Parameters

dataStoreId :  str, required
    The id of the data store.

Return Type

DataStore

Example

dataStore = digitalExchange.dataStores.getDataStore("dataStoreId.1234")

dataStore.ls()
dataStore.cd("dirname1")

getDataStoreDirectory

Gets the directory metadata and contents.

Parameters

dataStoreDirectoryId :  str, required
    The id of the directory.

Return Type

DataStoreDirectory

Example

dir = digitalExchange.dataStores.getDataStoreDirectory("dataStoreDirectoryId.1234")

getDataStoreFileMeta

Gets the metadata of a file.

Parameters

dataStoreFileId :  str, required
    The id of the file.

Return Type

DataStoreFile

Example

file = digitalExchange.dataStores.getDataStoreFileMeta("dataStoreFileId.1234")

downloadDataStoreFile

Downloads a copy of the file to the local file system.

Parameters

dataStoreFileId :  str, required
    The id of the file.
writeLocation :  str, required
    Path where the file should be written. If writeLocation is a directory, the filename will be its original name.

Return Type

DataStoreFile

Example

file = digitalExchange.dataStores.downloadDataStoreFile("dataStoreFileId.1234", "./")

uploadFileToDataStore

Uploads a local file to a data store.

Parameters

dataStoreId :  str, required
    The id of the data store you are writing to.
dataStoreDirectoryId :  str, required
    The id of the directory to place the file in.
localPath :  str, required
    The path of the file to be uploaded.
name :  str, optional
    A name for the file if naming different than the local file name.
description :  str, optional
    A description of the file.

Return Type

DataStoreFile

Example

file = digitalExchange.dataStores.uploadFileToDataStore("dataStoreId.1234", "dataStoreDirectoryId.1234", "./testFile.txt")

deleteDataStoreFile

Removes a file from the data store and deletes it from the Digital Exchange.

Parameters

dataStoreFileId :  str, required
    The id of the file to delete.

Return Type

DataStoreFile

Example

file = digitalExchange.dataStores.deleteDataStoreFile("dataStoreFileId.1234")

---

iot

getEndpoint

Gets information about the iot endpoint. Includes the digital twin data model.

Parameters

iotEndpointId :  str, required

Return Type

IotEndpoint

Example

iotEndpointId = "iotEndpointId.1234"
endpoint = digitalExchange.iot.getEndpoint(
     iotEndpointId)

for property in endpoint.properties:
    print(f"Property {property.name} has type {property.schemaType}.")
for telemetry in endpoint.telemetry:
    print(f"Telemetry attribute {telemetry.name} has type {telemetry.schemaType}.")

createEndpoint

Creates a new IoT Endpoint within an existing IoT Hub.

Parameters

iotHubId :  str, required
    The hub in which to create the endpoint in.
modelId :  str, required
    The digital twin model for this endpoint to use.
name :  str, required
    The name of the iot endpoint.
description :  str, optional

Return Type

IotEndpoint

Example

iotHubId = "iotHubId.1234"
modelId = "modelId.4321"
name = "Temperature Sensor"
description = "The sensor on top of the house"


endpoint = digitalExchange.iot.createEndpoint(
    iotHubId,
    modelId,
    name,
    description)

createModel

Creates a digital twin data model.

Parameters

name :  str, required
    Name should not contain spaces or special characters. Alphanumeric characters only.
description :  str, optional
properties :  [dict], required
    Array of dictionaries representing the properties to associate with the model. See https://learn.microsoft.com/en-us/azure/digital-twins/concepts-models#:~:text=the%20following%20fields%3A-,Property,-%2D%20Properties%20are%20data for more information on model properties.
    The expected dictionary keys are as follows:
        - name : str, required
            Name of the property
        - description : str, optional
            Description of the property
        - schemaType : str, required
            The variable type of the property. Valid values are "integer", "double", "string", "boolean", "dateTime", and "duration"
        - defaultValue : str, optional
            Wrap the value, regardless of schemaType, in a string
        - writable : bool, optional
            Determines if the proeprty value can be updated. Default value is False

telemetry :  [dict], required
    Array of dictionaries representing the telemetry variables to associate with the model. See https://learn.microsoft.com/en-us/azure/digital-twins/concepts-models#:~:text=and%20telemetry%20below.-,Telemetry,-%2D%20Telemetry%20fields%20represent for more information on model telemetry.
    The expected dictionary keys are as follows:
        - name : str, required
            Name of the telemetry
        - description : str, optional
            Description of the telemetry
        - schemaType : str, required
            The variable type of the telemetry. Valid values are "integer", "double", "string", "boolean", "dateTime", and "duration"

Return Type

DigitalTwinModel

Example

properties = [
    {
        "name": "Longitude",
        "description": "the last known longitude of the device",
        "schemaType": "double",
        "defaultValue": "-117.426048",
        "writable": True
    },
    {
        "name": "Latitude",
        "description": "the last known latitude of the device",
        "schemaType": "double",
        "defaultValue": "47.658779",
        "writable": True
    }
]

telemetry = [
    {
        "name": "Temperature",
        "description": "Temp from the sensor in celsius",
        "schemaType": "double"
    },
    {
        "name": "Humidity",
        "description": "Humidity from the sensor",
        "schemaType": "double"
    }
]

model = digitalExchange.iot.createModel(
    "MyFirstModel", 
    "A simple example digital twin model", 
    properties, 
    telemetry)

listEndpointLastValues

Queries for the last known values of the endpoint's attributes.

Parameters

iotEndpointId :  str, required
resultFileWriteLocation :  str, optional
    If you would like the results stored in a file, include a name or path here. 
    Results will be in JSON format.

Return Type

[dict]

[
    {
        'attributeName': 'name'
        'lastValue': 'value',
        'timestamp': 'time'
    }
]

Example

iotEndpointId = "iotEndpointId.1234"
result = digitalExchange.iot.listEndpointLastValues(
     iotEndpointId)

# Example iteration through result
for entry in result:
    attributeName = entry["attributeName"]
    lastValue = entry["lastValue"]
    timestamp = entry["timestamp"]

queryByTimeRange

Queries the endpoint data using attribute and time filters, and writes the results to a file.

You must provide a time range and the attributeNames that you wish to query. The method automatically iterates over all pagination tokens so the result will be compiled upon completion. When the compiled result is available, it will be written to the local file system.

Parameters

iotEndpointId :  str, required
    The id of the endpoint.
attributeNames :  [str], required
    The attributes to include in the query result.
startTime :  ISO8601 str, required
    The beginning of the time interval.
    Example: "2022-09-27T10:22:45.000Z"
endTime: ISO8601 str, required
    The end of the time interval.
    Example: "2022-10-27T10:22:45.000Z"
resultFileWriteLocation: str, required
    The location to store the result file on the local file system. If a directory is provided but not a file name, the file will be saved as result.csv in the specified directory.

Return Type

Bool indicating success or error

Example

result = digitalExchange.iot.queryByTimeRange(
    "iotEndpointId.1234", 
    ["SOC", "V", "V - Setpoint"], 
    "2022-09-25T20:13:04.465Z", 
    "2022-09-26T20:58:04.465Z", 
    "./")

queryDataByTimeRange

Queries the endpoint data using attribute and time filters, returning as a dictionary

You must provide a time range and the attributeNames that you wish to query. The method automatically iterates over all pagination tokens so the result will be compiled upon completion. When the compiled result is available.

Parameters

iotEndpointId :  str, required
    The id of the endpoint.
attributeNames :  [str], required
    The attributes to include in the query result.
startTime :  ISO8601 str, required
    The beginning of the time interval.
    Example: "2022-09-27T10:22:45.000Z"
endTime: ISO8601 str, required
    The end of the time interval.
    Example: "2022-10-27T10:22:45.000Z"

Return Type

List including the response values

[
    { 'iotEndpointId': 'iotEndpointId.1234', name: 'Iot Endpoint Name', timestamp: '2024-03-19 10:39:31.942000000', 'attribute1': 1.0, 'attribute2': 2.0'},
    { 'iotEndpointId': 'iotEndpointId.1234', name: 'Iot Endpoint Name', timestamp: '2024-03-19 10:40:31.942000000', 'attribute1': 1.1, 'attribute2': 2.1'}   
]

Example

result = digitalExchange.iot.queryDataByTimeRange(
    "iotEndpointId.1234", 
    ["SOC", "V", "V - Setpoint"], 
    "2022-09-25T20:13:04.465Z", 
    "2022-09-26T20:58:04.465Z")

publish

Publish telemetry values for an endpoint. One or more data records can be written at one time.

Parameters

iotEndpointId :  str, required
    The id of the endpoint.
data :  [dict], required
    Array of data records to write.  Each dict should contain the record's timestamp (defaults to current milliseconds), timeUnit ("MILLISECONDS", "SECONDS", "MICROSECONDS", "NANOSECONDS", "ISO8601") ("MILLISECONDS" by default), and a dicitonary of attribute name, value pairs. Example shown below.
    ISO8601 timestamps should match format yyyy-MM-ddTHH:mm:ss.SSSZ

Return Type

Dictionary shown below

{
    'recordsWritten': [
        {
            'timestamp': '1670008999', 
            'iotEndpointId': 'iotEndpointId.1234',
            'attributes': [
                {
                    'name': 'speed',
                    'value': '24.5'
                }
            ]
        }
    ],
    'recordsFailed': [
        {
            'record': {
                'timestamp': '1670007999',
                'iotEndpointId': 'iotEndpointId.1234',
                'attributes': [
                    {
                        'name': 'speed',
                        'value': '24.5'
                    }
                ]
            },
            'errors': [
                {
                    'errorType': 'DUPLICATE_RECORD',
                    'errorMessage': 'A record already exists for this endpoint at the given timestamp.'
                }
            ]
        }
    ]
}

Example

# Create the input data array
inputData = [
    {
        "timestamp": 1670008999,
        "timeUnit": "MILLISECONDS", # Optional - "MILLISECONDS" by default
        "attributes": {
            "speed": "40.2",
            "altitude": "2413.0"
        }
    }, {
        "timestamp": "2022-12-15T12:45:31.055Z",
        "timeUnit": "ISO8601",
        "attributes": {
            "speed": "40.2",
            "altitude": "2413.0"
        }
    }
]

# Write the records
result = digitalExchange.iot.publish('iotEndpointId.1234', inputData)

recordsWritten = result.recordsWritten
failedRecords = result.failedRecords

updateEndpointProperties

Update the properties (state variables) for an endpoint.

Parameters

iotEndpointId :  str, required
    The id of the endpoint.
properties :  dict, required
    The properties to update and the new values. See example below for expected dict structure.

Return Type

IotEndpoint

Example

properties = {
    'moving': False
}

digitalExchange.iot.updateEndpointProperties(
     "iotEndpointId.1234",
     properties)

---

dataCapture

startCapture

Commands a Data Capture to begin collecting data. Capture must be in 'Ready' state for startCapture to succeed.

Parameters

captureId :  str, required
    The id of the Data Capture.

Return Type

DxTypes.StartCaptureResult

Example

import asyncio
from avista_digital_exchange_sdk import AvistaDigitalExchange

# NOTE: The capture must be in 'READY' state to be started.
captureId = CAPTURE_ID
authenticationToken = AUTHENTICATION_TOKEN


async def main():
    # Create an instance of the AvistaDigitalExchange SDK
    # You may use a user authentication token or the authentication token of the Data Capture
    digitalExchange = AvistaDigitalExchange(authenticationToken)
    print("Instantiated AvistaDigitalExchange instance with authentication token")

    try:
        print("Calling dataCapture.startCapture")
        result = await digitalExchange.dataCapture.startCapture(
            captureId=captureId)
    except:
        print("dataCapture.startCapture failed")

if __name__ == "__main__":
    asyncio.run(main())

stopCapture

Commands a Data Capture to stop collecting data. Capture must be in 'CAPTURING' state for stopCapture to succeed.

Parameters

captureId :  str, required
    The id of the Data Capture.

Return Type

DxTypes.StopCaptureResult

Example

import asyncio
from avista_digital_exchange_sdk import AvistaDigitalExchange

# NOTE: The capture must be in 'CAPTURING' state to be stopped.
captureId = CAPTURE_ID
authenticationToken = AUTHENTICATION_TOKEN


async def main():
    # Create an instance of the AvistaDigitalExchange SDK
    # You may use a user authentication token or the authentication token of the Data Capture
    digitalExchange = AvistaDigitalExchange(authenticationToken)
    print("Instantiated AvistaDigitalExchange instance with authentication token")

    try:
        print("Calling dataCapture.stopCapture")
        result = await digitalExchange.dataCapture.stopCapture(
            captureId=captureId)
    except:
        print("dataCapture.stopCapture failed")

if __name__ == "__main__":
    asyncio.run(main())

publishData

Publish attribute values to an active Data Capture.

Parameters

captureId :  str, required
    The id of the Data Capture.
data :  [DxTypes.CaptureDataRecordInput], required
    Array of data records to write.  Each dict should contain the record's timestamp, timeUnit ("MILLISECONDS"), and a dicitonary of attributeId, value pairs. Example shown below.

Return Type

DxTypes.PublishCaptureDataResult

Example

import asyncio
import time
from avista_digital_exchange_sdk import AvistaDigitalExchange, DxTypes

authenticationToken = AUTHENTICATION_TOKEN_VALUE
captureId = YOUR_CAPTURE_ID

# Create an instance of the AvistaDigitalExchange SDK
# You may use a user authentication token or the authentication token of the Data Capture
digitalExchange = AvistaDigitalExchange(authenticationToken)
print("Instantiated AvistaDigitalExchange instance with authentication token")

# Specify the capture you are publishing data to.
# NOTE: The capture must be in Capturing state to accept data

def getCurrentMilliseconds():
    return int(f'{time.time() * 1000}'.split('.')[0])

async def publishDataExample():
    # Prepare the data to be published
    # Replace ATTRIBUTE_ID and ATTRIBUTE_VALUE
    attributeValueMap = {
        ATTRIBUTE_ID: ATTRIBUTE_VALUE,
        ATTRIBUTE_ID: ATTRIBUTE_VALUE,
        ATTRIBUTE_ID: ATTRIBUTE_VALUE,
    }

    data = [DxTypes.CaptureDataRecordInput(
            timestamp=f'{getCurrentMilliseconds()}',
            timeUnit="MILLISECONDS",
            attributeValues=attributeValueMap)]
    try:
        print("Calling dataCapture.publishData")
        result = await digitalExchange.dataCapture.publishData(
            captureId=captureId, data=data)
    except:
        print("dataCapture.publishData failed")

asyncio.run(publishDataExample())

subscribeToData

Listen for data being published to a Data Capture in realtime.

Parameters

captureId :  str, required
    The id of the Data Capture.

Return Type

AsyncIterator[DxTypes.PublishCaptureDataResult]

Example

import asyncio
from avista_digital_exchange_sdk import AvistaDigitalExchange

authenticationToken = AUTHENTICATION_TOKEN
captureId = CAPTURE_ID


async def main():
    # Create an instance of the AvistaDigitalExchange SDK
    # You may use a user authentication token or the authentication token of the Data Capture
    digitalExchange = AvistaDigitalExchange(authenticationToken)
    print("Instantiated AvistaDigitalExchange instance with authentication token")

    try:
        print("Calling dataCapture.subscribeToData")
        # The for-loop will run each time a data publish event is received.
        async for result in digitalExchange.dataCapture.subscribeToData(
                captureId=captureId):
            print("DataCapture.subscribeToData: Data received.")
            print("Do something with data...")
    except Exception as error:
        print(f"dataCapture.subscribeToData failed with error: {error}")
        raise error

if __name__ == "__main__":
    asyncio.run(main())

---

Types

User

Stores basic information about a Digital Exchange user.

Properties

userId : str
    The user's unique id number.
firstName : str
lastName : str
organization : Organization
    The organization that the user is a member of.

Organization

Stores basic information about a Digital Exchange member organization.

Properties

name : str
organizationId : str

DataStore

A Data Store Service object.

Properties

dataStoreId: str
name: str
description: str
ownerUserId: str
homeDirectoryId: str
    The dataStoreDirectoryId of the Data Store's home directory

Methods

cd

Change the working directory.

Parameters

path: str, required
  The path to the desired directory.

Example

ds = digitalExchange.getDataStore("dataStoreId.1234")
ds.cd("docs")

ls

List the contents of the working directory.

Parameters

path: str, optional
  The path to the desired directory.

Example

ds = digitalExchange.getDataStore("dataStoreId.1234")
ds.ls("./")

pwd

Prints the abosulte path of the working directory.

Parameters

none

Example

ds = digitalExchange.getDataStore("dataStoreId.1234")
ds.pwd()

uploadFile

Upload a file to the data store in the working directory.

Parameters

localFilePath: str, required
  The location of the file on your local file system.
name: str, optional
  What the file should be named in the data store.
description: string, optional
  A brief description of the file and it's contents.

Example

ds = digitalExchange.getDataStore("dataStoreId.1234")
ds.uploadFile("./exampleFile.json", "remoteFileName.json", "A file containing a basic json object.")

downloadFile

Download the specified file.

Parameters

filename: str, required
  Name of the remote file in the current directory.
writeLocation: str, optional
  The location to store the downloaded file. If a filename is not given it will be saved with it's remote name.

Example

ds = digitalExchange.getDataStore("dataStoreId.1234")
ds.downloadFile("remoteFileName.json", "./desiredLocalFileName.json")

deleteFile

Deletes the file from the working directory.

Parameters

filename: str, required
  The name of the file to delete.

Example

ds = digitalExchange.getDataStore("dataStoreId.1234")
ds.deleteFile("remoteFileName.json")

DataStoreDirectory

A directory in a Data Store file storage hierarchy.

Properties

dataStoreDirectoryId: str
dataStoreId: str
    The Data Store this directory belongs to.
name: str
    The name of the directory.
homeDirectory: Boolean
    Indicator of whether this directory is the home directory or a subdirectory.
parentDirectoryId: str
    The directory this directory is found within.
directories: [DataStoreDirectory]
    This directory's contents - directories.
files: [DataStoreFile]
    This directory's contents - files.

Methods

printContents

Lists all of the directory contents (directories and files), along with their ids.

Parameters

None

Example

dir = digitalExchange.getDataStoreDirectory("dataStoreDirectoryId.1234")
dir.printContents()

DataStoreFile

A file belonging to a Data Store.

Properties

dataStoreFileId: str
dataStoreId: str
    The Data Store this file belongs to.
dataStoreDirectoryId: str
    The directory that the file belongs to.
name: str
    The filename without the file extension.
description: str
fileExtension: str
storageSizeBytes: int
lastModified: str
contentType: str

DigitalTwinModel

Properties

modelId: str
    Id of the model
ownerUserId: str
    UserId of the owner of the model
displayName: str
    The name of the model
description: str
    Description of the model
properties: [ModelProperty]
    All properties associated with the model
telemetry: [ModelTelemetry]
    All telemetry associated with the model

ModelProperty

Data field that represents a state variable. (DTDL Property).

Properties

name: str
    Name of the property variable.
description: str
schemaType: str : "integer", "double", "string", "boolean", "dateTime", "duration"
    The type of the variable.
defaultValue: str
    The original value of the property.
writable: bool
    Indicates if the value of the property can be updated.

ModelTelemetry

Data field representing measurements or events. Values are stored in a time series database. (DTDL Telemetry).

Properties

name: str
    Name of the telmetry.
description: str
schemaType: str : "integer", "double", "string", "boolean", "dateTime", "duration"
    The type of the variable.

IotEndpoint

Properties

iotEndpointId: str
iotHubId: str
    The iot hub the endpoint belongs to.
modelId: str
    The model the endpoint uses.
name: str
    The name of the endpoint.
description: str
properties: [EndpointProperty]
    The properties and their values. Properties are associated through the digital twin model used by the endpoint.
telemetry: [EndpointTelemetry]
    The telemetry attributes of the endpoint. Telemetry variables are associated through the digital twin model used by the endpoint.

EndpointProperty

Data field that represents a state of the endpoint. No historical values. (DTDL Property).

Properties

name: str
    Name of the property.
description: str
schemaType: str : "integer", "double", "string", "boolean", "dateTime", "duration"
    The type of the variable.
writable: bool
    Indicates if the value of the property can be updated.
value: str
    The current value of the property. All variable types are wrapped in a string.
timestamp: str
    Timestamp (in milliseconds) that the current value was written.

EndpointTelemetry

Data field representing measurements or events. Values are stored in a time series database. Telemetry values must be queried for separately (queryByTimeRange, listEndpointLastValues). (DTDL Telemetry).

Properties

name: str
    Name of the telemetry.
description: str
schemaType: str : "integer", "double", "string", "boolean", "dateTime", "duration"
    The type of the variable.

DxTypes.CaptureDataRecordInput

A data record that is to be published to a Data Capture.

Properties

timestamp: str
    The timestamp of the data record
timeUnit: str
    The format of the timestamp
attributeValues: dict
    A dictionary containing the data to be written for the given timestmap. Key values should be attributeIds that map to the relevant attribute values. 

DxTypes.PublishCaptureDataResult

Result object for the DataCapture.publishData operation.

Properties

success: bool
    Indicates if the action was successful
captureId: str
    The captureId of the Data Capture published to.
recordsWritten: [DxTypes.PublishCaptureDataSuccessfulRecord] | None
    An array of input data records that were successfully written.
recordsFailed: [DxTypes.PublishCaptureDataFailedRecord] | None    
    An array of input data records that were not written.
error: str | None
    An error message if a problem was encountered.

DxTypes.PublishCaptureDataSuccessfulRecord

A record that was published and written to a Data Capture.

Properties

timestamp: str
    The timestamp of the data record
timeUnit: str
    The format of the timestamp
attributeId: str
attributeValue: str

DxTypes.PublishCaptureDataFailedRecord

A record that was published but not written to a Data Capture.

Properties

timestamp: str
    The timestamp of the data record
timeUnit: str
    The format of the timestamp
attributeId: str
attributeValue: str
message: str
    Explains why the data record failed.

DxTypes.StartCaptureResult

Result object for the DataCapture.startCapture operation.

Properties

success: bool
    Indicates if the action was successful
captureId: str
    The captureId of the Data Capture being started.
startedAt: str | None
    The timestamp at which the Data Capture was started (if successful).
error: str | None
    An error message if a problem was encountered.

DxTypes.StopCaptureResult

Result object for the DataCapture.stopCapture operation.

Properties

success: bool
    Indicates if the action was successful
captureId: str
    The captureId of the Data Capture being stopped.
stoppedAt: str | None
    The timestamp at which the Data Capture was stopped (if successful).
error: str | None
    An error message if a problem was encountered.

Troubleshooting

Development

Clone the repository with command git clone https://github.com/Avista-Digital-Innovation/avista-digital-exchange-sdk.git.

Use VS Code with the Python extension to utilize formatting and code completion.

Deployment related code is in the root directory and the package code is found in src/avista_digital_exchange_sdk.

Deployment

Follow the steps below to build and push the new package version to PyPi. (Python packaging reference used)

Steps

1. Update CHANGELOG.md with new release notes.
2. Update README.md if necessary.
3. PyPi deployment
   1. Update the package version in pyproject.toml. Follow this versioning method
   2. From the root directory of the repository:
      1. Run python3 -m build
      2. Run python3 -m twine upload dist/* to upload the new build to PyPi.
         1. Use username __token__
         2. Use your PyPi authentication token as the password.
   3. Test that the new version is available in pip by running pip3 install --upgrade avista-digital-exchange-sdk.
4. Push changes to git.
5. Merge changes to main.
6. Create a release branch from main with the name release/YYYY_MM_DD_vXX.XX.XX where XX.XX.XX is the new version number, and YYYY_MM_DD is the date the version was deployed.

Generating Python types and GraphQL Client from GraphQL Schema

See instructions in graphql_codegen/README.md

Fix for py-graphql-mapper module

The plugin was failing to assign the correct type for attributes of a schema type that were lists of other custom schema types (ie. IotEndpoint contains array of Properties and Telemetry).

I edited the function init_type(obj, fieldType, fieldName) in the file gql_init.py from commit bc0888b. There are conditions for different fieldTypes. I edited the condition if the field is a list. Changed ''' elif fieldType == list or (hasattr(fieldType, 'origin') and fieldType.origin == list): ''' to ''' elif fieldType == list or (hasattr(fieldType, 'origin') and fieldType.origin == list) or "gql_types.list" in str(fieldType): '''

websocketclient / websocket-client submodule

To mitigate an issue when using import websocket when both websocket and websocket-client packages are installed, we added a git submodule to reference websocket-client directly. Once the SDK moves away from the subscriptionClient.py usage, this can be removed.

Resources

1. Avista Digital Exchange
2. PyPi SDK project listing
3. GitHub project repository
4. Python package deployment tutorial
5. PyPi API Token