Impulse Tracker sample decompressor in Godot

itsex.gd

# Impulse Tracker sample decompressor in Godot
# based on kb's decompressor
# cs127 / 2025-11-21

class_name ITSEx
extends RefCounted
## IT sample decompressor

@warning_ignore_start("integer_division")
@warning_ignore_start("shadowed_variable")

## Whether 8-bit samples should be written in unsigned format like in WAV files. Useful for directly writing the
## decompressed data to a WAV file. In most cases, even when writing to the data stream of an [AudioStreamWAV], this
## should be disabled.
static var wav_unsigned_8bit := false

## Whether 16-bit samples should be written in big-endian. In most cases, this should be disabled.
static var big_endian_16bit := false

class Block:
	var cmp_buf: PackedByteArray
	var cmp_bit_pos: int = 0
	var b16: bool
	var it215: bool
	var decmp_length: int
	var decmp_pos: int = 0
	var width: int
	var d1: int = 0
	var d2: int = 0
	var error: Error = Error.OK
	
	static func bit_mask(n: int) -> int: return (1 << n) - 1
	
	static func to_signed_byte(x: int) -> int:
		x &= 0xFF
		if x & 0x80: x -= 0x100
		return x
		
	static func to_signed_word(x: int) -> int:
		x &= 0xFFFF
		if x & 0x8000: x -= 0x10000
		return x
	
	func read_bits(n: int) -> int:
		var bits := 0
		
		if n == 0: return 0
		
		if n < 0 || n > 32:
			error = Error.ERR_PARAMETER_RANGE_ERROR
			return 0
		
		if cmp_bit_pos + n > cmp_buf.size() * 8:
			error = Error.ERR_FILE_EOF
			return 0
		
		var cmp_byte_pos := cmp_bit_pos >> 3
		var cmp_bit_pos_rel := cmp_bit_pos & 7
		for i in range((n + cmp_bit_pos_rel + 7) >> 3): bits |= cmp_buf[cmp_byte_pos + i] << (i * 8)
		bits >>= cmp_bit_pos_rel
		bits &= bit_mask(n)
		cmp_bit_pos += n
		
		return bits
	
	func decompress() -> PackedByteArray:
		var decmp := PackedByteArray()
		decmp.resize(decmp_length * (2 if b16 else 1))
		
		var bit_depth := 16 if b16 else 8
		var to_signed := to_signed_word if b16 else to_signed_byte
		
		while decmp_pos < decmp_length:
			
			var value := read_bits(width)
			
			if width <= 6: # method 1 (1..6 bits)
				if value == 1 << (width - 1):
					# read new width and expand
					value = read_bits(4 if b16 else 3) + 1
					width = value + (0 if value < width else 1)
					continue
			elif width <= bit_depth: # method 2 (7..8/16 bits)
				var all_bits := bit_mask(bit_depth)
				var border := (all_bits >> (bit_depth + 1 - width)) - (bit_depth / 2) & all_bits
				if value > border and value <= border + bit_depth:
					# convert and expand width
					value -= border
					width = value + (0 if value < width else 1)
					continue
			elif width == bit_depth + 1: # method 3 (9/17 bits)
				if value & (1 << bit_depth):
					# new width
					width = (value + 1) & 0xFF
					continue
			else: # invalid method
				error = Error.ERR_INVALID_DATA
				return []
			
			var shift: int = max(bit_depth - width, 0)
			value = to_signed.call(value << shift) >> shift
			
			d1 += value
			if it215: d2 += d1 # integrate a second time for 215
			
			var smp: int = to_signed.call(d2 if it215 else d1)
			
			if b16:
				smp &= 0xFFFF # convert to unsigned to store as bytes
				if ITSEx.big_endian_16bit: smp = (smp << 8) | (smp >> 8)
				decmp[decmp_pos * 2 + 0] = (smp >> 0) & 0xFF
				decmp[decmp_pos * 2 + 1] = (smp >> 8) & 0xFF
			else:
				if ITSEx.wav_unsigned_8bit: smp += 0x80 # convert to unsigned
				else: smp &= 0xFF                       # one way or another
				decmp[decmp_pos] = smp
			
			decmp_pos += 1
		
		return decmp
	
	func _init(src: StreamPeer, decmp_length_left: int, b16: bool, it215: bool) -> void:
		var read := src.get_data(src.get_u16())
		if read[0]:
			error = read[0]
			return
		cmp_buf = PackedByteArray(read[1])
		
		self.b16 = b16
		self.it215 = it215
		decmp_length = min(decmp_length_left, 0x4000 if b16 else 0x8000)
		width = 17 if b16 else 9

## Reads a compressed sample data stream from [param src], decompresses it, and writes it into [param dst].[br]
## Any existing data in [param dst] will be lost.[br][br]
##
## [param length] is the sample length, [param b16] is whether the sample is 16-bit, and [param it215] is whether the
## audio signal was differentiated twice rather than once during compression.[br][br]
##
## For stereo samples, this function must be called twice, once for each channel.
static func decompress(src: StreamPeer, dst: PackedByteArray, length: int, b16: bool, it215: bool) -> Error:
	src.big_endian = false
	dst.clear()
	
	while length:
		var block := Block.new(src, length, b16, it215)
		if block.error: return block.error
		
		var decmp := block.decompress()
		if not decmp: return block.error
		
		dst.append_array(decmp)
		length -= block.decmp_length
	
	return Error.OK

@warning_ignore_restore("integer_division")
@warning_ignore_restore("shadowed_variable")