generalise python wave writer

.gitignore

@@ -10,4 +10,8 @@ neotools/**/*.pcm
 neotools/**/*.vgm
 neotools/**/*.vgz
 
+spctools/it
+spctools/spc
+spctools/sample
+
 .DS_Store

common/riffwriter.py

@@ -0,0 +1,37 @@
+from abc import ABC, abstractmethod
+from typing import BinaryIO, List
+
+
+class AbstractRiffChunk(ABC):
+	@abstractmethod
+	def fourcc(self) -> bytes: raise NotImplementedError
+
+	@abstractmethod
+	def size(self) -> int: raise NotImplementedError
+
+	@abstractmethod
+	def write(self, f: BinaryIO): raise NotImplementedError
+
+
+class RiffFile(AbstractRiffChunk):
+	def fourcc(self) -> bytes: return b"RIFF"
+
+	def size(self) -> int: return 4 + sum(8 + c.size() for c in self._chunks)
+
+	def __init__(self, type: bytes, chunks: List[AbstractRiffChunk]):
+		self._chunks = chunks
+		if len(type) != 4: raise ValueError
+		self._type = type
+
+	def write(self, f: BinaryIO):
+		f.writelines([
+			self.fourcc(),
+			self.size().to_bytes(4, "little", signed=False),
+			self._type])
+		for chunk in self._chunks:
+			size = chunk.size()
+			if size & 0x3: raise AssertionError("Unaligned chunks will produce malformed riff files")
+			f.writelines([
+				chunk.fourcc(),
+				size.to_bytes(4, "little", signed=False)])
+			chunk.write(f)

common/wavesampler.py

@@ -0,0 +1,107 @@
+import struct
+from enum import Enum
+from riffwriter import AbstractRiffChunk
+from typing import BinaryIO, List
+
+
+class WaveSamplerSMPTEOffset:
+	def __init__(self, hours: int=0, minutes: int=0, seconds: int=0, frames: int=0):
+		if -23 > hours > 23:  raise ValueError("Hours out of range")
+		if 0 > minutes > 59:  raise ValueError("Minutes out of range")
+		if 0 > seconds > 59:  raise ValueError("Seconds out of range")
+		if 0 > frames > 0xFF: raise ValueError("Frames out of range")
+		self._hours   = hours
+		self._minutes = minutes
+		self._seconds = seconds
+		self._frames  = frames
+
+	def hours(self) -> int:   return self._hours
+	def minutes(self) -> int: return self._minutes
+	def seconds(self) -> int: return self._seconds
+	def frames(self) -> int:  return self._frames
+
+	def pack(self) -> bytes:
+		#FIXME: endianess??
+		return struct.pack("<bBBB", self._hours, self._minutes, self._seconds, self._frames)
+
+
+class WaveSamplerLoopType(Enum):
+	FORWARD = 0
+	BIDIRECTIONAL = 1
+	REVERSE = 2
+
+
+class WaveSamplerLoop:
+	def __init__(self,
+ 			cueId: int=0,
+			type: int|WaveSamplerLoopType=0,
+			start: int=0,
+			end: int=0,
+			fraction: int=0,
+			loopCount: int=0):
+		self._cueId = cueId
+		self._type = type.value if type is WaveSamplerLoopType else type
+		self._start = start
+		self._end = end
+		self._fraction = fraction
+		self._loopCount = loopCount
+
+	def pack(self) -> bytes:
+		return struct.pack("<IIIIII",
+			self._cueId,      # Cue point ID
+			self._type,       # Loop type
+			self._start,      # Loop start
+			self._end,        # Loop end
+			self._fraction,   # Fraction (none)
+			self._loopCount)  # Loop count (infinite)
+
+
+class WaveSamplerChunk(AbstractRiffChunk):
+	def fourcc(self) -> bytes: return b"smpl"
+
+	def loopsSize(self) -> int: return len(self._loops) * 24
+
+	def size(self) -> int: return 36 + self.loopsSize()
+
+	def write(self, f: BinaryIO):
+		#TODO: unused data dummied out for now
+		f.write(struct.pack("<4sIiiii4sII",
+			self._manufacturer,         # MMA Manufacturer code
+			self._product,              # Product
+			self._period,               # Playback period (ns)
+			self._unityNote,            # MIDI unity note
+			self._fineTune,             # MIDI pitch fraction
+			self._smpteFormat,          # SMPTE format
+			self._smpteOffset.pack(),   # SMPTE offset
+
+			len(self._loops),           # Number of loops
+			self.loopsSize()))          # Loop data length
+		f.writelines(loop.pack() for loop in self._loops)
+
+	def __init__(self,
+			manufacturer: bytes|None=None,
+			product: int=0,
+			period: int=0,
+			midiUnityNote: int=0,
+			midiPitchFraction: int=0,
+			smpteFormat: int=0,
+			smpteOffset: WaveSamplerSMPTEOffset|None=None,
+			loops: List[WaveSamplerLoop]=None):
+		if manufacturer is not None:
+			if len(manufacturer) not in [1, 3]: raise ValueError("Malformed MIDI manufacturer code")
+			self._manufacturer = len(manufacturer).to_bytes(1, byteorder="little", signed=False)
+			self._manufacturer += manufacturer.rjust(3, b"\x00")
+		else:
+			self._manufacturer = b"\x00" * 4
+
+		if 0 > product > 0xFFFF: raise ValueError("Product code out of range")
+		self._product = product # Arbitrary vendor specific product code, dunno if this should be signed or unsigned
+		self._period = period # Sample period in ns, (1 / samplerate) * 10^9, who cares
+		if 0 > midiUnityNote > 127: raise ValueError("MIDI Unity note out of range")
+		self._unityNote = midiUnityNote # MIDI note that plays the sample unpitched, middle C=60
+		self._fineTune = midiPitchFraction # Finetune fraction, 256 == 100 cents
+		if smpteFormat not in [0, 24, 25, 29, 30]: raise ValueError("Invalid SMPTE format")
+		self._smpteFormat = smpteFormat
+		self._smpteOffset = smpteOffset if (smpteOffset and smpteFormat > 0) else WaveSamplerSMPTEOffset()
+		if self._smpteOffset.frames() > self._smpteFormat: raise ValueError("SMPTE frame offset can't exceed SMPTE format")
+		self._loops = loops if loops is not None else list()

common/wavewriter.py

@@ -0,0 +1,95 @@
+import struct
+from abc import abstractmethod
+from enum import Enum
+from typing import BinaryIO, List
+from riffwriter import RiffFile, AbstractRiffChunk
+
+
+class WaveSampleFormat(Enum):
+	PCM        = 0x0001
+	IEEE_FLOAT = 0x0003
+	ALAW       = 0x0006
+	MULAW      = 0x0007
+	EXTENSIBLE = 0xFFFE
+
+
+class WaveAbstractFormatChunk(AbstractRiffChunk):
+	def fourcc(self) -> bytes: return b"fmt "
+
+	def size(self) -> int: return 16
+
+	@abstractmethod
+	def sampleformat(self) -> WaveSampleFormat: raise NotImplementedError
+
+	@abstractmethod
+	def channels(self) -> int: raise NotImplementedError
+
+	@abstractmethod
+	def samplerate(self) -> int: raise NotImplementedError
+
+	@abstractmethod
+	def byterate(self) -> int: raise NotImplementedError
+
+	@abstractmethod
+	def align(self) -> int: raise NotImplementedError
+
+	@abstractmethod
+	def bitdepth(self) -> int: raise NotImplementedError
+
+	def write(self, f: BinaryIO):
+		f.write(struct.pack("<HHIIHH",
+			self.sampleformat().value,
+				self.channels(),
+				self.samplerate(),
+				self.byterate(),
+				self.align(),
+				self.bitdepth()))
+
+
+class WavePcmFormatChunk(WaveAbstractFormatChunk):
+	def sampleformat(self) -> WaveSampleFormat: return WaveSampleFormat.PCM
+
+	def channels(self) -> int: return self._channels
+
+	def samplerate(self) -> int: return self._samplerate
+
+	def byterate(self) -> int: return self._samplerate * self._channels * self._bytedepth
+
+	def align(self) -> int: return self._channels * self._bytedepth
+
+	def bitdepth(self) -> int: return self._bytedepth * 8
+
+	def __init__(self, channels: int, samplerate: int, bitdepth: int):
+		if channels < 0 or channels >= 256: raise ValueError
+		if samplerate < 1 or samplerate > 0xFFFFFFFF: raise ValueError
+		if bitdepth not in [8, 16, 32]: raise ValueError
+		self._channels = channels
+		self._samplerate = samplerate
+		self._bytedepth = bitdepth // 8
+
+
+class WaveDataChunk(AbstractRiffChunk):
+	def fourcc(self) -> bytes: return b"data"
+
+	def size(self) -> int: return len(self._data)
+
+	def write(self, f: BinaryIO): f.write(self._data)
+
+	def __init__(self, data: bytes):
+		self._data = data
+
+
+class WaveCommentChunk(AbstractRiffChunk):
+	def fourcc(self) -> bytes: return b"clm "
+
+	def size(self) -> int: return len(self._comment)
+
+	def write(self, f: BinaryIO): f.write(self._comment)
+
+	def __init__(self, comment: bytes):
+		self._comment = comment
+
+
+class WaveFile(RiffFile):
+	def __init__(self, format: WaveAbstractFormatChunk, chunks: List[AbstractRiffChunk]):
+		super().__init__(b"WAVE", [format] + chunks)

spctools/ripsamples.py

@@ -1,21 +1,29 @@
 #!/usr/bin/env python3
 
 # ripsamples.py -- a python script for mass extracting samples from SPC files.
-# (C) 2018 neoadpcmextract.c (C) 2018 a dinosaur (zlib)
+# (C) 2018, 2023 a dinosaur (zlib)
 
 import os
 import subprocess
 import pathlib
 import struct
 import hashlib
+from typing import BinaryIO
+from io import BytesIO
 
-# Directory constants.
+import sys
+sys.path.append("../common")
+from wavewriter import WaveFile, WavePcmFormatChunk, WaveDataChunk
+from wavesampler import WaveSamplerChunk, WaveSamplerLoop
+
+
+# Directory constants
 SPCDIR = "./spc"
 ITDIR  = "./it"
 SMPDIR = "./sample"
 
-# External programs used by this script.
+# External programs used by this script
-SPC2IT = "spc2it"
+SPC2IT = "spc2it/spc2it"
 
 
 class Sample:
@@ -24,63 +32,35 @@ class Sample:
 	loopEnd = 0
 	rate    = 0
 
-	data = None
+	data: bytes = None
 
-def writesmp(smp, path):
+
+def writesmp(smp: Sample, path: str):
+	print(path)
 	with open(path, "wb") as wav:
-
+		# Make sure sample rate is nonzero
-		# Make sure sample rate is nonzero.
 		#TODO: figure out why this even happens...
 		if smp.rate == 0:
 			smp.rate = 32000
-			#print(path + " may be corrupted...")
+			#print(path + " may be corrupted")
 
-		writeLoop = True if smp.loopEnd > smp.loopBeg else False
+		fmtChunk = WavePcmFormatChunk(  # Audio format (uncompressed)
+			1,                          # Channel count (mono)
+			smp.rate,                   # Samplerate
+			16)                         # Bits per sample (16 bit)
+		dataChunk = WaveDataChunk(smp.data)
+		loopChunk = None
+		if smp.loopEnd > smp.loopBeg:
+			loopChunk = WaveSamplerChunk(loops=[WaveSamplerLoop(
+				start=smp.loopBeg,  # Loop start
+				end=smp.loopEnd)])  # Loop end
 
-		# Write RIFF chunk.
+		WaveFile(fmtChunk,
-		wav.write(b"RIFF")
+			[dataChunk] if loopChunk is None else [loopChunk, dataChunk]
-		# Size of entire file following
+		).write(wav)
-		riffSize = 104 if writeLoop else 36
-		wav.write(struct.pack("<I", riffSize + smp.length * 2))
-		wav.write(b"WAVE")
 
-		# Write fmt chunk.
-		wav.write(b"fmt ")
-		wav.write(struct.pack("<I", 16))            # Subchunk size.
-		wav.write(struct.pack("<H", 1))             # Audio format (uncompressed)
-		wav.write(struct.pack("<H", 1))             # Channel count (mono)
-		wav.write(struct.pack("<I", smp.rate))      # Samplerate
-		wav.write(struct.pack("<I", smp.rate * 2 )) # Byte rate (16 bit mono)
-		wav.write(struct.pack("<H", 2))             # Bytes per sample (16 bit mono)
-		wav.write(struct.pack("<H", 16))            # Bits per sample (16 bit)
 
-		# Write sampler chunk (if looped).
+def readsmp(f: BinaryIO, ofs: int, idx: int):
-		if writeLoop:
-			wav.write(b"smpl")
-			wav.write(struct.pack("<I", 60))    # Chunk size (36 + loops * 24)
-			wav.write(b"\x00\x00\x00\x00")      # Manufacturer
-			wav.write(b"\x00\x00\x00\x00")      # Product
-			wav.write(b"\x00\x00\x00\x00")      # Sample period
-			wav.write(b"\x00\x00\x00\x00")      # MIDI unity note
-			wav.write(b"\x00\x00\x00\x00")      # MIDI pitch fraction
-			wav.write(b"\x00\x00\x00\x00")      # SMPTE format
-			wav.write(b"\x00\x00\x00\x00")      # SMPTE offset
-			wav.write(struct.pack("<I", 1))     # Loop count
-			wav.write(struct.pack("<I", 24))    # Loop data length
-
-			wav.write(struct.pack("<I", 0))           # Cue point ID (none)
-			wav.write(struct.pack("<I", 0))           # Loop type (forward)
-			wav.write(struct.pack("<I", smp.loopBeg)) # Loop start
-			wav.write(struct.pack("<I", smp.loopEnd)) # Loop end
-			wav.write(struct.pack("<I", 0))           # Fraction (none)
-			wav.write(struct.pack("<I", 0))           # Loop count (infinite)
-
-		# Write data chunk.
-		wav.write(b"data")
-		wav.write(struct.pack("<I", smp.length * 2))
-		wav.write(smp.data)
-
-def readsmp(f, ofs, idx):
 	# List of assumptions made:
 	# - Samples are 16 bit
 	# - Samples are mono
@@ -93,34 +73,34 @@ def readsmp(f, ofs, idx):
 	f.seek(ofs)
 	if f.read(4) != b"IMPS": return None
 
-	# Skip fname to flags & read.
+	# Skip fname to flags & read
 	f.seek(ofs + 0x12)
 	flags = int.from_bytes(f.read(1), byteorder="little", signed=False)
 
 	# Read flag values.
-	if not flags & 0b00000001: return None # Check sample data bit.
+	if not flags & 0b00000001: return None # Check sample data bit
 	loopBit = True if flags & 0b00010000 else False
 
 	smp = Sample()
 
-	# Read the rest of the header.
+	# Read the rest of the header
 	f.seek(ofs + 0x30)
 	smp.length = int.from_bytes(f.read(4), byteorder="little", signed=False)
 	if loopBit:
 		smp.loopBeg = int.from_bytes(f.read(4), byteorder="little", signed=False)
 		smp.loopEnd = int.from_bytes(f.read(4), byteorder="little", signed=False)
 	else:
-		f.seek(8, 1) # Skip over.
+		f.seek(8, 1) # Skip over
 		smp.loopBeg = 0
 		smp.loopEnd = 0
 	smp.rate = int.from_bytes(f.read(4), byteorder="little", signed=False)
-	f.seek(8, 1) # Skip over sustain shit.
+	f.seek(8, 1) # Skip over sustain shit
 
-	# Read sample data.
+	# Read sample data
 	dataOfs = int.from_bytes(f.read(4), byteorder="little", signed=False)
 	smp.data = f.read(smp.length * 2)
 
-	# Compute hash of data.
+	# Compute hash of data
 	#FIXME: This actually generates a butt ton of collisions...
 	# there's got to be a better way!
 	h = hashlib.md5(struct.pack("<pII", smp.data, smp.loopBeg, smp.loopEnd))
@@ -128,15 +108,16 @@ def readsmp(f, ofs, idx):
 
 	return smp
 
-def readit(path, outpath):
+
+def readit(path: str, outpath: str):
 	with open(path, "r+b") as f:
 
-		# Don't bother scanning non IT files.
+		# Don't bother scanning non IT files
 		if f.read(4) != b"IMPM": return
 
 		#print("Song name: " + f.read(26).decode('utf-8'))
 
-		# Need order list size and num instruments to know how far to skip.
+		# Need order list size and num instruments to know how far to skip
 		f.seek(0x20)
 		ordNum = int.from_bytes(f.read(2), byteorder="little", signed=False)
 		insNum = int.from_bytes(f.read(2), byteorder="little", signed=False)
@@ -161,7 +142,8 @@ def readit(path, outpath):
 					pathlib.Path(outpath).mkdir(parents=True, exist_ok=True)
 					writesmp(smp, outwav)
 
-def scanit(srcPath, dstPath):
+
+def scanit(srcPath: str, dstPath: str):
 	for directory, subdirectories, files in os.walk(srcPath):
 		for file in files:
 			if file.endswith(".it"):
@@ -169,18 +151,19 @@ def scanit(srcPath, dstPath):
 				outpath = dstPath + path[len(srcPath):-len(file)]
 				readit(path, outpath)
 
-def scanspc(srcPath, dstPath):
+
+def scanspc(srcPath: str, dstPath: str):
 	for directory, subdirectories, files in os.walk(srcPath):
 
-		# Create output dir for each game.
+		# Create output dir for each game
 		for sub in subdirectories:
 			path = os.path.join(dstPath, sub)
 			pathlib.Path(path).mkdir(parents=True, exist_ok=True)
 
-		# Convert spc files.
+		# Convert spc files
 		for file in files:
 			if file.endswith(".spc"):
-				# Don't convert files that have already been converted.
+				# Don't convert files that have already been converted
 				itpath = os.path.join(dstPath + directory[len(srcPath):], file[:-3] + "it")
 				if not os.path.isfile(itpath):
 					path = os.path.join(directory, file)
@@ -190,6 +173,7 @@ def scanspc(srcPath, dstPath):
 						os.rename(path, itpath)
 
 
-# Actual main stuff.
+# Actual main stuff
-scanspc(SPCDIR, ITDIR)
+if __name__ == "__main__":
-scanit(ITDIR, SMPDIR)
+	scanspc(SPCDIR, ITDIR)
+	scanit(ITDIR, SMPDIR)