feat(aiv): working roundtrip between aiv and json, needs adjustment, tile offsets are slightly off

Author: gynt

resources/aiv/empty.aiv

@@ -1,6 +1,7 @@
 import pathlib
 
-from .conversion import to_json
+from sourcehold.tool.convert.aiv.exports import to_json
+from sourcehold.tool.convert.aiv.imports import from_json
 
 
 def convert_aiv(args):
@@ -19,16 +20,32 @@ def convert_aiv(args):
   if not inp_format:
       if inp.endswith(".aiv"):
         inp_format = 'aiv'
+      elif inp.endswith(".json"):
+        inp_format = "json"
   out_format = args.to_format
-  if inp_format == "aiv":
-    out_format = "json"
+  if not out_format:
+    if inp_format == "aiv":
+      out_format = "json"
+    elif inp_format == "json":
+      out_format = "aiv"
   if args.output == None:
-    args.output = f"{pathlib.Path(inp).name}.json"
+    if out_format == "json":
+      args.output = "-"
+    elif out_format == "aiv":
+      args.output = f"{pathlib.Path(inp).name}.aiv"
+    #args.output = f"{pathlib.Path(inp).name}.json"
 
   if args.debug:
     print(f"converting '{inp_format}' file '{inp}' to '{out_format}' file '{args.output}'")
   if inp_format == 'aiv' and out_format == "json":
-    pathlib.Path(args.output).write_text(to_json(path = inp, include_extra=args.extra, report=args.debug))
+    conv = to_json(path = inp, include_extra=args.extra, report=args.debug)
+    if args.output == "-":
+      print(conv)
+    else:
+      pathlib.Path(args.output).write_text(conv)
+  elif inp_format == "json" and out_format == "aiv":
+    conv = from_json(path = inp, report=args.debug)
+    conv.to_file(args.output)
   else:
     raise NotImplementedError(f"combination of in-format '{inp_format}' and out-format '{out_format}' not implemented")
 

sourcehold/tool/convert/aiv/__init__.py

@@ -0,0 +1,175 @@
+import sys
+
+from sourcehold.tool.convert.aiv.info import *
+try:
+  import sourcehold
+except:
+  import os, pathlib
+  sys.path.insert(0, str(pathlib.Path(".").parent))
+
+import sourcehold.aivs
+from sourcehold.aivs.AIV import AIV
+
+# from matplotlib import pyplot
+
+import struct
+
+import numpy as np
+
+import json
+
+
+
+
+def to_json(aiv=None, path: str='', include_extra=False, invert_y=True, report=False):
+  if aiv == None and not path:
+    raise Exception()
+  if aiv == None:
+    aiv = AIV().from_file(path)
+  
+  if report:
+    print(f"INFO: aiv has version: {aiv.directory.version_number}")
+
+  select_all = np.ones((100,100), 'bool')
+
+  constructions = get_constructions_matrix()
+  constructions[select_all] = struct.unpack(CONSTRUCTIONS_STRUCT_FORMAT, aiv.directory[INDEX_CONSTRUCTIONS].get_data())
+
+  steps = get_steps_matrix()
+  steps[select_all] = struct.unpack(STEPS_STRUCT_FORMAT, aiv.directory[INDEX_STEPS].get_data())
+
+# This value can lie!
+  step_count = struct.unpack(STEP_COUNT_STRUCT_FORMAT, aiv.directory[INDEX_STEP_COUNT].get_data())[0]
+  step_count_max = steps.max().item()
+  if step_count < step_count_max:
+    print(f"WARNING: step count ({step_count}) is lower than max step value from steps matrix ({step_count_max}), taking max step value as truth", file=sys.stderr)
+    step_count = step_count_max
+
+  pauses_raw = aiv.directory[INDEX_PAUSES].get_data()
+  if len(pauses_raw) != 50 * 4:
+    print(f"WARNING: pauses is not expected size of {50*4} but {len(pauses_raw)}, adjusting", file=sys.stderr)
+    pauses = list(struct.unpack(PAUSES_STRUCT_FORMAT_10, pauses_raw))  
+  else:
+    pauses = list(struct.unpack(PAUSES_STRUCT_FORMAT_50, pauses_raw))  
+
+  units = get_units_matrix(list(struct.unpack(f"<{24*10}i", aiv.directory[INDEX_MISC].get_data())))
+
+  pauseDelayAmount = struct.unpack(PAUSE_DELAY_AMOUNT_STRUCT_FORMAT, aiv.directory[INDEX_PAUSE].get_data())[0]
+
+  output = {}
+  frames = [None] * step_count # + 1 is already stored in the aiv format
+  miscItems = []
+
+  output["frames"] = frames
+  output["miscItems"] = miscItems
+  output["pauseDelayAmount"] = pauseDelayAmount
+
+  processed = {}
+
+  buildings = 0
+  offset = -1
+  for i in x_range():
+    for j in y_range(invert_y=invert_y):
+      offset += 1
+      
+      if offset in processed:
+        continue
+
+      construction = constructions[i, j].item()
+      step = steps[i, j].item()
+
+      if construction == 0:
+        processed[offset] = True
+        continue
+
+      # Not in source code, but added here
+      elif construction == 1:
+        processed[offset] = True
+        continue
+      
+      elif construction == 2:
+        processed[offset] = True
+        continue
+      
+      elif construction == 38: #keep2
+        processed[offset] = True
+        continue
+      
+      buildingType = convertAIVEnumToMapperEnum(v = construction)
+      
+      if buildingType in [25, 46, 26, 35, 106, 99]: # [10, 11, 12, 13, 20, 21, 22, 23, 24]:
+        continue # not processed yet
+      
+      processed[offset] = True
+      
+      if step >= len(frames):
+        raise Exception(f"step size too high at ({i}, {j}, {offset}): {step} => {len(frames)}")
+      if frames[step] is None:
+        buildings += 1
+        shouldPause = step in pauses
+        frames[step] = {'itemType': buildingType, 'tilePositionOfsets': [offset], 'shouldPause': shouldPause}
+
+  if report:
+    print(f"INFO: buildings: {buildings}")
+
+  nonbuildings = 0
+
+  offset = -1
+  for i in x_range():
+    for j in y_range(invert_y=invert_y):
+      offset += 1
+      if offset not in processed:
+        # do all the special stuff
+        construction = constructions[i, j].item()
+        step = steps[i, j].item()
+        buildingType = convertAIVEnumToMapperEnum(v = construction)
+        shouldPause = step in pauses
+        if frames[step] is None:
+          frames[step] = {'itemType': buildingType, 'tilePositionOfsets': [], 'shouldPause': shouldPause}
+        frames[step]['tilePositionOfsets'].append(offset)
+        nonbuildings += 1
+  
+  if report:
+    print(f"INFO: walls | pitch | moat: {nonbuildings}")
+
+  misc = 0
+  for unitType in range(24):
+    for entry in range(10):
+      location = units[unitType, entry].item()
+      if location == 0:
+        continue
+      misc += 1
+      miscItems.append({'itemType': unitType, 'positionOfset': location, 'number': entry})
+
+  if report:
+    print(f"INFO: units | flags | brazier | misc: {misc}")
+
+  emptyFrames = [index for index, frame in enumerate(output['frames']) if not frame]
+  knownEmptyFrames = [0,1]
+  unexpectedEmptyFrames = list(index for index in emptyFrames if index not in knownEmptyFrames)
+  if len(unexpectedEmptyFrames) > 0:
+    print(f"WARNING: unexpected count of empty frames, expected one (the first), but there were: {len(unexpectedEmptyFrames)} more: {','.join(str(v) for v in unexpectedEmptyFrames)}", file=sys.stderr)
+  output['frames'] = [frame for frame in output['frames'] if frame]
+
+  if include_extra:
+    output['extra'] = {
+      'version' : aiv.directory.version_number,
+      'directory_size' : aiv.directory.directory_size,
+      'size' : aiv.directory.size,
+      'u1' : aiv.directory.directory_u1,
+      'u7' : aiv.directory.directory_u7,
+      'sections': [i for i in aiv.directory.section_indices if i],
+      # 'rng': aiv.directory[2003].get_data() # RNG
+      '2001': int.from_bytes(aiv.directory[2001].get_data(), 'little'),
+      '2002': int.from_bytes(aiv.directory[2001].get_data(), 'little'),
+      'recent_selected_step_in_editor': int.from_bytes(aiv.directory[2010].get_data(), 'little'), # Last selected step in editor?
+      'step_count': step_count,
+      # 'typeData': np.frombuffer(aiv.directory[2004].get_data(), dtype = 'uint8').reshape((100,100)),
+      # '2005': # wall edges or something?
+      # '2006' : texture noise
+      # '2013' : misc and unit locations in tilemap form?
+    }
+
+
+  return json.dumps(output, indent=2)
+

sourcehold/tool/convert/aiv/exports.py

@@ -0,0 +1,96 @@
+import json
+import struct
+from turtle import position
+from typing import Dict
+import numpy as np
+from sourcehold.aivs.AIV import AIV
+from sourcehold.tool.convert.aiv.info import AIV_HEIGHT, AIV_WIDTH, CONSTRUCTIONS_STRUCT_FORMAT, INDEX_CONSTRUCTIONS, INDEX_MISC, INDEX_PAUSE, INDEX_PAUSES, INDEX_STEP_COUNT, INDEX_STEPS, MAPPER_BUILDING_SIZES, MAPPERS_SH1_VK, PAUSES_STRUCT_FORMAT_50, STEPS_STRUCT_FORMAT, UNITS_STRUCT_FORMAT, convertMapperEnumToAIVEnum, get_constructions_matrix, get_steps_matrix, get_units_matrix, x_range, y_range
+
+mapping_xyinversed_offset = np.zeros(shape=(AIV_WIDTH, AIV_HEIGHT), dtype='uint32')
+mapping_xy_offset = np.zeros(shape=(AIV_WIDTH, AIV_HEIGHT), dtype='uint32')
+
+offset = -1
+for x in x_range():
+  for y in y_range(invert_y=False):
+    offset += 1
+    mapping_xy_offset[x, y] = offset
+
+offset = -1
+for x in x_range():
+  for y in y_range(invert_y=True):
+    offset += 1
+    mapping_xyinversed_offset[x, y] = offset
+
+def convert_offset(offset, invert_y=False):
+  return (offset // AIV_WIDTH, offset % AIV_HEIGHT if not invert_y else (AIV_HEIGHT - 1) - (offset % AIV_HEIGHT))
+
+def convert_offsets(offsets, invert_y=False):
+  for loc in offsets:
+    yield convert_offset(loc, invert_y=invert_y)
+
+def from_json(path = None, data: Dict | None = None, invert_y = True, report = False):
+  if not path and not data:
+    raise Exception()
+  
+  if path and not data:
+    with open(path, 'rb') as f:
+      data = json.load(f)
+  
+  if data == None:
+    raise Exception()
+  
+  pauses = []
+  pauseDelayAmount = data['pauseDelayAmount']
+  constructions = get_constructions_matrix()
+  steps = get_steps_matrix()
+  frames = data['frames']
+  units = get_units_matrix()
+  for stepMin1, frame in enumerate(frames):
+    step = stepMin1 + 1
+    shouldPause = frame['shouldPause']
+    if shouldPause:
+      pauses.append(step)
+    mapperID = frame['itemType']
+    mapperName = MAPPERS_SH1_VK[mapperID]
+
+    if mapperID in [25, 46, 26, 35, 106, 99]:
+      # Special type
+      aivID = convertMapperEnumToAIVEnum(v = mapperID)
+      locations = frame['tilePositionOfsets']
+      for x, y in convert_offsets(locations, invert_y=invert_y):
+        constructions[x, y] = aivID
+        steps[x, y] = step
+    else:
+      # Buildings  
+      aivID = convertMapperEnumToAIVEnum(v = mapperID)
+      if not mapperName in MAPPER_BUILDING_SIZES:
+        raise Exception(f"unsupported mapper size: {mapperName}")
+      size = MAPPER_BUILDING_SIZES[mapperName]
+      if len(frame['tilePositionOfsets']) > 1:
+        raise Exception(f'too many locations for a building: {frame['tilePositionOfsets']}')
+      location = frame['tilePositionOfsets'][0]
+      x, y = convert_offset(location, invert_y=invert_y)
+      constructions[x:(x+size),y:(y+size)] = aivID
+      steps[x:(x+size),y:(y+size)] = step
+  for miscItem in data['miscItems']:
+    itemType = miscItem['itemType']
+    number = miscItem['number']
+    positionOffset = miscItem['positionOfset']
+    units[itemType, number] = positionOffset
+
+  pauses = sorted(pauses)
+  if len(pauses) < 50:
+    pauses += [0] * (50 - len(pauses))
+  
+  select_all = np.ones((100,100), 'bool')
+  aiv = AIV().from_file("resources/aiv/empty.aiv")
+  aiv.directory[INDEX_CONSTRUCTIONS].set_data(struct.pack(CONSTRUCTIONS_STRUCT_FORMAT, *constructions[select_all]))
+  aiv.directory[INDEX_STEPS].set_data(struct.pack(STEPS_STRUCT_FORMAT, *steps[select_all]))
+  aiv.directory[INDEX_STEP_COUNT].set_data(struct.pack("<i", len(frames) + 1))
+  aiv.directory[INDEX_PAUSES].set_data(struct.pack(PAUSES_STRUCT_FORMAT_50, *pauses))
+  aiv.directory[INDEX_PAUSE].set_data(struct.pack("<i", pauseDelayAmount))
+  select_all_misc = np.ones((24, 10), 'bool')
+  aiv.directory[INDEX_MISC].set_data(struct.pack(UNITS_STRUCT_FORMAT, *units[select_all_misc]))
+  return aiv
+  
+