|
- # Copyright 2022 PCL
- #
- # Licensed under the Apache License, Version 2.0 (the "License");
- # you may not use this file except in compliance with the License.
- # You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- # ============================================================================
- import requests
- from loguru import logger
- import json
- import argparse
-
-
- class aisynergy():
- def __init__(self, key='pcl202212'):
- self.address = '192.168.206.136:30004'
- self.key = key
-
- def start_server(self, server_name, numRounds=100):
- url = f"http://{self.address}/ais_server/start"
- print(f"url:{url}")
- headers = {'Content-Type': 'application/json'}
- body = {"key": f"{self.key}",
- "serverName": f"{server_name}",
- "numRounds": f"{numRounds}"}
- try:
- res = requests.post(url=url, headers=headers, data=json.dumps(body))
- logger.info(f'start : {res.text}')
- except:
- logger.error('start error!!')
- return False
- return res.text
-
- def get_server_status(self, server_name):
- url = f"http://{self.address}/ais_server/status"
- headers = {'Content-Type': 'application/json'}
- body = {"key": f"{self.key}", "serverName": f"{server_name}"}
-
- res = requests.post(url=url, headers=headers, data=json.dumps(body))
- logger.info(f'status : {res.text}')
-
- return res.text
-
- def stop_server(self, server_name):
- url = f"http://{self.address}/ais_server/stop"
- headers = {'Content-Type': 'application/json'}
- body = {"key": f"{self.key}", "serverName": f"{server_name}"}
- res = requests.post(url=url, headers=headers, data=json.dumps(body))
- logger.info(f'stop : {res.text}')
- return res.text
-
- def get_server_list(self):
- url = f"http://{self.address}/ais_server/list"
- headers = {'Content-Type': 'application/json'}
- body = {"key": f"{self.key}"}
- res = requests.post(url=url, headers=headers, data=json.dumps(body))
- logger.info(f'server_list : {res.text}')
- try:
- server_list = json.loads(res.text)
- except:
- logger.error('stop server error!!')
- return False
- return server_list
-
- def stop_all_server(self):
- server_list = self.get_server_list()
- for server_name in server_list:
- self.stop_server(server_name)
- return True
-
- parser = argparse.ArgumentParser(description='mPanGu dataset sample')
- parser.add_argument('--uid', type=int, default=666000)
- args = parser.parse_args()
-
- if __name__ == '__main__':
- ais = aisynergy()
- ais.stop_server(str(args.uid))
- port = ais.start_server(str(args.uid))
|