When I investigated one case of translating model to SVF (format for Forge Viewer), I found an old script of Python that we have not migrated. So I took time to make it working with latest version of Forge services. It shows the whole workflow from getting token, creating bucket, uploading object to requesting translating the object. I think it would be useful to some programmers of Python, so forked to my Github:
https://github.com/xiaodongliang/forge.workflow-python-sample
The code is enclosed here to be more searchable on internet.
#!/usr/bin/env python
# pyadva.py - demonstrate Autodesk View and Data API authorisation and translation process in Python
#
# Copyright (C) 2017 by Forge Partnership, Autodesk Inc.
#
import base64, json, md5, os.path, requests, shutil, time
from optparse import OptionParser
_version = '2.0'
BASE_URL = 'https://developer.api.autodesk.com/'
BUCKET_KEY = '<your bucket name>'
_file_missing_prompt = "Error: specified %s file '%s' does not exist.\n"
def parse_credentials(filename):
"Parse credentials from given text file."
f = open( filename )
lines = f.readlines()
f.close()
credentials = []
for line in lines:
i = line.find('#')
if -1 < i: line = line[0:i]
i = line.find(':')
if -1 < i: line = line[i+1:]
line = line.strip()
if 0 < len(line):
print line
line = line.strip("\"'")
credentials.append(line)
if 2 != len(credentials):
raise "Invalid credentials: expected two entries, consumer key and secret;\nread %s lines, %s after stripping comments." % (len(lines),len(credentials))
credentials = null
return credentials
def main():
"Drive Autodesk 3D viewer authorisation and translation process."
global BUCKET_KEY
progname = 'pylmv'
usage = 'usage: %s [options]' % progname
parser = OptionParser( usage, version = progname + ' ' + _version )
parser.add_option( '-b', '--bucketskip', action='store_true', dest='bucketskip', help = 'skip bucket creation' )
parser.add_option( '-c', '--credentials', dest='credentials_filename', help = 'credentials filename', metavar="FILE", default='credentials.txt' )
parser.add_option( '-m', '--model', dest='model_filename', help = 'model filename', metavar="FILE", default='samples/mytestmodel.rvt' )
parser.add_option( '-q', '--quiet', dest='quiet', action='store_true', default=False, help = 'reduce verbosity' )
parser.add_option( '-u', '--urn', dest='urn', help = 'specify urn of already uploaded model file', default='' )
(options, args) = parser.parse_args()
print options
print args
verbose = not options.quiet
if 1 < len( args ):
raise SystemExit(parser.print_help() or 1)
model_filepath = options.model_filename
if not model_filepath:
print 'Please specify a model file to process.'
raise SystemExit(parser.print_help() or 2)
if not os.path.exists( model_filepath ):
print _file_missing_prompt % ('model', model_filepath)
raise SystemExit(parser.print_help() or 3)
# Step 1: Register and create application, retrieve credentials
if not os.path.exists( options.credentials_filename ):
print _file_missing_prompt % ('credentials', options.credentials_filename)
raise SystemExit(parser.print_help() or 4)
credentials = parse_credentials(options.credentials_filename)
if not credentials:
print "Invalid credentials specified in '%s'." % options.credentials_filename
raise SystemExit(parser.print_help() or 5)
consumer_key = credentials[0]
consumer_secret = credentials[1]
BUCKET_KEY =(BUCKET_KEY + '-' + consumer_key).lower ()
# Step 2: Get your access token
# curl -k \
# --data "client_id=<your client id>&client_secret=<your client secret>&grant_type=client_credentials&scope=data:read data:write bucket:create bucket:read" \
# https://developer.api.autodesk.com/authentication/v1/authenticate \
# --header "Content-Type: application/x-www-form-urlencoded"
url = BASE_URL + 'authentication/v1/authenticate'
data = {
'client_id' : consumer_key,
'client_secret' : consumer_secret,
'grant_type' : 'client_credentials',
'scope': 'data:read data:write bucket:create bucket:read'
}
headers = {
'Content-Type' : 'application/x-www-form-urlencoded'
}
r = requests.post(url, data=data, headers=headers)
content = eval(r.content)
if verbose or 200 != r.status_code:
print r.status_code
print r.headers['content-type']
print type(r.content)
print content
# -- example results --
# 200
# application/json
# {"token_type":"Bearer","expires_in":1799,"access_token":"ESzsFt7OZ90tSUBGh6JrPoBjpdEp"}
if 200 != r.status_code:
print "Authentication returned status code %s." % r.status_code
raise SystemExit(6)
access_token = content['access_token']
print 'Step 2 returns access token', access_token
# Step 3: Create a bucket
if not options.bucketskip:
# Check for prior existence:
# curl -k -X GET \
# -H "Authorization: Bearer lEaixuJ5wXby7Trk6Tb77g6Mi8IL" \
# https://developer.api.autodesk.com/oss/v2/buckets/<your bucket name>/details
url = BASE_URL + 'oss/v2/buckets/' + BUCKET_KEY + '/details'
headers = {
'Authorization' : 'Bearer ' + access_token
}
print 'Step 3: Check whether bucket exists'
print 'curl -k -X GET -H "Authorization: Bearer %s" %s' % (access_token, url)
print url
print headers
r = requests.get(url, headers=headers)
if verbose:
print r.status_code
print r.headers['content-type']
print r.content
# -- example results --
# 200
# application/json; charset=utf-8
# {
# "key":"jtbucket",
# "owner":"NjEasFuPL6WAsNctq3VCgXDnTUBGa858",
# "createDate":1404399358062,
# "permissions":[{"serviceId":"NjEasFuPL6WAsNctq3VCgXDnTUBGa858","access":"full"}],
# "policyKey":"transient"
# }
if 200 != r.status_code:
# Create a new bucket:
# curl -k \
# --header "Content-Type: application/json" --header "Authorization: Bearer fDqpZKYM7ExcC2694eQ1pwe8nwnW" \
# --data '{\"bucketKey\":\"<your bucket name>\",\"policyKey\":\"transient\"}' \
# https://developer.api.autodesk.com/oss/v2/buckets
url = BASE_URL + 'oss/v2/buckets'
data = {
'bucketKey' : BUCKET_KEY,
'policyKey' : 'transient'
}
headers = {
'Content-Type' : 'application/json',
'Authorization' : 'Bearer ' + access_token
}
print 'Step 3: Create a bucket'
print 'curl -k -H "Authorization: Bearer %s" -H "Content-Type:application/json" --data "{\\"bucketKey\\":\\"%s\\",\\"policyKey\\":\\"transient\\"}" %s' % (access_token, BUCKET_KEY, url)
print url
print json.dumps(data)
print headers
r = requests.post(url, data=json.dumps(data), headers=headers)
if verbose or 200 != r.status_code:
print r.status_code
print r.headers['content-type']
print r.content
# -- example results --
# The Python request call failed, but the curl
# command that it generated worked and produced
# the following result:
#
# {
# "key":"jtbucket",
# "owner":"NjEasFuPL6WAsNctq3VCgXDnTUBGa858",
# "createDate":1404399358062,
# "permissions":[{"serviceId":"NjEasFuPL6WAsNctq3VCgXDnTUBGa858","access":"full"}],
# "policyKey":"transient"
# }
if 200 != r.status_code:
print "Bucket creation returned status code %s." % r.status_code
raise SystemExit(7)
# Step 4: Upload a file
if options.urn:
urn = options.urn
else:
# curl -k \
# --header "Authorization: Bearer K16B98iaYNElzVheldlUAUqOoMRC" \
# -H "Content-Type:application/octet-stream" \
# --upload-file "<your object name>"
# -X PUT https://developer.api.autodesk.com/oss/v2/buckets/<your bucket name>/objects/<your object name>
filesize = os.path.getsize( model_filepath )
model_filename = os.path.basename( model_filepath ).replace(' ', '+')
#model_filename = model_filename.replace('.','_')
url = 'https://developer.api.autodesk.com/oss/v2/buckets/' + BUCKET_KEY + '/objects/' + model_filename
headers = {
'Content-Type' : 'application/octet-stream',
#'Content-Length' : str(filesize),
'Authorization' : 'Bearer ' + access_token,
#'Expect' : ''
}
print "Step 4: starting upload of model file '%s', %s bytes..." % (model_filename,filesize)
print 'curl -k -H "Authorization: Bearer %s" -H "Content-Type:application/octet-stream" -T "%s" -X PUT %s' % (access_token, model_filepath, url)
with open(model_filepath, 'rb') as f:
#files = { model_filename : f }
#r = requests.put(url, headers=headers, files=files)
#uploading does not aceept multi-parts objects.
#see http://docs.python-requests.org/en/latest/api/
# files: Dictionary of 'name': file-like-objects (or {'name': ('filename', fileobj)}) for multipart encoding upload.
# data: Dictionary, bytes, or file-like object to send in the body of the Request.
r = requests.put(url, headers=headers, data=f)
#with open(model_filepath, 'rb') as f:
# request = requests.put(url, headers=headers, data=f)
if verbose:
print r.status_code
print r.headers['content-type']
print r.content
# -- example results --
# The Python request call failed, but the curl
# command that it generated worked and produced
# the following result:
#
# {
# "bucket-key" : "<your bucket name>",
# "objects" : [ {
# "location" : "https://developer.api.autodesk.com/oss/v1/buckets/jtbucket/objects/two_columns_rvt",
# "size" : 4165632,
# "key" : "two_columns_rvt",
# "id" : "urn:adsk.objects:os.object:jtbucket/two_columns_rvt",
# "sha-1" : "cb15374248562743c5a99e0bdb0535f508a19848",
# "content-type" : "application/octet-stream"
# } ]
# }
content = eval(r.content)
#urn = content['objects'][0]['id']
urn = content['objectId']
print 'id:', urn
# import base64
# base64.b64encode("urn:adsk.objects:os.object:<your bucket name>/your object name")
# 'dXJuOmFkc2sub2JqZWN0czpvcy5vYmplY3Q6anRidWNrZXQvdHdvX2NvbHVtbnNfcnZ0'
urn = base64.b64encode(urn)
print 'urn:', urn
# Step 6: Register Data with the Viewing Services
# curl -k \
# -H "Content-Type: application/json" \
# -H "Authorization:Bearer 1f4bEhzvxJ9CMvMPSHD4gXO4SYEr" \
# -i -d "{\"urn\":\"dXJuOmFkc2sub2JqZWN0czpvcy5vYmplY3Q6bXlidWNrZXQvc2t5c2NwcjEuM2Rz\"}" \
# https://developer.api.autodesk.com/modelderivative/v2/designdata/job
url = BASE_URL + 'modelderivative/v2/designdata/job'
data = {
"input": {
"urn": urn
},
"output": {
"destination": {
"region": "us"
},
"formats": [
{
"type": "svf",
"views":["2d", "3d"]
}]
}
}
headers = {
'Content-Type' : 'application/json',
'Authorization' : 'Bearer ' + access_token
}
print 'Step 6: Request to translate the object'
print 'curl -k -H "Authorization: Bearer %s" -H "Content-Type:application/json" -i -d "{\\"urn\\":\\"%s\\"}" %s' % (access_token, urn, url)
print url
print json.dumps(data)
print headers
r = requests.post(url, data=json.dumps(data), headers=headers)
if verbose or 200 != r.status_code:
print r.status_code
print r.headers['content-type']
print r.content
# -- example results --
# The Python request call failed, but the curl
# command that it generated worked and produced
# the following result:
#
if 200 != r.status_code:
print "Register data returned status code %s." % r.status_code
raise SystemExit(9)
if __name__ == '__main__':
main()
The original author of this sample is my colleague Jeremy Tammik. Recently he also produced another blog on Python scripts.
https://forge.autodesk.com/cloud_and_mobile/2016/12/forge-formats-python-scripts.html
Xiaodong Liang has been with Autodesk since 2007, focusing on providing programming support, consulting, training and evangelism to external developers. He started his career China and now lives in Beijing, China. Xiaodong is currently a developer consultant in the team Developer Technical Services (DevTech), the worldwide team of API gurus providing technical services through the Autodesk Developer Network. He...