dusk/tools/asset/cutscene/__main__.py

# Copyright (c) 2026 Dominic Masters
#
# This software is released under the MIT License.
# https://opensource.org/licenses/MIT

"""
Generates DCTS binary cutscene files from raw JSONC cutscene definitions.

JSONC input (assetsraw/cutscenes/<name>.jsonc) - plain JSON plus // and
/* */ comments, stripped before parsing:
  {
    // optional flag names, default matches CUTSCENE_PAUSE_DEFAULT
    "pause": ["NPC", "PLAYER"],
    "items": [
      { "type": "TEXT", "text": "Hello!" },
      { "type": "WAIT", "seconds": 1.0 },
      { "type": "MARKER", "name": "GREET" },
      ...
    ]
  }

Output: assetsraw/cutscenes/<name>.jsonc -> assets/cutscenes/<name>.cts

Only a subset of cutscene item types is supported (v1) - anything needing a
native callback (CALLBACK, MAP_AREA_ADD, the multi-option form of MODAL) or
recursive item data (CUTSCENE, CONCURRENT) isn't representable in a file
yet; those still have to be authored as compiled-in CUTSCENE(...) macros.
See src/dusk/asset/loader/cutscene/assetcutsceneloader.c for the C reader
this must match byte-for-byte, and src/dusk/rpg/cutscene/item/cutsceneitem.h
for the authoritative enum - ITEM_TYPE below MUST match its declared order.

DCTS format (little-endian throughout):
  Header (12 bytes):
    magic       b"DCTS"  4 bytes
    version     u32      ASSET_CUTSCENE_FILE_VERSION
    pauseType   u8       cutscenepause_t bitmask
    itemCount   u8
    poolSize    u16      bytes in the pool blob following the item stream
  Item stream (itemCount records back to back):
    type        u8       cutsceneitemtype_t
    ...type-specific payload; embedded strings are length-prefixed (u8 len
       + bytes, no null terminator); references into the pool are a u16
       byte offset...
  Pool (poolSize bytes): null-terminated strings and/or raw fixed-width
    array data (e.g. worldpos_t triples), referenced by offset from the
    item stream. Every pool entry starts 4-byte aligned.
"""

import argparse
import json
import os
import struct
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
PROJECT_ROOT = os.path.normpath(os.path.join(_HERE, '..', '..', '..'))
ASSETSRAW_DIR = os.path.join(PROJECT_ROOT, 'assetsraw')
ASSETS_DIR = os.path.join(PROJECT_ROOT, 'assets')

FILE_MAGIC = b'DCTS'
VERSION_OUT = 1

# Must match cutsceneitemtype_t's declared order in
# src/dusk/rpg/cutscene/item/cutsceneitem.h exactly. Types with no entry
# here are v2 (native callback / recursive item data) and unsupported.
ITEM_TYPE = {
  'TEXT': 1,
  'TEXT_MINI': 2,
  'TEXT_MINI_HIDE': 3,
  'WAIT': 5,
  'ENTITY_TELEPORT': 7,
  'ENTITY_WALK_TO': 8,
  'FADE': 9,
  'SET_PAUSE': 10,
  'ITEM_GIVE': 12,
  'ENTITY_REMOVE': 13,
  'ENTITY_ADD': 14,
  'ENTITY_TURN': 15,
  'ENTITY_WALK_TO_ENTITY': 16,
  'MAP_AREA_REMOVE': 18,
  'MAP_AREA_WAIT': 19,
  'START_BATTLE': 20,
  'EMOJI': 21,
  'SHAKE': 22,
  'BATTLE_WAIT_STATE': 23,
  'BATTLE_FORCE_ACTION': 24,
  'MODAL': 25,
  'MODAL_OPTIONS_MARKERS': 26,
  'MODAL_CLOSE': 27,
  'PRINT': 28,
  'MARKER': 29,
  'RESTART': 30,
  'SCENE': 31,
  'SAVE_DEVICE_CHECK': 32,
  'SAVE_LOAD_ALL_SLOTS': 33,
  'AUDIO_PLAY': 35,
  'AUDIO_STOP': 36,
  'AUDIO_PAUSE': 37,
  'AUDIO_RESUME': 38,
  'AUDIO_FADE': 39,
  'AUDIO_FADE_WAIT': 40,
  'AUDIO_SET_PAN': 41,
  'AUDIO_SET_LOOP': 42,
  'AUDIO_SET': 43,
  'IDLE': 44,
}

PAUSE_FLAG = {'NPC': 1 << 0, 'PLAYER': 1 << 1, 'WORLD': 1 << 2, 'BATTLE': 1 << 3}
PAUSE_DEFAULT = PAUSE_FLAG['NPC'] | PAUSE_FLAG['PLAYER']

ENTITY_SENTINEL = {'INTERACT': 0xFE, 'INTERACTED': 0xFD}
AREA_SENTINEL = {'LAST_CREATED': 0xFF}

ENTITY_TYPE = {'NULL': 0, 'PLAYER': 1, 'NPC': 2, 'ITEM': 3}
ENTITY_DIR = {
  'NORTH': 0, 'EAST': 1, 'SOUTH': 2, 'WEST': 3,
  'UP': 0, 'RIGHT': 1, 'DOWN': 2, 'LEFT': 3,
}
EASING = {
  name: i for i, name in enumerate([
    'LINEAR', 'IN_SINE', 'OUT_SINE', 'IN_OUT_SINE',
    'IN_QUAD', 'OUT_QUAD', 'IN_OUT_QUAD',
    'IN_CUBIC', 'OUT_CUBIC', 'IN_OUT_CUBIC',
    'IN_QUART', 'OUT_QUART', 'IN_OUT_QUART',
    'IN_BACK', 'OUT_BACK', 'IN_OUT_BACK',
  ])
}
UI_EMOJI = {'NULL': 0, 'QUESTION_MARK': 1, 'EXCLAMATION_MARK': 2}
BATTLE_ENCOUNTER = {'REGULAR': 0, 'PLAYER_ADVANTAGE': 1, 'BACK_ATTACK': 2}
BATTLE_STATE = {
  'NONE': 0, 'OPENING': 1, 'PRE_ROUND': 2, 'PLAYER_SELECTION': 3,
  'AI_SELECTION': 4, 'MOVES_EXECUTING': 5, 'POST_ROUND': 6, 'ENDED': 7,
}
SCENE_TYPE = {'NULL': 0, 'INITIAL': 1, 'MAIN_MENU': 2, 'OVERWORLD': 3, 'BATTLE': 4}
AUDIO_CHANNEL = {
  'BGM_0': 0, 'VOICE_0': 1, 'VOICE_1': 2,
  'SFX_0': 3, 'SFX_1': 4, 'SFX_2': 5, 'SFX_3': 6,
}

CUTSCENE_TEXT_MAX_CHARS = 256
CUTSCENE_TEXT_MINI_MAX_CHARS = 128
CUTSCENE_PRINT_MAX_CHARS = 128
CUTSCENE_MODAL_TITLE_MAX_CHARS = 64
CUTSCENE_MODAL_MESSAGE_MAX_CHARS = 256
CUTSCENE_MODAL_OPTIONS_MARKERS_MAX = 2
CUTSCENE_MAP_AREA_WAIT_MAX = 4
CUTSCENE_START_BATTLE_ENEMY_COUNT_MAX = 4
AUDIO_PATH_MAX = 256


class Pool:
  def __init__(self):
    self.data = bytearray()

  def _align(self, n):
    while len(self.data) % n != 0:
      self.data += b'\x00'

  def add_string(self, s):
    self._align(4)
    offset = len(self.data)
    self.data += s.encode('utf-8') + b'\x00'
    return offset

  def add_bytes(self, b):
    self._align(4)
    offset = len(self.data)
    self.data += b
    return offset


def resolve_entity_index(v):
  if isinstance(v, str):
    return ENTITY_SENTINEL[v.upper()]
  return int(v)


def resolve_area_id(v):
  if isinstance(v, str):
    return AREA_SENTINEL[v.upper()]
  return int(v)


def write_string_field(buf, s, max_len):
  encoded = s.encode('utf-8')
  if len(encoded) >= max_len:
    raise ValueError(f"String exceeds max length {max_len}: {s!r}")
  buf += struct.pack('<B', len(encoded))
  buf += encoded


def write_worldpos(buf, pos):
  x, y, z = pos
  buf += struct.pack('<hhh', int(x), int(y), int(z))


def encode_item(item, pool):
  item_type = item['type']
  type_id = ITEM_TYPE.get(item_type)
  if type_id is None:
    raise ValueError(
      f"Unsupported cutscene item type for file-based cutscenes: {item_type}"
    )

  buf = bytearray()
  buf += struct.pack('<B', type_id)

  if item_type == 'TEXT':
    write_string_field(buf, item['text'], CUTSCENE_TEXT_MAX_CHARS)

  elif item_type == 'TEXT_MINI':
    write_string_field(buf, item['text'], CUTSCENE_TEXT_MINI_MAX_CHARS)
    x, y, z = item['position']
    buf += struct.pack('<fff', float(x), float(y), float(z))
    buf += struct.pack('<f', float(item['duration']))

  elif item_type == 'TEXT_MINI_HIDE':
    buf += struct.pack('<B', int(item['index']))

  elif item_type == 'WAIT':
    buf += struct.pack('<f', float(item['seconds']))

  elif item_type == 'ENTITY_TELEPORT':
    buf += struct.pack('<B', resolve_entity_index(item['entityIndex']))
    write_worldpos(buf, item['target'])

  elif item_type == 'ENTITY_WALK_TO':
    buf += struct.pack('<B', resolve_entity_index(item['entityIndex']))
    buf += struct.pack('<B', 1 if item.get('walkAround', True) else 0)
    positions = item['positions']
    buf += struct.pack('<B', len(positions))
    positions_bytes = bytearray()
    for pos in positions:
      write_worldpos(positions_bytes, pos)
    offset = pool.add_bytes(bytes(positions_bytes))
    buf += struct.pack('<H', offset)

  elif item_type == 'FADE':
    buf += bytes(int(c) for c in item['from'])
    buf += bytes(int(c) for c in item['to'])
    buf += struct.pack('<f', float(item['duration']))
    buf += struct.pack('<B', EASING[item.get('easing', 'LINEAR').upper()])

  elif item_type == 'SET_PAUSE':
    value = 0
    for name in item['flags']:
      value |= PAUSE_FLAG[name.upper()]
    buf += struct.pack('<B', value)

  elif item_type == 'ITEM_GIVE':
    buf += struct.pack('<H', int(item['item']))
    buf += struct.pack('<B', int(item['quantity']))

  elif item_type == 'ENTITY_REMOVE':
    buf += struct.pack('<B', resolve_entity_index(item['entityIndex']))

  elif item_type == 'ENTITY_ADD':
    buf += struct.pack('<B', ENTITY_TYPE[item['entityType'].upper()])
    write_worldpos(buf, item['position'])

  elif item_type == 'ENTITY_TURN':
    buf += struct.pack('<B', resolve_entity_index(item['entityIndex']))
    buf += struct.pack('<B', ENTITY_DIR[item['direction'].upper()])

  elif item_type == 'ENTITY_WALK_TO_ENTITY':
    buf += struct.pack('<B', resolve_entity_index(item['entityIndex']))
    buf += struct.pack('<B', resolve_entity_index(item['targetEntityIndex']))
    buf += struct.pack('<hh', int(item['offsetX']), int(item['offsetY']))

  elif item_type == 'MAP_AREA_REMOVE':
    buf += struct.pack('<B', resolve_area_id(item['areaId']))

  elif item_type == 'MAP_AREA_WAIT':
    area_ids = item['areaIds']
    if len(area_ids) > CUTSCENE_MAP_AREA_WAIT_MAX:
      raise ValueError("MAP_AREA_WAIT areaIds exceeds maximum of 4")
    buf += struct.pack('<B', len(area_ids))
    ids_bytes = bytes(resolve_area_id(a) for a in area_ids)
    offset = pool.add_bytes(ids_bytes)
    buf += struct.pack('<H', offset)

  elif item_type == 'START_BATTLE':
    buf += struct.pack('<B', BATTLE_ENCOUNTER[item['encounterType'].upper()])
    buf += struct.pack('<B', 1 if item.get('fleeAvailable', True) else 0)
    enemies = item['enemies']
    if len(enemies) > CUTSCENE_START_BATTLE_ENEMY_COUNT_MAX:
      raise ValueError("START_BATTLE enemies exceeds maximum of 4")
    buf += struct.pack('<B', len(enemies))
    for enemy in enemies:
      stats = enemy['stats']
      buf += struct.pack(
        '<HHHHHHH',
        int(stats['attack']), int(stats['defense']), int(stats['magic']),
        int(stats['speed']), int(stats['luck']),
        int(enemy['healthMax']), int(enemy['mpMax']),
      )

  elif item_type == 'EMOJI':
    buf += struct.pack('<B', resolve_entity_index(item['entityIndex']))
    buf += struct.pack('<f', float(item['duration']))
    buf += struct.pack('<B', UI_EMOJI[item['emojiType'].upper()])

  elif item_type == 'SHAKE':
    buf += struct.pack('<B', int(item['amount']))
    buf += struct.pack('<f', float(item['duration']))

  elif item_type == 'BATTLE_WAIT_STATE':
    buf += struct.pack('<B', BATTLE_STATE[item['state'].upper()])

  elif item_type == 'BATTLE_FORCE_ACTION':
    buf += struct.pack('<B', int(item['fighterIndex']))
    buf += struct.pack('<B', int(item['targetIndex']))

  elif item_type == 'MODAL':
    write_string_field(buf, item.get('title', ''), CUTSCENE_MODAL_TITLE_MAX_CHARS)
    write_string_field(buf, item['message'], CUTSCENE_MODAL_MESSAGE_MAX_CHARS)
    buf += struct.pack('<B', 0)  # v1: message-only, no options/callback

  elif item_type == 'MODAL_OPTIONS_MARKERS':
    write_string_field(buf, item.get('title', ''), CUTSCENE_MODAL_TITLE_MAX_CHARS)
    write_string_field(buf, item['message'], CUTSCENE_MODAL_MESSAGE_MAX_CHARS)
    options = item['options']
    if not (1 <= len(options) <= CUTSCENE_MODAL_OPTIONS_MARKERS_MAX):
      raise ValueError("MODAL_OPTIONS_MARKERS options must have 1 or 2 entries")
    buf += struct.pack('<B', len(options))
    for option in options:
      text_offset = pool.add_string(option['text'])
      marker_offset = pool.add_string(option['marker'])
      buf += struct.pack('<HH', text_offset, marker_offset)

  elif item_type in ('MODAL_CLOSE', 'RESTART', 'IDLE'):
    pass

  elif item_type == 'PRINT':
    write_string_field(buf, item['text'], CUTSCENE_PRINT_MAX_CHARS)

  elif item_type == 'MARKER':
    offset = pool.add_string(item['name'])
    buf += struct.pack('<H', offset)

  elif item_type == 'SCENE':
    buf += struct.pack('<B', SCENE_TYPE[item['sceneType'].upper()])

  elif item_type in ('SAVE_DEVICE_CHECK', 'SAVE_LOAD_ALL_SLOTS'):
    success_offset = pool.add_string(item['successMarker'])
    failure_offset = pool.add_string(item['failureMarker'])
    buf += struct.pack('<HH', success_offset, failure_offset)

  elif item_type == 'AUDIO_PLAY':
    write_string_field(buf, item['file'], AUDIO_PATH_MAX)
    buf += struct.pack('<B', AUDIO_CHANNEL[item['channel'].upper()])
    buf += struct.pack('<f', float(item.get('volume', 1.0)))
    buf += struct.pack('<f', float(item.get('pan', 0.0)))
    buf += struct.pack('<B', 1 if item.get('looping', False) else 0)
    buf += struct.pack('<B', int(item.get('loopCount', 0)))
    buf += struct.pack('<f', float(item.get('loopStart', -1.0)))
    buf += struct.pack('<f', float(item.get('loopTo', 0.0)))

  elif item_type in ('AUDIO_STOP', 'AUDIO_PAUSE', 'AUDIO_RESUME', 'AUDIO_FADE_WAIT'):
    buf += struct.pack('<B', AUDIO_CHANNEL[item['channel'].upper()])

  elif item_type == 'AUDIO_FADE':
    buf += struct.pack('<B', AUDIO_CHANNEL[item['channel'].upper()])
    buf += struct.pack('<f', float(item['from']))
    buf += struct.pack('<f', float(item['to']))
    buf += struct.pack('<f', float(item['duration']))
    buf += struct.pack('<B', EASING[item.get('easing', 'LINEAR').upper()])

  elif item_type == 'AUDIO_SET_PAN':
    buf += struct.pack('<B', AUDIO_CHANNEL[item['channel'].upper()])
    buf += struct.pack('<f', float(item['pan']))

  elif item_type == 'AUDIO_SET_LOOP':
    buf += struct.pack('<B', AUDIO_CHANNEL[item['channel'].upper()])
    buf += struct.pack('<B', 1 if item.get('looping', True) else 0)
    buf += struct.pack('<B', int(item.get('loopCount', 0)))
    buf += struct.pack('<f', float(item.get('loopStart', -1.0)))
    buf += struct.pack('<f', float(item.get('loopTo', 0.0)))

  elif item_type == 'AUDIO_SET':
    buf += struct.pack('<B', AUDIO_CHANNEL[item['channel'].upper()])
    buf += struct.pack('<f', float(item['pan']))
    buf += struct.pack('<B', 1 if item.get('looping', True) else 0)
    buf += struct.pack('<B', int(item.get('loopCount', 0)))
    buf += struct.pack('<f', float(item.get('loopStart', -1.0)))
    buf += struct.pack('<f', float(item.get('loopTo', 0.0)))

  else:
    raise ValueError(f"Unhandled cutscene item type: {item_type}")

  return bytes(buf)


def build_cutscene(source):
  pause_value = 0
  for name in source.get('pause', None) or []:
    pause_value |= PAUSE_FLAG[name.upper()]
  if 'pause' not in source:
    pause_value = PAUSE_DEFAULT

  items = source['items']
  if len(items) > 255:
    raise ValueError("Cutscene has more than 255 items")

  pool = Pool()
  item_stream = bytearray()
  for item in items:
    item_stream += encode_item(item, pool)

  header = bytearray()
  header += FILE_MAGIC
  header += struct.pack('<I', VERSION_OUT)
  header += struct.pack('<B', pause_value)
  header += struct.pack('<B', len(items))
  header += struct.pack('<H', len(pool.data))

  return bytes(header) + bytes(item_stream) + bytes(pool.data)


def strip_jsonc_comments(text):
  """Strips // line comments and /* */ block comments from JSONC text,
  leaving string literal contents (including any // or /* inside them)
  untouched."""
  result = []
  i = 0
  n = len(text)
  in_string = False
  escape = False

  while i < n:
    c = text[i]

    if in_string:
      result.append(c)
      if escape:
        escape = False
      elif c == '\\':
        escape = True
      elif c == '"':
        in_string = False
      i += 1
      continue

    if c == '"':
      in_string = True
      result.append(c)
      i += 1
      continue

    if c == '/' and i + 1 < n and text[i + 1] == '/':
      i += 2
      while i < n and text[i] != '\n':
        i += 1
      continue

    if c == '/' and i + 1 < n and text[i + 1] == '*':
      i += 2
      while i + 1 < n and not (text[i] == '*' and text[i + 1] == '/'):
        i += 1
      i += 2
      continue

    result.append(c)
    i += 1

  return ''.join(result)


def process_json(json_path):
  with open(json_path, 'r', encoding='utf-8') as f:
    text = f.read()
  source = json.loads(strip_jsonc_comments(text))

  data = build_cutscene(source)

  name = os.path.splitext(os.path.basename(json_path))[0]
  out_path = os.path.join(ASSETS_DIR, 'cutscenes', f'{name}.cts')
  os.makedirs(os.path.dirname(out_path), exist_ok=True)
  with open(out_path, 'wb') as f:
    f.write(data)

  print(f"{json_path} -> {out_path} ({len(data)} bytes)")


def main():
  parser = argparse.ArgumentParser(
    description="Generate DCTS binary cutscene files from raw JSONC "
                 "cutscene definitions"
  )
  parser.add_argument(
    'paths', nargs='*',
    help='JSONC cutscene file(s) to process. If omitted, processes all '
         'assetsraw/cutscenes/*.jsonc'
  )
  args = parser.parse_args()

  if not args.paths:
    cutscenes_dir = os.path.join(ASSETSRAW_DIR, 'cutscenes')
    if not os.path.isdir(cutscenes_dir):
      print(f"No directory found: {cutscenes_dir}")
      sys.exit(1)
    json_files = sorted(
      os.path.join(cutscenes_dir, f)
      for f in os.listdir(cutscenes_dir)
      if f.endswith('.jsonc')
    )
    if not json_files:
      print(f"No JSONC files found in {cutscenes_dir}")
      sys.exit(0)
    for p in json_files:
      process_json(p)
    return

  for p in args.paths:
    process_json(p)


if __name__ == '__main__':
  main()