Page MenuHomeWildfire Games
Paste P277

Production Queue Validator
ActivePublic

Authored by Stan on Jun 17 2022, 10:58 PM.
#!/usr/bin/env python3
# -*- mode: python-mode; python-indent-offset: 4; -*-
# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: © 2022 Stanislas Daniel Claude Dolcini
from html import entities
from logging import getLogger, StreamHandler, INFO, WARNING, Formatter, Filter
from urllib import request
from xml.etree import ElementTree
import argparse
import json
import os
import pathlib
import psutil
import subprocess
import sys
class SingleLevelFilter(Filter):
def __init__(self, passlevel, reject):
self.passlevel = passlevel
self.reject = reject
def filter(self, record):
if self.reject:
return (record.levelno != self.passlevel)
else:
return (record.levelno == self.passlevel)
class ProductionQueueValidator():
def __init__(self, vfs_root, mods=None, url=None, verbose=False):
self.mods = mods if mods is not None else []
self.vfs_root = pathlib.Path(vfs_root)
self.__init_logger
self.verbose = verbose
self.url = url
@property
def __init_logger(self):
logger = getLogger(__name__)
logger.setLevel(INFO)
# create a console handler, seems nicer to Windows and for future uses
ch = StreamHandler(sys.stdout)
ch.setLevel(INFO)
ch.setFormatter(Formatter('%(levelname)s - %(message)s'))
f1 = SingleLevelFilter(INFO, False)
ch.addFilter(f1)
logger.addHandler(ch)
errorch = StreamHandler(sys.stderr)
errorch.setLevel(WARNING)
errorch.setFormatter(Formatter('%(levelname)s - %(message)s'))
logger.addHandler(errorch)
self.logger = logger
def get_templates_from_engine(self):
pyrogenesis_path = self.vfs_root / 'binaries/system/pyrogenesis'
mod_args = ''
for mod in self.mods:
mod_args += f'-mod="{mod}" '
args = f' {mod_args}--rl-interface="{self.url}" --autostart-nonvisual --autostart="skirmishes/acropolis_bay_2p"'
process = subprocess.Popen(str(pyrogenesis_path) + args, stdout=subprocess.PIPE, shell=True)
output = b""
while b'RL interface listening on 127.0.0.1:9090\n' != output:
output = process.stdout.readline()
if self.verbose:
self.logger.info(output)
if self.verbose:
self.logger.info(f"Launching 0.A.D with the following command {str(pyrogenesis_path) + args}");
try:
data = '''{
const cmpTemplateManager = Engine.QueryInterface(SYSTEM_ENTITY, IID_TemplateManager);
const templates = cmpTemplateManager.FindAllTemplates(false);
const templateMap = new Map();
for (const template of templates)
{
const temp = cmpTemplateManager.GetTemplate(template);
if (temp['Trainer'] || temp['Researcher'])
{
const ent = Engine.AddEntity(template);
const cmpProductionQueue = Engine.QueryInterface(ent, IID_ProductionQueue);
let result = Object.assign({}, temp)
if (cmpProductionQueue)
{
// It seems the game doesn't send empty components through the RL interface.
result.ProductionQueue = {};
}
templateMap[template] = result;
}
}
templateMap
}'''
if self.verbose:
self.logger.info(f"Calling http://{self.url}/evaluate to get data");
req = request.Request(f'http://{self.url}/evaluate', data=data.encode('utf-8'))
resp = request.urlopen(req)
return json.loads(resp.read())
except Exception as err:
self.logger.error(err)
finally:
if self.verbose:
self.logger.info(f"Killing 0 A.D.");
self.kill(process.pid)
def kill(self, proc_pid):
process = psutil.Process(proc_pid)
for proc in process.children(recursive=True):
proc.kill()
process.kill()
def find_files_relative(self, vfs_path, *ext_list):
return self.find_files(pathlib.Path(self.vfs_root / 'binaries/data/mods/'), self.mods, vfs_path, *ext_list)
def find_files(self, vfs_root, mods, vfs_path, *ext_list):
"""
returns a list of 2-size tuple with:
- Path relative to the mod base
- full Path
"""
full_exts = ['.' + ext for ext in ext_list]
def find_recursive(dp, base):
"""(relative Path, full Path) generator"""
if dp.is_dir():
if dp.name != '.svn' and dp.name != '.git' and not dp.name.endswith('~'):
for fp in dp.iterdir():
yield from find_recursive(fp, base)
elif dp.suffix in full_exts:
relative_file_path = dp.relative_to(base)
yield (relative_file_path, dp.resolve())
return [(rp, fp) for mod in mods for (rp, fp) in find_recursive(vfs_root / mod / vfs_path, vfs_root / mod)]
def run (self):
template_data = self.get_templates_from_engine()
if self.verbose:
self.logger.info(f"Looking for templates with missing production queue.");
for template in template_data:
if ('Trainer' in template_data[template] or 'Researcher' in template_data[template]) and not 'ProductionQueue' in template_data[template]:
self.logger.error(f"{template} has no production queue.");
break;
if __name__ == '__main__':
script_dir = os.path.dirname(os.path.realpath(__file__))
default_root = os.path.join(script_dir, '..', '..', '..')
parser = argparse.ArgumentParser(description='Production Queue Validator.')
parser.add_argument('-r', '--root', action='store', dest='root', default=default_root)
parser.add_argument('-m', '--mods', action='store', dest='mods', default='mod,public')
parser.add_argument('-u', '--url', action='store', dest='url', default='127.0.0.1:9090')
parser.add_argument('-v', '--verbose', action='store_true', default=False,
help="Log validation errors.")
args = parser.parse_args()
validator = ProductionQueueValidator(args.root, args.mods.split(','), args.url, args.verbose)
validator.run()

Event Timeline

Stan created this paste.Jun 17 2022, 10:58 PM
Stan created this object with visibility "Public (No Login Required)".
Stan mentioned this in D4695: Fix Naval Shipyard..
Stan edited the content of this paste. (Show Details)Feb 3 2023, 12:02 AM
Stan edited the content of this paste. (Show Details)Feb 3 2023, 12:15 PM