Files
AIGame/Tools/enemy_character_pipeline.py
T

194 lines
11 KiB
Python

"""Recipe-driven character art production: same rig -> three clips -> visual prefab -> QA.
Default is local cached production. --generate explicitly requests DGXA inference.
Unity-managed assets are imported/authored only through Unity Pipeline.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import time
REPO=Path(__file__).resolve().parents[1]
PROJECT=REPO/'AIClient'
def sha(path):
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def resolve(path):
result=(REPO/path).resolve()
if not result.is_relative_to(REPO):
raise ValueError(f'Path leaves repository: {path}')
return result
def recipe(path):
r=json.loads(path.read_text(encoding='utf-8-sig'))
if r.get('schemaVersion')!=1 or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_]*',r['id']):
raise ValueError('Unsupported schema / invalid actor id')
if r['profile']!='biped_short':
raise ValueError('This offline motion adapter supports biped_short. Add and validate a different anatomy adapter before using it.')
r.setdefault('usage','enemy_visual')
if r['usage'] not in ('enemy_visual','companion_visual'):
raise ValueError('Unsupported character usage')
r.setdefault('visualPrefab',r.get('enemyPrefab',''))
r.setdefault('attackStyle','body_slam')
if r['attackStyle'] not in ('body_slam','ranged_cast'):
raise ValueError('Unsupported attack style')
r.setdefault('attackEvent','OnCompanionAttackImpact' if r['usage']=='companion_visual' else 'OnEnemyAttackImpact')
if not re.fullmatch(r'[A-Za-z][A-Za-z0-9_]*',r['attackEvent']):
raise ValueError('Invalid attack event')
if r.get('boneMappingReady') is False:
raise ValueError('Inspect this character\'s generated rig and complete its bone mapping before production')
if r['assetRoot']!=f"Assets/Bundles/Characters/{r['kind']}/{r['id']}":
raise ValueError('assetRoot must match kind/id')
for key in ('assetRoot','visualPrefab','staticPrefab','reviewScene'):
if not r[key].startswith('Assets/Bundles/') or '..' in Path(r[key]).parts:
raise ValueError(f'Invalid Unity asset path: {key}')
for key in ('approvedMesh','rigSource','outputDir'):
resolve(r[key])
if not (.05<r['height']<20 and 0<r['walkStride']<r['height'] and 0<r['walkLift']<r['height']*.2):
raise ValueError('Invalid dimensions/stride')
if not 0<r['attackWindup']<r['attackImpact']<r['attackRecover']<1:
raise ValueError('Invalid attack phase order')
if not 0<=r.get('attackReach',0)<r['height']*.25:
raise ValueError('Invalid attack reach')
if not 0<r.get('walkSourceWeight',1)<=1:
raise ValueError('Invalid Walk source weight')
if {m['name'] for m in r['motions']}!={'Idle','Walk','Attack'} or len(r['motions'])!=3:
raise ValueError('Exactly Idle, Walk and Attack required')
for m in r['motions']:
resolve(m['sourceDir'])
if not m['prompt'].isascii() or ('jump' in m['prompt'].lower() and m['name']=='Walk'):
raise ValueError('Use explicit English action captions; do not mix walk and jump')
return r
def unity(command,**kwargs):
args=['unity','--json','command','--project-path',str(PROJECT),command]
for k,v in kwargs.items():
args.extend(['--'+k, json.dumps(v,ensure_ascii=False) if isinstance(v,(list,dict,bool)) else str(v)])
proc=subprocess.run(args,cwd=REPO,capture_output=True,text=True,encoding='utf-8',errors='replace',timeout=180)
if proc.returncode:
raise RuntimeError(proc.stderr or proc.stdout)
data=json.loads(proc.stdout)
result=data.get('data',{}).get('result')
if not data.get('success') or isinstance(result,dict) and result.get('success') is False:
raise RuntimeError(json.dumps(data,ensure_ascii=False))
return result
def script(file,entry,args):
result=unity('run_script',file='AgentScripts/'+file+'.cs',entry=entry,args=args)
print(result.get('result',result),flush=True)
return result
def wait_editor(play_mode):
deadline=time.monotonic()+60
while time.monotonic()<deadline:
try:
state=unity('editor_status')
if state['playMode']==play_mode and not state['compiling'] and not state.get('domainReloadInProgress'):
return
except RuntimeError:
pass # Pipeline reconnects during the expected domain reload.
time.sleep(.25)
raise TimeoutError('Editor did not reach '+play_mode)
def cached_sources(r):
rig_hash=sha(resolve(r['rigSource']))
receipts={}
for m in r['motions']:
folder=resolve(m['sourceDir']);receipt=folder/'motion_request.json'
if not receipt.is_file() and (folder/'walk_request.json').is_file():
receipt=folder/'walk_request.json'
if not (folder/'animated.fbx').is_file() and m.get('cachedFbx'):
# Only accept a cache whose accompanying source rig is byte-identical.
if sha(resolve(m['cachedRig']))!=rig_hash:
raise ValueError('Cached action was generated from another rig: '+m['name'])
folder.mkdir(parents=True,exist_ok=True)
shutil.copy2(resolve(m['cachedFbx']),folder/'animated.fbx')
receipt=folder/'motion_request.json'
receipt.write_text(json.dumps(dict(rig_sha256=rig_hash,motion=m['name'],
prompt=m['cachedPrompt'],cache_original=m['cachedFbx'],cache_rig=m['cachedRig'],
note='Existing same-rig generated seed. Offline authoring defines final action semantics.'),indent=2),encoding='utf-8')
if not receipt.is_file() or not (folder/'animated.fbx').is_file():
raise FileNotFoundError('Missing source motion/provenance: '+m['name'])
record=json.loads(receipt.read_text(encoding='utf-8-sig'))
if record['rig_sha256'].lower()!=rig_hash:
raise ValueError('Motion rig hash mismatch: '+m['name'])
if record.get('source_fbx_sha256') and record['source_fbx_sha256']!=sha(folder/'animated.fbx'):
raise ValueError('Motion source FBX hash mismatch: '+m['name'])
receipts[m['name']]=dict(source=str(folder/'animated.fbx'),sha256=sha(folder/'animated.fbx'),request=record)
return receipts
def main():
parser=argparse.ArgumentParser(description=__doc__)
parser.add_argument('--recipe',type=Path,required=True)
parser.add_argument('--generate',nargs='+',choices=('Idle','Walk','Attack'))
parser.add_argument('--generation-device',choices=('cuda','cpu'),default='cuda')
parser.add_argument('--build-review',action='store_true')
parser.add_argument('--play-check',action='store_true',help='With --build-review, verify all states in actual Play Mode and return to Edit Mode')
parser.add_argument('--validate-recipe',action='store_true')
args=parser.parse_args()
recipe_path=args.recipe.resolve();r=recipe(recipe_path)
if args.validate_recipe:
print('PASS recipe',r['id']);return
if args.play_check and not args.build_review:
parser.error('--play-check requires --build-review')
unity('set_autotick',enable=True)
status=unity('editor_status')
if status['playMode']!='stopped':raise RuntimeError('Use Edit Mode')
if args.generate:
subprocess.run([sys.executable,str(REPO/'Tools/generate_enemy_motions.py'),'--recipe',str(recipe_path),'--device',args.generation_device,'--motions',*args.generate],cwd=REPO,check=True)
sources=cached_sources(r)
output=resolve(r['outputDir']);output.mkdir(parents=True,exist_ok=True)
(output/'production_manifest.json').write_text(json.dumps(dict(schemaVersion=1,status='building',id=r['id'])),encoding='utf-8')
blender=os.environ.get('BLENDER_EXE',r'D:\Tools\Blender\blender-5.0.0-windows-x64\blender.exe')
rigged=output/(r['id']+'_Rigged.fbx')
subprocess.run([blender,'-b','--factory-startup','--python',str(REPO/'Tools/model_tools/bl_transfer_skin.py'),'--',
str(resolve(r['rigSource'])),str(resolve(r['approvedMesh'])),str(rigged),'--low-up',r['lowUp'],'--align-bounds',
'--weight-overrides',str(recipe_path)],cwd=REPO,check=True)
unity('import_asset',source=str(rigged),path=r['assetRoot']+'/Rigged/'+r['id']+'_Rigged.fbx',confirm=True)
for name,record in sources.items():
unity('import_asset',source=record['source'],path=r['assetRoot']+'/Anims/'+r['id']+'_'+name+'.fbx',confirm=True)
script('SetupAnimalAnimator','SetupAnimalAnimator.Run',[r['kind'],r['id'],r['height'],False])
script('FinalizeAnimalWalk','FinalizeAnimalWalk.RunFor',[r['assetRoot'],r['id'],r['height'],r['rootBone'],r['rightFoot'],r['leftFoot'],r['walkStride'],r['walkLift'],r['walkSeconds'],r['outputDir'],r.get('walkBindCalibration',False),r.get('walkSourceWeight',1)])
relative=recipe_path.relative_to(REPO).as_posix()
script('FinalizeEnemyMotions','FinalizeEnemyMotions.Run',[relative])
script('VerifyEnemyMotions','VerifyEnemyMotions.Run',[relative])
script('BuildEnemyReview','BuildEnemyReview.Publish',[relative])
if args.build_review:
script('BuildEnemyReview','BuildEnemyReview.Build',[relative])
script('BuildEnemyReview','BuildEnemyReview.Capture',[relative])
live_result=None
if args.play_check:
unity('editor_play')
try:
wait_editor('playing')
live_result=script('VerifyEnemyMotions','VerifyEnemyMotions.Live',[relative])['result']
prefix='CompanionReview_'+r['id'] if r['usage']=='companion_visual' else 'EnemyReview'
role='companion' if r['usage']=='companion_visual' else 'enemy'
unity('capture_game_view',camera=prefix+'Camera',width=1900,height=760,
save_path='Assets/Screenshots/'+role+'_'+r['id'].lower()+'_play.png')
finally:
unity('editor_stop')
wait_editor('stopped')
(output/'play_verification.txt').write_text(live_result,encoding='utf-8')
manifest=dict(schemaVersion=1,status='validated' if live_result else 'edit_validated',id=r['id'],usage=r['usage'],recipe=relative,
approved_mesh_sha256=sha(resolve(r['approvedMesh'])),rig_sha256=sha(resolve(r['rigSource'])),
sources=sources,rigged_fbx=str(rigged),visual_prefab=r['visualPrefab'],
motions=['Idle','Walk','Attack'],controller=dict(Motion='0=Idle,1=Walk',Attack='Trigger, one shot'),
attack_impact_seconds=r['attackSeconds']*r['attackImpact'],
qa=dict(edit='passed',play=live_result or 'not_run'),
report=str(output/'verification.txt'),generated_at_utc=time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime()))
if r['usage']=='enemy_visual':manifest['enemy_prefab']=r['visualPrefab']
(output/'production_manifest.json').write_text(json.dumps(manifest,ensure_ascii=False,indent=2),encoding='utf-8')
print('PASS character production',r['visualPrefab'],flush=True)
if __name__=='__main__':
main()