# portal_api.py
import requests
import json
import os
import hashlib
PORTAL_TOKEN = os.environ.get('EI_PORTAL_TOKEN')
PORTAL_ID = os.environ.get('EI_PORTAL_ID')
JWT_TOKEN = os.environ.get('EI_JWT_TOKEN')
if not PORTAL_TOKEN:
print('Missing EI_PORTAL_TOKEN environmental variable.')
print('Go to a portal, and copy the part after "?token=" .')
print('Then run:')
print(' export EI_PORTAL_TOKEN=ec61e...')
print('')
print('You can add the line above to your ~/.bashrc or ~/.zshrc file to automatically load the token in the future.')
exit(1)
if not PORTAL_ID:
print('Missing EI_PORTAL_ID environmental variable.')
print('Go to a portal, open the browser console, and look for "portalId" in the "Hello world from Edge Impulse" object to find it.')
print('Then run:')
print(' export EI_PORTAL_ID=122')
print('')
print('You can add the line above to your ~/.bashrc or ~/.zshrc file to automatically load the token in the future.')
exit(1)
if not JWT_TOKEN:
print('WARN: Missing EI_JWT_TOKEN environmental variable, you will only have write-only access to the portal')
print('Run `python3 get_jwt_token.py` for instructions on how to set the token')
print('(this requires access to the organization that owns the portal)')
def get_file_hash(path):
with open(path, 'rb') as f:
return hashlib.md5(f.read()).hexdigest()
def create_upload_link(file_name_in_portal, path):
url = "https://studio.edgeimpulse.com/v1/api/portals/" + PORTAL_ID + "/upload-link"
payload = json.dumps({
'fileName': file_name_in_portal,
"fileSize": os.path.getsize(path),
"fileHash": get_file_hash(path)
})
headers = {
'accept': "application/json",
'content-type': "application/json",
'x-token': PORTAL_TOKEN
}
response = requests.request("POST", url, data=payload, headers=headers)
if (response.status_code != 200):
raise Exception('status code was not 200, but ' + str(response.status_code) + ' - ' + response.text)
j = response.json()
if (not j['success']):
raise Exception('api request did not succeed ' + str(response.status_code) + ' - ' + response.text)
return j['url']
def upload_file_to_s3(signed_url, path):
with open(path, 'rb') as f:
response = requests.request("PUT", signed_url, data=f, headers={})
if (response.status_code != 200):
raise Exception('status code was not 200, but ' + str(response.status_code) + ' - ' + response.text)
def upload_file_to_portal(file_name_in_portal, path):
print('Uploading', file_name_in_portal + '...')
link = create_upload_link(file_name_in_portal, path)
upload_file_to_s3(link, path)
print('Uploading', file_name_in_portal, 'OK')
print('')
def create_download_link(file_name_in_portal):
url = "https://studio.edgeimpulse.com/v1/api/portals/" + PORTAL_ID + "/files/download"
payload = json.dumps({
'path': file_name_in_portal,
})
headers = {
'accept': "application/json",
'content-type': "application/json",
'x-token': PORTAL_TOKEN,
'x-jwt-token': JWT_TOKEN
}
response = requests.request("POST", url, data=payload, headers=headers)
if (response.status_code != 200):
raise Exception('status code was not 200, but ' + str(response.status_code) + ' - ' + response.text)
j = response.json()
if (not j['success']):
raise Exception('api request did not succeed ' + str(response.status_code) + ' - ' + response.text)
return j['url']
def download_file_from_s3(signed_url):
response = requests.request("GET", signed_url, headers={})
if (response.status_code != 200):
raise Exception('status code was not 200, but ' + str(response.status_code) + ' - ' + response.text)
return response.content
def download_file_from_portal(file_name_in_portal):
print('Downloading', file_name_in_portal + '...')
link = create_download_link(file_name_in_portal)
f = download_file_from_s3(link)
print('Downloading', file_name_in_portal, 'OK')
print('')
return f
def list_files_in_portal(prefix):
url = "https://studio.edgeimpulse.com/v1/api/portals/" + PORTAL_ID + "/files"
payload = json.dumps({
'prefix': prefix,
})
headers = {
'accept': "application/json",
'content-type': "application/json",
'x-token': PORTAL_TOKEN
}
response = requests.request("POST", url, data=payload, headers=headers)
if (response.status_code != 200):
raise Exception('status code was not 200, but ' + str(response.status_code) + ' - ' + response.text)
j = response.json()
if (not j['success']):
raise Exception('api request did not succeed ' + str(response.status_code) + ' - ' + response.text)
return j['files']
# this is how you upload files to a portal using the Edge Impulse API
# first argument is the path in the portal, second is the location of the file
upload_file_to_portal('test.jpg', '/Users/janjongboom/Downloads/test.jpg')
# uploading to a subdirectory
upload_file_to_portal('flowers/daisy.jpg', '/Users/janjongboom/Downloads/daisy-resized.jpg')
# listing files
print('files in root folder', list_files_in_portal(''))
print('files in "flowers/"', list_files_in_portal('flowers/'))
# downloading a file
if JWT_TOKEN:
buffer = download_file_from_portal('flowers/daisy.jpg')
with open('output.jpg', 'wb') as f:
f.write(buffer)
else:
print('Not downloading files, EI_JWT_TOKEN not set')
print('Done!')