Skip to content
Snippets Groups Projects
test_server_configurations.py 2.24 KiB
Newer Older
import requests
import argparse
import base64
from glob import glob
from skimage.io import imread
    

if __name__ == '__main__':
    ap = argparse.ArgumentParser()
    
    ap.add_argument('-c', '--config', required=False, help='path to config folder')
    ap.add_argument('-s', '--server', required=False, help='server url address')
    ap.add_argument('-p', '--port', required=False, help='server port')
    ap.add_argument('-i', '--image', required=False, help='path to test image')

    args = vars(ap.parse_args())
    
    config_path = args['config']
    server_url = args['server']
    port = args['port'] if args['port'] is not None else '10123'
    img_path = args['image']
    
    config_url = server_url + ':' + port + '/set_config'

    config_path = config_path+'/' if config_path[-1]=='/' else config_path
    configs = [p.split('/')[-1] for p in glob(config_path+'*') if '_seneca' in p]
    print(configs)
    img = imread(img_path)

    for conf in configs:
        print('TEST: {}'.format(conf))
        endpoint = None
        if 'config_micronuclei_' in conf:
            endpoint = 'mni_classify' 
        elif 'config_round-nuclei_' in conf:
            endpoint = 'round_classify' 
        elif 'config_round-nuclei-combined_' in conf:
            endpoint = 'round_classify_combined_features' 
        else:
            print('skipping configuration: {}'.format(conf))
            continue
        res = requests.post(config_url, json={'filepath': config_path+conf})
        if res.status_code != 200:
            raise RuntimeError('setting configuration failed. {}'.format(conf))
        byte_content = base64.b64encode(img)  # to bytes
        string_content = byte_content.decode('utf-8')  # to string

        # request
        json_req = {'filename': 'test_image',
                    'img': string_content,
                    'width': img.shape[1],
                    'height': img.shape[0],
                    'dtype': img.dtype.str}
        
        pipeline_url = server_url + ':' + port + '/' + endpoint
        res = requests.post(pipeline_url, json=json_req)
        if res.status_code != 200:
            print('test failed at {}'.format(conf))
        else:
            print('TEST SUCCESSFUL')
    
    print('\nTest completed.')