apis/examples_python/transport__engine_8py_source.html

가져오기io

가져오기os

가져오기sys

가져오기시간

가져오기zip파일


가져오기인증서

가져오기슬롯 사이트 추천engine

from게임가져오기GamsWorkspace

from슬롯 사이트 추천engine.api가져오기jobs_api


if__name__ =="__main__":

sys_dir = sys.argv[1]iflen(sys.argv) > 1그밖에 없음

work_dir = sys.argv[2]iflen(sys.argv) > 2그밖에 없음

ws = GamsWorkspace(system_directory=sys_dir,working_directory=work_dir)


모델 ="trnsport"

ws.gamslib(모델)


model_data_path = os.path.join(ws.working_directory, 모델 +".zip")


    와zipfile.Zip파일(model_data_path,"w", zipfile.ZIP_DEFLATED)as모델_데이터:

model_data.write(

os.path.join(ws.working_directory, 모델 +".gms"), 아크이름=모델 +".gms"

        )


stdout_filename ="log_stdout.txt"  # str | stdout을 캡처하는 파일 이름 (기본값은 'log_stdout.txt')

인수 = [

        "gdx=기본값"

]# 목록[str] | 슬롯 사이트 추천 호출에 전달될 인수(선택사항)


데이터 =없음  # 파일 | zip 형식의 데이터가 포함된 파일(선택 사항)

pf_file_name =없음  # str | zip에 있는 pf 파일의 이름(있는 경우)(선택 사항)

text_entries = []# 목록[str] | (선택사항)

stream_entries = []# 목록[str] | (선택사항)

inex_file =없음  # 파일 | 결과 zip 파일 내부 내용을 필터링하기 위한 선택적 파일(선택 사항)


구성 = 슬롯 사이트 추천engine.Configuration(

호스트=os.environ["ENGINE_URL"],

사용자 이름=os.environ["엔진_사용자"],

비밀번호=os.environ["엔진_비밀번호"],

ssl_ca_cert=certifi.where(),

    )

configuration.temp_folder_path = ws.working_directory

네임스페이스 = os.environ[

        "ENGINE_NAMESPACE"

]# 문자열 | 모델을 포함하는(또는 포함할) 네임스페이스


    # API 클라이언트 인스턴스로 컨텍스트를 입력

    함께슬롯 사이트 추천engine.ApiClient(구성)asapi_client:

        # API 클래스의 인스턴스 생성

job_api_instance = jobs_api.JobsApi(api_client)

        시도:

인쇄(f"모델 게시 중")


            와열기(model_data_path,"rb") asf:

file_content = f.read()


create_job_response = job_api_instance.create_job(

모델,

네임스페이스,

stdout_filename=stdout_filename,

model_data=(os.path.basename(model_data_path), file_content),

인수=인수,

            )


토큰 = create_job_response.token

인쇄(f"작업 토큰: 토큰")


        제외슬롯 사이트 추천engine.ApiExceptionase:

인쇄(f"JobsApi.create_job() 호출 시 예외: e\n")

sys.exit(1)


time_spent = 0.0

        그동안 사실입니다:

            시도해 보세요:

resp = job_api_instance.pop_job_logs(token)

print(resp.message, end="")

                ifresp.queue_finished:

                    휴식

time.sleep(0.5)

            제외슬롯 사이트 추천engine.ApiExceptionase:

                ife.status == 403:

인쇄("작업이 아직 대기열에 있습니다. 0.5초 동안 기다리십시오.")

time.sleep(0.5)

time_spent += 0.5

                    iftime_spent > 120:

인쇄(

                            "엔진 인스턴스가 바쁜 것 같습니다. 나중에 다시 시도하십시오."

                        )

sys.exit(1)

                그밖에:

                    올림e


        ifjob_api_instance.get_job(token).process_status != 0:

인쇄("작업이 성공적으로 종료되지 않았습니다.")


        시도해 보세요:

인쇄(f"모델 결과 가져오는 중: model")

zip_content = job_api_instance.get_job_zip(토큰)

            와zipfile.ZipFile(io.BytesIO(zip_content))으로zf:

gdx_file = zf.extract(모델 +".gdx", 경로=ws.working_directory)

        제외슬롯 사이트 추천engine.ApiExceptionase:

인쇄(f"JobsApi.get_job_zip() 호출 시 예외: e\n")

sys.exit(1)


        시도해 보세요:

            # 서버에서 결과 제거

job_api_instance.delete_job_zip(토큰)

        제외슬롯 사이트 추천engine.ApiExceptionase:

인쇄(f"JobsApi.delete_job_zip() 호출 시 예외: e\n")

sys.exit(1)


result_db = ws.add_database_from_gdx(

os.path.join(ws.working_directory, gdx_file)

        )


        forrec에결과_db["x"]:

인쇄(

f"x(rec.key(0),rec.key(1)): 레벨=rec.level 한계=rec.marginal"

            )