gptbot/0ad
1
0
Fork 0
forked from mirrors/0ad
Code Pull Requests Activity

Add an interface for Reinforcement Learning.

Browse Source 
Implement a simple HTTP server to start games, receive the gamestate and
pass commands to the simulation.
This is mainly intended for training reinforcement learning agents in 0
AD. As such, a python client and a small example are included.

This option can be enabled using the -rl-interface flag.

Patch by: irishninja
Reviewed By: wraitii, Itms
Fixes #5548

Differential Revision: https://code.wildfiregames.com/D2199
This was SVN commit r23917.
This commit is contained in:
wraitii
2020-08-01 10:52:59 +00:00
parent 164af0742a
commit 5473393e30
19 changed files with 1091 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
source/tools/rlclient/python/README.md
+50
View File
@@ -0,0 +1,50 @@
# 0 AD Python Client
This directory contains `zero_ad`, a python client for 0 AD which enables users to control the environment headlessly.
## Installation
`zero_ad` can be installed with `pip` by running the following from the current directory:
```
pip install .
```
Development dependencies can be installed with `pip install -r requirements-dev.txt`. Tests are using pytest and can be run with `python -m pytest`.
## Basic Usage
If there is not a running instance of 0 AD, first start 0 AD with the RL interface enabled:
```
pyrogenesis --rl-interface=127.0.0.1:6000
```
Next, the python client can be connected with:
```
import zero_ad
from zero_ad import ZeroAD
game = ZeroAD('http://localhost:6000')
```
A map can be loaded with:
```
with open('./samples/arcadia.json', 'r') as f:
arcadia_config = f.read()
state = game.reset(arcadia_config)
```
where `./samples/arcadia.json` is the path to a game configuration JSON (included in the first line of the commands.txt file in a game replay directory) and `state` contains the initial game state for the given map. The game engine can be stepped (optionally applying actions at each step) with:
```
state = game.step()
```
For example, enemy units could be attacked with:
```
my_units = state.units(owner=1)
enemy_units = state.units(owner=2)
actions = [zero_ad.actions.attack(my_units, enemy_units[0])]
state = game.step(actions)
```
For a more thorough example, check out samples/simple-example.py!
source/tools/rlclient/python/requirements-dev.txt
+1
View File
@@ -0,0 +1 @@
pytest
source/tools/rlclient/python/samples/arcadia.json
+53
View File
@@ -0,0 +1,53 @@
{
"settings": {
"TriggerScripts": [
"scripts/TriggerHelper.js",
"scripts/ConquestCommon.js",
"scripts/ConquestUnits.js"
],
"VictoryConditions": [
"conquest_units"
],
"Name": "Arcadia",
"mapType": "scenario",
"AISeed": 0,
"Seed": 0,
"CheatsEnabled": true,
"Ceasefire": 0,
"WonderDuration": 10,
"RelicDuration": 10,
"RelicCount": 2,
"Size": 256,
"PlayerData": [
{
"Name": "Player 1",
"Civ": "spart",
"Color": {
"r": 150,
"g": 20,
"b": 20
},
"AI": "",
"AIDiff": 3,
"AIBehavior": "random",
"Team": 1
},
{
"Name": "Player 2",
"Civ": "spart",
"Color": {
"r": 150,
"g": 20,
"b": 20
},
"AI": "",
"AIDiff": 3,
"AIBehavior": "random",
"Team": 2
}
]
},
"mapType": "scenario",
"map": "maps/scenarios/arcadia",
"gameSpeed": 1
}
source/tools/rlclient/python/samples/simple-example.py
+98
View File
@@ -0,0 +1,98 @@
# This script provides an overview of the zero_ad wrapper for 0 AD
from os import path
import zero_ad
# First, we will define some helper functions we will use later.
import math
def dist (p1, p2):
return math.sqrt(sum((math.pow(x2 - x1, 2) for (x1, x2) in zip(p1, p2))))
def center(units):
sum_position = map(sum, zip(*map(lambda u: u.position(), units)))
return [x/len(units) for x in sum_position]
def closest(units, position):
dists = (dist(unit.position(), position) for unit in units)
index = 0
min_dist = next(dists)
for (i, d) in enumerate(dists):
if d < min_dist:
index = i
min_dist = d
return units[index]
# Connect to a 0 AD game server listening at localhost:6000
game = zero_ad.ZeroAD('http://localhost:6000')
# Load the Arcadia map
samples_dir = path.dirname(path.realpath(__file__))
scenario_config_path = path.join(samples_dir, 'arcadia.json')
with open(scenario_config_path, 'r') as f:
arcadia_config = f.read()
state = game.reset(arcadia_config)
# The game is paused and will only progress upon calling "step"
state = game.step()
# Units can be queried from the game state
citizen_soldiers = state.units(owner=1, type='infantry')
# (including gaia units like trees or other resources)
nearby_tree = closest(state.units(owner=0, type='tree'), center(citizen_soldiers))
# Action commands can be created using zero_ad.actions
collect_wood = zero_ad.actions.gather(citizen_soldiers, nearby_tree)
female_citizens = state.units(owner=1, type='female_citizen')
house_tpl = 'structures/spart_house'
x = 680
z = 640
build_house = zero_ad.actions.construct(female_citizens, house_tpl, x, z, autocontinue=True)
# These commands can then be applied to the game in a `step` command
state = game.step([collect_wood, build_house])
# We can also fetch units by id using the `unit` function on the game state
female_id = female_citizens[0].id()
female_citizen = state.unit(female_id)
# A variety of unit information can be queried from the unit:
print('female citizen\'s max health is', female_citizen.max_health())
# Raw data for units and game states are available via the data attribute
print(female_citizen.data)
# Units can be built using the "train action"
civic_center = state.units(owner=1, type="civil_centre")[0]
spearman_type = 'units/spart_infantry_spearman_b'
train_spearmen = zero_ad.actions.train([civic_center], spearman_type)
state = game.step([train_spearmen])
# Let's step the engine until the house has been built
is_unit_busy = lambda state, unit_id: len(state.unit(unit_id).data['unitAIOrderData']) > 0
while is_unit_busy(state, female_id):
state = game.step()
# The units for the other army can also be controlled
enemy_units = state.units(owner=2)
walk = zero_ad.actions.walk(enemy_units, *civic_center.position())
game.step([walk], player=[2])
# Step the game engine a bit to give them some time to walk
for _ in range(150):
state = game.step()
# Let's attack with our entire military
state = game.step([zero_ad.actions.chat('An attack is coming!')])
while len(state.units(owner=2, type='unit')) > 0:
attack_units = [ unit for unit in state.units(owner=1, type='unit') if 'female' not in unit.type() ]
target = closest(state.units(owner=2, type='unit'), center(attack_units))
state = game.step([zero_ad.actions.attack(attack_units, target)])
while state.unit(target.id()):
state = game.step()
game.step([zero_ad.actions.chat('The enemies have been vanquished. Our home is safe again.')])
source/tools/rlclient/python/setup.py
+13
View File
@@ -0,0 +1,13 @@
import os
from setuptools import setup
setup(name='zero_ad',
version='0.0.1',
description='Python client for 0 AD',
url='https://code.wildfiregames.com',
author='Brian Broll',
author_email='brian.broll@gmail.com',
install_requires=[],
license='MIT',
packages=['zero_ad'],
zip_safe=False)
source/tools/rlclient/python/tests/test_actions.py
+100
View File
@@ -0,0 +1,100 @@
import zero_ad
import json
import math
from os import path
game = zero_ad.ZeroAD('http://localhost:6000')
scriptdir = path.dirname(path.realpath(__file__))
with open(path.join(scriptdir, '..', 'samples', 'arcadia.json'), 'r') as f:
config = f.read()
def dist (p1, p2):
return math.sqrt(sum((math.pow(x2 - x1, 2) for (x1, x2) in zip(p1, p2))))
def center(units):
sum_position = map(sum, zip(*map(lambda u: u.position(), units)))
return [x/len(units) for x in sum_position]
def closest(units, position):
dists = (dist(unit.position(), position) for unit in units)
index = 0
min_dist = next(dists)
for (i, d) in enumerate(dists):
if d < min_dist:
index = i
min_dist = d
return units[index]
def test_construct():
state = game.reset(config)
female_citizens = state.units(owner=1, type='female_citizen')
house_tpl = 'structures/spart_house'
house_count = len(state.units(owner=1, type=house_tpl))
x = 680
z = 640
build_house = zero_ad.actions.construct(female_citizens, house_tpl, x, z, autocontinue=True)
# Check that they start building the house
state = game.step([build_house])
while len(state.units(owner=1, type=house_tpl)) == house_count:
state = game.step()
def test_gather():
state = game.reset(config)
female_citizen = state.units(owner=1, type='female_citizen')[0]
trees = state.units(owner=0, type='tree')
nearby_tree = closest(state.units(owner=0, type='tree'), female_citizen.position())
collect_wood = zero_ad.actions.gather([female_citizen], nearby_tree)
state = game.step([collect_wood])
while len(state.unit(female_citizen.id()).data['resourceCarrying']) == 0:
state = game.step()
def test_train():
state = game.reset(config)
civic_centers = state.units(owner=1, type="civil_centre")
spearman_type = 'units/spart_infantry_spearman_b'
spearman_count = len(state.units(owner=1, type=spearman_type))
train_spearmen = zero_ad.actions.train(civic_centers, spearman_type)
state = game.step([train_spearmen])
while len(state.units(owner=1, type=spearman_type)) == spearman_count:
state = game.step()
def test_walk():
state = game.reset(config)
female_citizens = state.units(owner=1, type='female_citizen')
x = 680
z = 640
initial_distance = dist(center(female_citizens), [x, z])
walk = zero_ad.actions.walk(female_citizens, x, z)
state = game.step([walk])
distance = initial_distance
while distance >= initial_distance:
state = game.step()
female_citizens = state.units(owner=1, type='female_citizen')
distance = dist(center(female_citizens), [x, z])
def test_attack():
state = game.reset(config)
units = state.units(owner=1, type='cavalry')
target = state.units(owner=2, type='female_citizen')[0]
initial_health = target.health()
state = game.step([zero_ad.actions.reveal_map()])
attack = zero_ad.actions.attack(units, target)
state = game.step([attack])
while state.unit(target.id()).health() >= initial_health:
state = game.step()
def test_debug_print():
state = game.reset(config)
debug_print = zero_ad.actions.debug_print('hello world!!')
state = game.step([debug_print])
def test_chat():
state = game.reset(config)
chat = zero_ad.actions.chat('hello world!!')
state = game.step([chat])
source/tools/rlclient/python/zero_ad/__init__.py
+4
View File
@@ -0,0 +1,4 @@
from . import actions
from . import environment
ZeroAD = environment.ZeroAD
GameState = environment.GameState
source/tools/rlclient/python/zero_ad/actions.py
+69
View File
@@ -0,0 +1,69 @@
def construct(units, template, x, z, angle=0, autorepair=True, autocontinue=True, queued=False):
unit_ids = [ unit.id() for unit in units ]
return {
'type': 'construct',
'entities': unit_ids,
'template': template,
'x': x,
'z': z,
'angle': angle,
'autorepair': autorepair,
'autocontinue': autocontinue,
'queued': queued,
}
def gather(units, target, queued=False):
unit_ids = [ unit.id() for unit in units ]
return {
'type': 'gather',
'entities': unit_ids,
'target': target.id(),
'queued': queued,
}
def train(entities, unit_type, count=1):
entity_ids = [ unit.id() for unit in entities ]
return {
'type': 'train',
'entities': entity_ids,
'template': unit_type,
'count': count,
}
def debug_print(message):
return {
'type': 'debug-print',
'message': message
}
def chat(message):
return {
'type': 'aichat',
'message': message
}
def reveal_map():
return {
'type': 'reveal-map',
'enable': True
}
def walk(units, x, z, queued=False):
ids = [ unit.id() for unit in units ]
return {
'type': 'walk',
'entities': ids,
'x': x,
'z': z,
'queued': queued
}
def attack(units, target, queued=False, allow_capture=True):
unit_ids = [ unit.id() for unit in units ]
return {
'type': 'attack',
'entities': unit_ids,
'target': target.id(),
'allowCapture': allow_capture,
'queued': queued
}
source/tools/rlclient/python/zero_ad/api.py
+29
View File
@@ -0,0 +1,29 @@
import urllib
from urllib import request
import json
class RLAPI():
def __init__(self, url):
self.url = url
def post(self, route, data):
response = request.urlopen(url=f'{self.url}/{route}', data=bytes(data, 'utf8'))
return response.read()
def step(self, commands):
post_data = '\n'.join((f'{player};{json.dumps(action)}' for (player, action) in commands))
return self.post('step', post_data)
def reset(self, scenario_config, player_id, save_replay):
path = 'reset?'
if save_replay:
path += 'saveReplay=1&'
if player_id:
path += f'playerID={player_id}&'
return self.post(path, scenario_config)
def get_templates(self, names):
post_data = '\n'.join(names)
response = self.post('templates', post_data)
return zip(names, response.decode().split('\n'))
source/tools/rlclient/python/zero_ad/environment.py
+113
View File
@@ -0,0 +1,113 @@
from .api import RLAPI
import json
import math
from xml.etree import ElementTree
from itertools import cycle
class ZeroAD():
def __init__(self, uri='http://localhost:6000'):
self.api = RLAPI(uri)
self.current_state = None
self.cache = {}
self.player_id = 1
def step(self, actions=[], player=None):
player_ids = cycle([self.player_id]) if player is None else cycle(player)
cmds = zip(player_ids, actions)
cmds = ((player, action) for (player, action) in cmds if action is not None)
state_json = self.api.step(cmds)
self.current_state = GameState(json.loads(state_json), self)
return self.current_state
def reset(self, config='', save_replay=False, player_id=1):
state_json = self.api.reset(config, player_id, save_replay)
self.current_state = GameState(json.loads(state_json), self)
return self.current_state
def get_template(self, name):
return self.get_templates([name])[0]
def get_templates(self, names):
templates = self.api.get_templates(names)
return [ (name, EntityTemplate(content)) for (name, content) in templates ]
def update_templates(self, types=[]):
all_types = list(set([unit.type() for unit in self.current_state.units()]))
all_types += types
template_pairs = self.get_templates(all_types)
self.cache = {}
for (name, tpl) in template_pairs:
self.cache[name] = tpl
return template_pairs
class GameState():
def __init__(self, data, game):
self.data = data
self.game = game
self.mapSize = self.data['mapSize']
def units(self, owner=None, type=None):
filter_fn = lambda e: (owner is None or e['owner'] == owner) and \
(type is None or type in e['template'])
return [ Entity(e, self.game) for e in self.data['entities'].values() if filter_fn(e) ]
def unit(self, id):
id = str(id)
return Entity(self.data['entities'][id], self.game) if id in self.data['entities'] else None
class Entity():
def __init__(self, data, game):
self.data = data
self.game = game
self.template = self.game.cache.get(self.type(), None)
def type(self):
return self.data['template']
def id(self):
return self.data['id']
def owner(self):
return self.data['owner']
def max_health(self):
template = self.get_template()
return float(template.get('Health/Max'))
def health(self, ratio=False):
if ratio:
return self.data['hitpoints']/self.max_health()
return self.data['hitpoints']
def position(self):
return self.data['position']
def get_template(self):
if self.template is None:
self.game.update_templates([self.type()])
self.template = self.game.cache[self.type()]
return self.template
class EntityTemplate():
def __init__(self, xml):
self.data = ElementTree.fromstring(f'<Entity>{xml}</Entity>')
def get(self, path):
node = self.data.find(path)
return node.text if node is not None else None
def set(self, path, value):
node = self.data.find(path)
if node:
node.text = str(value)
return node is not None
def __str__(self):
return ElementTree.tostring(self.data).decode('utf-8')