sc2_base.py

"""
Copyright 2017 Google Inc.

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

https://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 asyncio
import websockets
from s2clientprotocol import sc2api_pb2 as sc_pb
from s2clientprotocol import common_pb2 as comm_pb
f = open("D.txt", "w+")
def makeGameRequest():
	req = sc_pb.Request()
	req.create_game.battlenet_map_name = "Ohana LE"
	req.create_game.disable_fog = True
	
	me = req.create_game.player_setup.add()
	me.type = sc_pb.Participant
	me.race = comm_pb.Protoss
	
	opponent = req.create_game.player_setup.add()
	opponent.type = sc_pb.Computer
	opponent.race = comm_pb.Terran
	opponent.difficulty = sc_pb.CheatInsane
	
	f.write(f"{req}")
	
	return req

def makeJoinGameRequest():
	req = sc_pb.Request()
	req.join_game.race = comm_pb.Protoss
	req.join_game.options.raw = True
	
	f.write(f"{req}")
	
	return req

def makeStepRequest():
	req = sc_pb.Request()
	req.step.count = 8
	
	f.write(f"{req}")
	
	return req

def makeObservationRequest():
	req = sc_pb.Request()
	req.observation.SetInParent()
	
	f.write(f"{req}")
	
	return req

def makeLeaveRequest():
	req = sc_pb.Request()
	req.leave_game.SetInParent()
	
	f.write(f"{req}")
	
	return req

def makeDataRequest():
	req = sc_pb.Request()
	req.data.SetInParent()
	
	f.write(f"{req}")
	
	return req

async def hello():
	async with websockets.connect('ws://127.0.0.1:5000/sc2api') as websocket:
		await websocket.send(makeGameRequest().SerializeToString())
		
		response = sc_pb.Response()
		response_bytes = await websocket.recv()
		response.ParseFromString(response_bytes)
		f.write(f"{response}")
		
		await websocket.send(makeJoinGameRequest().SerializeToString())
		
		response = sc_pb.Response()
		response_bytes = await websocket.recv()
		response.ParseFromString(response_bytes)
		f.write(f"{response}")
		
		still_going = True
		while still_going:
			await websocket.send(makeObservationRequest().SerializeToString())
		
			response = sc_pb.Response()
			response_bytes = await websocket.recv()
			response.ParseFromString(response_bytes)
			f.write(f"{response}")
			if len(response.observation.player_result) > 0:
				still_going = False
			
			await websocket.send(makeStepRequest().SerializeToString())
		
			response = sc_pb.Response()
			response_bytes = await websocket.recv()
			response.ParseFromString(response_bytes)
			f.write(f"{response}")
		
		await websocket.send(makeLeaveRequest().SerializeToString())
		
		response = sc_pb.Response()
		response_bytes = await websocket.recv()
		response.ParseFromString(response_bytes)
		f.write(f"{response}")

asyncio.get_event_loop().run_until_complete(hello())
f.close()