Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
ck_dir=args.ck_dir
if args.ck_dir is None:
ck_dir='test-ck-master'
shutil.rmtree(ck_dir, ignore_errors=True)
r = subprocess.call('git clone https://github.com/ctuning/ck.git ' + ck_dir, shell=True)
if 0 < r:
print('Error: failed to clone CK!')
die(1)
sys.path.append(ck_dir)
import ck.kernel as ck
test_repo_name = 'ck-crowdnode-auto-tests'
test_repo_cid = test_repo_name + '::'
r = ck.access({'module_uoa': 'repo', 'data_uoa': test_repo_name, 'action': 'remove', 'force': 'yes', 'all': 'yes'})
r = ck.access({'remote': 'yes', 'module_uoa': 'repo', 'url': 'http://localhost:3333', 'quiet': 'yes', 'data_uoa': test_repo_name, 'action': 'add'})
if r['return']>0:
print('Unable to create test repo. ' + r.get('error', ''))
die(1)
tests_dir = os.path.join(script_dir, 'tests')
module_cfg = {
'secret_key': 'c4e239b4-8471-11e6-b24d-cbfef11692ca',
'platform': platform.system(),
'repo_name': test_repo_name,
'cid': test_repo_cid
}
def access_test_repo(param_dict, checkFail=True):
d = {'secretkey': module_cfg['secret_key'], 'cid': module_cfg['cid']}
def loadTestsFromModule(self, module, pattern=None):
module.ck = ck
module.cfg = module_cfg
module.access_test_repo = access_test_repo
module.files_dir = files_dir
return unittest.TestLoader.loadTestsFromModule(self, module, pattern)
import ck.kernel as ck
r=ck.access('list #* out=none')
if r['return']>0:
print 'Error: '+r['error']
exit(1)
lst=r['lst']
r=ck.load_json_file({'json_file':'convert2.json'})
if r['return']>0:
print 'Error: '+r['error']
exit(1)
d1=r['dict']
dtags={}
for q in lst:
m=q['module_uoa']
def do(i):
# Detect basic platform info.
ii={'action':'detect',
'module_uoa':'platform',
'out':'out'}
r=ck.access(ii)
if r['return']>0: return r
# Host and target OS params.
hos=r['host_os_uoa']
hosd=r['host_os_dict']
tos=r['os_uoa']
tosd=r['os_dict']
tdid=r['device_id']
# Fix cmd key here since it may be used to get extra run-time deps.
cmd_key='imagenet-val'
# Load TensorRT-test program meta and desc to check deps.
ii={'action':'load',
'module_uoa':'program',
udepl=r['deps']['library'].get('choices',[]) # All UOAs of env for TF lib
if len(udepl)==0:
return {'return':1, 'error':'no installed TensorFlow'}
cdeps['lib-tensorflow']['uoa']=udepl[0]
depm=copy.deepcopy(cdeps['model-and-weights'])
ii={'action':'resolve',
'module_uoa':'env',
'host_os':hos,
'target_os':tos,
'device_id':tdid,
'out':'con',
'deps':{'weights':copy.deepcopy(depm)},
'quiet':'yes'
}
r=ck.access(ii)
if r['return']>0: return r
udepm=r['deps']['weights'].get('choices',[])
if len(udepm)==0:
return {'return':1, 'error':'no installed Weights'}
cdeps['lib-tensorflow']['uoa']=udepl[0]
cdeps['model-and-weights']['uoa']=udepm[0]
ii={'action':'pipeline',
'prepare':'yes',
'dependencies':cdeps,
'module_uoa':'program',
'data_uoa':program,
'target_os':tos,
def get(self, resource, **kwargs):
# Return path to the APK for Workload resource.owner.name
if (resource.uiauto):
return None
# Remember if debugging/changing to use self.logger.debug as below
# self.logger.debug(dir(resource.owner))
r = ck.access({'module_uoa':'apk',
'action':'load',
'data_uoa':resource.owner.package})
path = r['path']
r_version = kwargs.get('version', getattr(resource, 'version', None))
alt_abi = None
if resource.owner.device.abi in ['arm64-v8','arm64']:
alt_abi = 'armeabi'
if resource.owner.device.abi in ['x86_64']:
alt_abi = 'x86'
for apk in r['dict']['apks']:
if r_version is not None and apk['version'] != r_version:
continue
if resource.owner.device.abi in apk['abis']:
return os.path.join(path,apk['apk_name'])
if alt_abi is not None and alt_abi in apk['abis']:
return os.path.join(path,apk['apk_name'])
# Process arguments.
if (arg.accuracy):
experiment_type = 'accuracy'
num_repetitions = 1
else:
experiment_type = 'performance'
num_repetitions = arg.repetitions
random_name = arg.random_name
share_platform = arg.share_platform
# Detect basic platform info.
ii={'action':'detect',
'module_uoa':'platform',
'out':'con'}
if share_platform: ii['exchange']='yes'
r=ck.access(ii)
if r['return']>0: return r
# Keep to prepare ReQuEST meta.
platform_dict=copy.deepcopy(r)
# Host and target OS params.
hos=r['host_os_uoa']
hosd=r['host_os_dict']
tos=r['os_uoa']
tosd=r['os_dict']
tdid=r['device_id']
# program='mobilenets-armcl-opencl'
program='image-classification-tf-py'
ii={'action':'show',
udepl=r['deps']['lib-tensorflow'].get('choices',[]) # All UOAs of env for Tensorflow libs.
if len(udepl)==0:
return {'return':1, 'error':'no installed Tensorflow libs'}
# Tensorflow models.
depm=copy.deepcopy(cdeps['model-and-weights'])
ii={'action':'resolve',
'module_uoa':'env',
'host_os':hos,
'target_os':tos,
'device_id':tdid,
'out':'con',
'deps':{'tensorflow_model':copy.deepcopy(depm)},
'quiet':'yes'
}
r=ck.access(ii)
if r['return']>0: return r
udepm=r['deps']['tensorflow_model'].get('choices',[]) # All UOAs of env for Tensorflow models.
if len(udepm)==0:
return {'return':1, 'error':'no installed TensorFlow models'}
# Load dataset path.
# FIXME: Does not have to be ImageNet val.
ii={'action':'show',
'module_uoa':'env',
'tags':'dataset,imagenet,val,raw'}
rx=ck.access(ii)
if len(rx['lst'])==0: return rx
# FIXME: Can also be 'CK_ENV_DATASET_IMAGE_DIR'.
img_dir=rx['lst'][0]['meta']['env']['CK_ENV_DATASET_IMAGENET_VAL']