Upload Firmware with Python API

Learn how to upload a custom firmware package using the AVH Asynchronous API.

The AVH platform offers many different ways to interact with your virtual devices.

In this article, we demonstrate how to upload custom firmware and connect the the Raspberry Pi 4 board's console via a WebSocket using AVH Asynchronous Python API.

Overview of the Python Script

The script performs the following tasks:

Log in using your API token.
Find your project and save the VPN details.
Upload your custom firmware package.
Create a device using your uploaded image.
Wait for the device to boot.
Find the SSH connection details.
Stream the board's console in UTF-8 via a WebSocket.

Example of the script's output.

user1@comp1-2 ~ % python3 /Users/user1/api_example.py https://app.avh.corellium.com/api 7bf64295bfe46191f7bf.d735bb3128d2903c5b19f4d443f424ecb12cdce22ddbcf43aef3adf473e635702ef3f6923afff8ca548957f762d5874487fccbaeda72804e87f3f140b7a4eb31
Starting the script ...
Logging in ...
Logged in.
Finding the project ...
Found project 18370783-d77c-4382-bf20-61c024d4eabf.
Uploading the custom firmware package ...
Uploaded image cafcb47f-8e48-4ca8-8f00-df2b07793671.
Creating an instance named Ubuntu Desktop via Python API ...
Instance 71849293-74ff-4dbf-85f1-1aa5cb8ee41c is creating.
Waiting for the instance to boot up ...
After 30 seconds, the instance is creating.
After 60 seconds, the instance is creating.
After 90 seconds, the instance is creating.
After 120 seconds,j the instance is on.
Getting the connection details ...
To SSH, connect to the VPN then use `ssh ubuntu@10.11.0.13`.
Connecting to the WebSocket ...
{Begin stream of your Raspberry Pi 4 board's console output}
...
g/HQLlFVcJqnetdzkWXnPXF1GX7XSZqIazZ2/2E= root@ubuntu
-----END SSH HOST KEY KEYS-----
[   22.338550] cloud-init[1333]: Cloud-init v. 22.2-0ubuntu1~22.04.3 finished at Mon, 27 Jun 2022 18:29:02 +0000. Datasource DataSourceNoCloud [seed=/dev/mmcblk0p1][dsmode=net].  Up 22.33 seconds
Finished running main() due to AsyncIO timeout.
Finished the script.

:::Note This script will fail for packages larger than 2.00 gigabytes. For larger firmware packages, we recommend using the Create a New Image call from our REST API, which you can run with a cURL command from within Python. :::

Python Script

The Python script is as follows:

import asyncio
import time
import sys
import websockets
#from pprint import pprint # Optional for testing and development purposes.
import avh_api_async as AvhAPI
from avh_api_async.rest import ApiException as AvhAPIException
  
if len(sys.argv) < 3:
  print('Usage: %s <ApiEndpoint> <ApiToken>', sys.argv[0])
  exit(-1)
  
apiEndpoint = sys.argv[1]
apiToken = sys.argv[2]
  
instanceFlavor = 'rpi4b' # use flavor=rpi4b for the Raspberry Pi 4b board
ubuntuImageType = 'fwpackage' # use type=fwpackage for the Ubuntu Server firmware package
ubuntuImageEncoding = 'plain' # use encoding=plain
ubuntuImageVersion = '22.04.1' # the OS version that matches Info.plist
ubuntuImagePath = '/Users/user1/Desktop/' # directory that contains your firmware package
ubuntuImageFilename = 'rpi4b-ubuntu-server.zip' # name of your firmware package
ubuntuInstanceName = 'Ubuntu Server via Python API' # arbitraty name the instance
  
async def main():
  configuration = AvhAPI.Configuration(host = apiEndpoint)
  
  # Enter a context with an instance of the API client
  async with AvhAPI.ApiClient(configuration=configuration) as api_client:
    # Create an instance of the API class
    api_instance = AvhAPI.ArmApi(api_client)
  
    # Log in
    print('Logging in ...')
    try:
        token_response = await api_instance.v1_auth_login({
          "apiToken": apiToken
        })
        configuration.access_token = token_response.token
    except AvhAPIException as e:
        print('Exception when calling v1_auth_login: %s\n' % e)
        exit(1)
    print('Logged in.')
    
    # Get the project
    print('Finding the project ...')
    api_response = await api_instance.v1_get_projects()
    projectId = api_response[0].id
    print('Found project ' + projectId + '.')
  
    # Upload the image
    print('Uploading the custom firmware package ...')
    try:
      api_response = await api_instance.v1_create_image(_request_timeout=3600, type=ubuntuImageType, encoding=ubuntuImageEncoding, name=ubuntuImageFilename, project=projectId, file=ubuntuImagePath+ubuntuImageFilename)
      # The timeout for v1_create_image() is set to 1 hour.
      #pprint(api_response)
      ubuntuImageId = api_response.id
    except AvhAPIException as e:
      print('Exception when calling v1_create_image: %s\n' % e)
      exit(1)
    print('Uploaded image ' + ubuntuImageId + '.')
  
    # Create the instance
    print('Creating an instance named ' + ubuntuInstanceName + ' ...')
    try:
      api_response = await api_instance.v1_create_instance({
        'project': projectId,
        'name': ubuntuInstanceName,
        'flavor': instanceFlavor,
        'os': ubuntuImageVersion,
        'fwpackage': ubuntuImageId
      })
      #pprint(api_response)
      ubuntuInstanceId = api_response.id
      ubuntuInstanceState = api_response.state
    except AvhAPIException as e:
      print('Exception when calling v1_create_instance: %s\n' % e)
      exit(1)
    print('Instance ' + ubuntuInstanceId + ' is ' + ubuntuInstanceState + '.')
    
    # Wait for the instance to turn on
    count = 0
    secondsToPause = 30
    print('Waiting for the instance to boot up ...')
    while ubuntuInstanceState == 'creating':
      count+=1
      time.sleep(secondsToPause)
      api_response = await api_instance.v1_get_instance_state(ubuntuInstanceId)
      ubuntuInstanceState = api_response
      print('After ' + str(secondsToPause*count) + ' seconds, the instance is ' + ubuntuInstanceState + '.')
    
    # Get the SSH details
    print('Getting the connection details ...')
    try:
      api_response = await api_instance.v1_get_instance(ubuntuInstanceId)
      #pprint(api_response)
      wifiIP = api_response.wifi_ip
    except AvhAPIException as e:
      print('Exception when calling v1_get_instance: %s\n' % e)
      exit(1)
    print('To SSH, connect to the VPN then use `ssh ubuntu@' + wifiIP + '`.')
  
    # Connect to the console WebSocket
    print('Requesting the instance console WebSocket URL ...')
    try:
      api_response = await api_instance.v1_get_instance_console(ubuntuInstanceId)
      ubuntuConsoleWebSocketURL = api_response.url
    except AvhAPIException as e:
      print('Exception when calling v1_get_instance_console: %s\n' % e)
      exit(1)
    print('Connecting to the WebSocket ...')
    async with websockets.connect(ubuntuConsoleWebSocketURL) as ws:
      await handler(ws)
      await asyncio.Future()  # keep WebSocket connection open
  
async def handler(websocket):
  while True:
    message = await websocket.recv() # encoded in Unicode
    print(message.decode(encoding='UTF-8',errors='replace')) # decoded to UTF-8
  
if __name__ == '__main__':
  print('Starting the script ...')
  try:
    asyncio.run(asyncio.wait_for(main(), 7200)) # The timeout is 2 hours.
  except Exception as e:
    print('Finished running main() due to AsyncIO timeout.')
  print('Finished the script.')
  exit(0)

How to Run the Script

In this example, we will be uploading the rpi4b-ubuntu-server.zip custom firmware package from our Package Ubuntu Server Firmware for AVH article.

Follow the steps below to run the Python script. For more details, refer to the Python API Readme and the Reference for avh-api-async Methods documents.

Ensure you are running at least Python 3.4 and install the required packages including avh-api-async. Refer to the Python API Readme.md for more details.
```
pip3 install avh-api avh-api-async websockets python-dateutil aiohttp six urllib3
```
Save the Python script to a local file on your computer ending in .py. For our example, we save the script to the local user's desktop.
Update the hardcoded ubuntuImagePath and ubuntuImageFilename in the script to reference your local firmware package. The script will save a VPN configuration file to this path.
```
ubuntuImagePath = '/Users/user1/Desktop/'
ubuntuImageFilename = 'rpi4b-ubuntu-server.zip'
```

Run the script by referencing the AVH endpoint and your AVH API token, see Generating an AVH API Token for more information.

python3 /Users/user1/Destkop/api_example.py https://app.avh.corellium.com/api 7bf64295bfe46191f7bf.d735bb3128d2903c5b19f4d443f424ecb12cdce22ddbcf43aef3adf473e635702ef3f6923afff8ca548957f762d5874487fccbaeda72804e87f3f140b7a4eb31

Overview of the Python Script​

Python Script​

How to Run the Script​

Overview of the Python Script

Python Script

How to Run the Script