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