dlmw/julius_baer_onboarding
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Player async function and integration with app.py

Browse Source
This commit is contained in:
luca
2025-04-12 12:59:36 +02:00
parent 321217db60
commit 92c86a48b6
2 changed files with 20 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
app.py
View File

@ -4,7 +4,7 @@ from flask import Flask
import config import config
from dto.requests import GameStartRequestDTO, GameDecisionRequestDTO from dto.requests import GameStartRequestDTO, GameDecisionRequestDTO
from services.julius_baer_api_client import JuliusBaerApiClient from services.player import Player
app = Flask(__name__) app = Flask(__name__)
@ -15,19 +15,9 @@ def hello_world(): # put application's code here
if __name__ == '__main__': if __name__ == '__main__':
jb_client = JuliusBaerApiClient()
# game_start_request = GameStartRequestDTO(player_name=config.API_TEAM) player = Player()
# res = jb_client.start_game(game_start_request) player.play_on_separate_thread()
#
# game_decision_request = GameDecisionRequestDTO(decision="Accept", client_id=res.client_id, session_id=res.session_id)
# decision_response = jb_client.send_decision(game_decision_request)
#
# while decision_response.status == "active":
# game_decision_request = GameDecisionRequestDTO(decision="Accept", client_id=res.client_id, session_id=res.session_id)
# decision_response = jb_client.send_decision(game_decision_request)
#
# if decision_response.status == "gameover":
# logging.info("Game over")
app.run() app.run()

19
services/player.py
View File

@ -1,4 +1,6 @@
import logging import logging
import threading
import time
from typing import Literal, Dict, Any from typing import Literal, Dict, Any
import config import config
from dto.requests import GameStartRequestDTO, GameDecisionRequestDTO from dto.requests import GameStartRequestDTO, GameDecisionRequestDTO
@ -9,18 +11,29 @@ class Player:
def __init__(self): def __init__(self):
self.client = JuliusBaerApiClient() self.client = JuliusBaerApiClient()
self._thread = None
def start(self): def start(self):
self.play() self.play()
def play_on_separate_thread(self):
if self._thread and self._thread.is_alive():
logging.warning('Game loop already running.')
return self._thread
self._thread = threading.Thread(target=self.play, daemon=True)
self._thread.start()
return self._thread
def play(self): def play(self):
print('playing')
payload = GameStartRequestDTO(player_name=config.API_TEAM) payload = GameStartRequestDTO(player_name=config.API_TEAM)
start_response = self.client.start_game(payload) start_response = self.client.start_game(payload)
logging.info(start_response) logging.info(start_response)
status = '' status = ''
decision = self.make_decision(start_response.client_data) decision = self.make_decision(start_response.client_data)
while status != 'gameover': while status not in ['gameover', 'complete']:
payload = GameDecisionRequestDTO( payload = GameDecisionRequestDTO(
decision=decision, decision=decision,
@ -29,10 +42,10 @@ class Player:
) )
decision_response = self.client.send_decision(payload) decision_response = self.client.send_decision(payload)
logging.info(decision_response) print(decision_response.status, decision_response.score)
status = decision_response.status status = decision_response.status
decision = self.make_decision(decision_response.client_data) decision = self.make_decision(decision_response.client_data)
time.sleep(1.5)
def make_decision(self, client_data: Dict[str, Any]) -> Literal["Accept", "Reject"]: def make_decision(self, client_data: Dict[str, Any]) -> Literal["Accept", "Reject"]:
# Do your magic! # Do your magic!