Current section
Files
Jump to
Current section
Files
lib/ksc/stream.ex
defmodule Ksc.Stream do
@moduledoc "Lightweight runtime for Kaitai Struct generated parsers."
import Bitwise
@doc "Parse items repeatedly until binary is exhausted."
def repeat_eos(data, parse_fn) do
repeat_eos_acc(data, parse_fn, [])
end
defp repeat_eos_acc(<<>>, _parse_fn, acc), do: {Enum.reverse(acc), <<>>}
defp repeat_eos_acc(data, parse_fn, acc) do
{item, rest} = parse_fn.(data)
repeat_eos_acc(rest, parse_fn, [item | acc])
end
@doc "Parse items repeatedly until condition is met."
def repeat_until(data, parse_fn, until_fn) do
repeat_until_acc(data, parse_fn, until_fn, [])
end
defp repeat_until_acc(data, parse_fn, until_fn, acc) do
{item, rest} = parse_fn.(data)
new_acc = [item | acc]
if until_fn.(item, new_acc) do
{Enum.reverse(new_acc), rest}
else
repeat_until_acc(rest, parse_fn, until_fn, new_acc)
end
end
@doc "Parse items repeatedly until binary is exhausted, with index tracking."
def repeat_eos_idx(data, parse_fn) do
repeat_eos_idx_acc(data, parse_fn, [], 0)
end
defp repeat_eos_idx_acc(<<>>, _parse_fn, acc, _idx), do: {Enum.reverse(acc), <<>>}
defp repeat_eos_idx_acc(data, parse_fn, acc, idx) do
{item, rest} = parse_fn.(data, idx)
repeat_eos_idx_acc(rest, parse_fn, [item | acc], idx + 1)
end
@doc "Parse items repeatedly until condition is met, with index tracking."
def repeat_until_idx(data, parse_fn, until_fn) do
repeat_until_idx_acc(data, parse_fn, until_fn, [], 0)
end
defp repeat_until_idx_acc(data, parse_fn, until_fn, acc, idx) do
{item, rest} = parse_fn.(data, idx)
new_acc = [item | acc]
if until_fn.(item, new_acc) do
{Enum.reverse(new_acc), rest}
else
repeat_until_idx_acc(rest, parse_fn, until_fn, new_acc, idx + 1)
end
end
@doc "Parse items until parse fn signals done (returns {item, rest, true})."
def repeat_until_check(data, parse_fn) do
repeat_until_check_acc(data, parse_fn, [])
end
defp repeat_until_check_acc(data, parse_fn, acc) do
{item, rest, done} = parse_fn.(data)
new_acc = [item | acc]
if done,
do: {Enum.reverse(new_acc), rest},
else: repeat_until_check_acc(rest, parse_fn, new_acc)
end
@doc "Parse items until parse fn signals done, with index tracking."
def repeat_until_check_idx(data, parse_fn) do
repeat_until_check_idx_acc(data, parse_fn, [], 0)
end
defp repeat_until_check_idx_acc(data, parse_fn, acc, idx) do
{item, rest, done} = parse_fn.(data, idx)
new_acc = [item | acc]
if done,
do: {Enum.reverse(new_acc), rest},
else: repeat_until_check_idx_acc(rest, parse_fn, new_acc, idx + 1)
end
@doc "Parse bit items repeatedly until binary is exhausted."
def repeat_eos_bits(data, num_bits, bit_fn) do
bits_state = {0, 0, data}
repeat_eos_bits_acc(bits_state, num_bits, bit_fn, [])
end
defp repeat_eos_bits_acc({_, 0, <<>>}, _num_bits, _bit_fn, acc), do: {Enum.reverse(acc), <<>>}
defp repeat_eos_bits_acc({_bits_acc, bits_left, data} = state, num_bits, bit_fn, acc) do
# Check if we have enough bits left to read
total_bits = bits_left + byte_size(data) * 8
if total_bits < num_bits do
{Enum.reverse(acc), align_to_byte(state)}
else
{item, new_state} = bit_fn.(state, num_bits)
repeat_eos_bits_acc(new_state, num_bits, bit_fn, [item | acc])
end
end
@doc "Strip trailing pad bytes from binary."
def strip_pad_right(data, pad_byte) when is_binary(data) and is_integer(pad_byte) do
do_strip_pad_right(data, byte_size(data), pad_byte)
end
defp do_strip_pad_right(_data, 0, _pad_byte), do: <<>>
defp do_strip_pad_right(data, pos, pad_byte) do
if :binary.at(data, pos - 1) == pad_byte do
do_strip_pad_right(data, pos - 1, pad_byte)
else
binary_part(data, 0, pos)
end
end
@doc "Terminate binary at first occurrence of byte."
def terminate_at(data, term_byte, include \\ false) do
case :binary.match(data, <<term_byte>>) do
{pos, 1} ->
if include do
binary_part(data, 0, pos + 1)
else
binary_part(data, 0, pos)
end
:nomatch ->
data
end
end
@doc "Terminate at terminator, then strip pad bytes only when terminator was NOT found."
def terminate_and_pad(data, term_byte, include, pad_byte) do
if :binary.match(data, <<term_byte>>) == :nomatch do
strip_pad_right(data, pad_byte)
else
terminate_at(data, term_byte, include)
end
end
@doc "Read a null-terminated string from binary, returning {string, rest}."
def read_strz(data, _encoding \\ "UTF-8") do
case :binary.match(data, <<0>>) do
{pos, 1} ->
str = binary_part(data, 0, pos)
rest = binary_part(data, pos + 1, byte_size(data) - pos - 1)
{str, rest}
:nomatch ->
{data, <<>>}
end
end
@doc "Read a null-terminated string with consume control."
def read_strz_consume(data, consume) do
case :binary.match(data, <<0>>) do
{pos, 1} ->
str = binary_part(data, 0, pos)
rest =
if consume do
binary_part(data, pos + 1, byte_size(data) - pos - 1)
else
binary_part(data, pos, byte_size(data) - pos)
end
{str, rest}
:nomatch ->
{data, <<>>}
end
end
@doc """
Read bytes from binary until a terminator byte is found.
Returns {bytes_read, rest_of_binary}.
- consume: if true (default), the terminator byte is consumed from the stream
- include: if true, the terminator byte is included in the returned value
"""
def read_terminated(data, term_byte, consume \\ true, include \\ false) do
case :binary.match(data, <<term_byte>>) do
{pos, 1} ->
result = if include, do: binary_part(data, 0, pos + 1), else: binary_part(data, 0, pos)
rest_start = if consume, do: pos + 1, else: pos
rest = binary_part(data, rest_start, byte_size(data) - rest_start)
{result, rest}
:nomatch ->
{data, <<>>}
end
end
@doc """
Read N bits from a binary, big-endian bit order.
Returns {value, rest_binary}.
"""
def read_bits(data, num_bits) when is_binary(data) and is_integer(num_bits) do
read_bits_be(data, num_bits)
end
@doc """
Read N bits in big-endian bit order from a {bits_remaining, bit_count, binary} tuple or binary.
Returns {value, {bits_remaining, bit_count, rest_binary}}.
This supports consecutive bit reads without byte-realignment.
"""
def read_bits_be(data, num_bits) when is_binary(data) do
read_bits_be({0, 0, data}, num_bits)
end
def read_bits_be({bits_acc, bits_left, data}, num_bits) do
{bits_acc, bits_left, data} = ensure_bits(bits_acc, bits_left, data, num_bits)
# Extract top num_bits from bits_acc
shift = bits_left - num_bits
value = bsr(bits_acc, shift) |> band(bsl(1, num_bits) - 1)
# Remove those bits from accumulator
remaining_bits = bits_left - num_bits
remaining_acc = band(bits_acc, bsl(1, remaining_bits) - 1)
{value, {remaining_acc, remaining_bits, data}}
end
@doc """
Read N bits in little-endian bit order.
"""
def read_bits_le(data, num_bits) when is_binary(data) do
read_bits_le({0, 0, data}, num_bits)
end
def read_bits_le({bits_acc, bits_left, data}, num_bits) do
{bits_acc, bits_left, data} = ensure_bits_le(bits_acc, bits_left, data, num_bits)
# Extract bottom num_bits from bits_acc
value = band(bits_acc, bsl(1, num_bits) - 1)
remaining_acc = bsr(bits_acc, num_bits)
remaining_bits = bits_left - num_bits
{value, {remaining_acc, remaining_bits, data}}
end
@doc "Align bit state back to byte boundary, returning binary."
def align_to_byte({_bits_acc, _bits_left, data}), do: data
def align_to_byte(data) when is_binary(data), do: data
defp ensure_bits(bits_acc, bits_left, data, num_bits) do
if bits_left >= num_bits do
{bits_acc, bits_left, data}
else
bytes_needed = div(num_bits - bits_left + 7, 8)
<<new_bytes::binary-size(bytes_needed), rest::binary>> = data
new_bits = accum_bits_be(new_bytes, bits_acc)
{new_bits, bits_left + bytes_needed * 8, rest}
end
end
defp accum_bits_be(<<b, rest::binary>>, acc), do: accum_bits_be(rest, bor(bsl(acc, 8), b))
defp accum_bits_be(<<>>, acc), do: acc
defp ensure_bits_le(bits_acc, bits_left, data, num_bits) do
if bits_left >= num_bits do
{bits_acc, bits_left, data}
else
bytes_needed = div(num_bits - bits_left + 7, 8)
<<new_bytes::binary-size(bytes_needed), rest::binary>> = data
new_bits = accum_bits_le(new_bytes, bits_acc, bits_left)
{new_bits, bits_left + bytes_needed * 8, rest}
end
end
defp accum_bits_le(<<b, rest::binary>>, acc, shift),
do: accum_bits_le(rest, bor(acc, bsl(b, shift)), shift + 8)
defp accum_bits_le(<<>>, acc, _shift), do: acc
@doc "Floor division (Python-style: result rounds towards negative infinity)."
def floor_div(a, b) when is_integer(a) and is_integer(b) do
d = div(a, b)
r = rem(a, b)
if r != 0 and bxor(r, b) < 0, do: d - 1, else: d
end
def floor_div(a, b), do: floor_div(trunc(a), trunc(b))
@doc "Floor modulo (Python-style: result has same sign as divisor)."
def floor_mod(a, b) when is_integer(a) and is_integer(b) do
r = rem(a, b)
if r != 0 and bxor(r, b) < 0, do: r + b, else: r
end
def floor_mod(a, b), do: floor_mod(trunc(a), trunc(b))
@doc "Length of a string (character count) or size of a list/binary."
def kaitai_length(bin) when is_binary(bin), do: String.length(bin)
def kaitai_length(list) when is_list(list), do: length(list)
def kaitai_length(nil), do: 0
@doc "Get size of a list or byte_size of a binary."
def kaitai_size(nil), do: 0
def kaitai_size(bin) when is_binary(bin), do: byte_size(bin)
def kaitai_size(list) when is_list(list), do: length(list)
def kaitai_size(%{_io_size: size}), do: size
def kaitai_size(%{} = map), do: map_size(map)
@doc "Get the sizeof a parsed type or field. Uses stored _sizeof metadata."
def kaitai_sizeof(%{_sizeof: size}), do: size
def kaitai_sizeof(_), do: 0
@doc "Get the IO stream size for a parsed object. Uses stored _io_size metadata."
def kaitai_io_size(%{_io_size: size}), do: size
def kaitai_io_size(_), do: 0
@doc "Get minimum value from a list or binary (treating bytes as values)."
def kaitai_min(<<first, rest::binary>>), do: do_bin_min(rest, first)
def kaitai_min(list) when is_list(list), do: Enum.min(list)
defp do_bin_min(<<b, rest::binary>>, m) when b < m, do: do_bin_min(rest, b)
defp do_bin_min(<<_, rest::binary>>, m), do: do_bin_min(rest, m)
defp do_bin_min(<<>>, m), do: m
@doc "Get maximum value from a list or binary."
def kaitai_max(<<first, rest::binary>>), do: do_bin_max(rest, first)
def kaitai_max(list) when is_list(list), do: Enum.max(list)
defp do_bin_max(<<b, rest::binary>>, m) when b > m, do: do_bin_max(rest, b)
defp do_bin_max(<<_, rest::binary>>, m), do: do_bin_max(rest, m)
defp do_bin_max(<<>>, m), do: m
@doc "Access element at index (supports both lists and binaries)."
def kaitai_at(bin, idx) when is_binary(bin), do: :binary.at(bin, idx)
def kaitai_at(list, idx) when is_list(list), do: Enum.at(list, idx)
@doc "Get first element of a list or first byte of a binary."
def kaitai_first(nil), do: nil
def kaitai_first(bin) when is_binary(bin), do: :binary.at(bin, 0)
def kaitai_first(list) when is_list(list), do: List.first(list)
def kaitai_first(%{first: val}), do: val
def kaitai_first(%{} = map), do: map
@doc "Get last element of a list or last byte of a binary."
def kaitai_last(nil), do: nil
def kaitai_last(bin) when is_binary(bin), do: :binary.at(bin, byte_size(bin) - 1)
def kaitai_last(list) when is_list(list), do: List.last(list)
def kaitai_last(%{last: val}), do: val
def kaitai_last(%{} = map), do: map
@doc "Convert value to integer (like Ruby's .to_i)."
def to_i(true), do: 1
def to_i(false), do: 0
def to_i(x) when is_integer(x), do: x
def to_i(x) when is_float(x), do: trunc(x)
def to_i(x) when is_binary(x) do
case Integer.parse(x) do
{val, _} -> val
:error -> 0
end
end
def to_i(x) when is_atom(x), do: 0
@doc "Convert value to integer with enum reverse lookup."
def to_i(true, _reverse_map), do: 1
def to_i(false, _reverse_map), do: 0
def to_i(x, reverse_map) when is_atom(x), do: Map.get(reverse_map, x, 0)
def to_i(x, _reverse_map), do: to_i(x)
@doc "XOR each byte in data with a single-byte key."
def process_xor(data, key) when is_binary(data) and is_integer(key) do
for <<b <- data>>, into: <<>>, do: <<bxor(b, key)>>
end
def process_xor(data, key) when is_binary(data) and is_binary(key) do
key_len = byte_size(key)
do_xor_key(data, key, key_len, 0, [])
end
def process_xor(data, key) when is_binary(data) and is_list(key) do
process_xor(data, :binary.list_to_bin(key))
end
defp do_xor_key(<<b, rest::binary>>, key, kl, i, acc) do
do_xor_key(rest, key, kl, i + 1, [bxor(b, :binary.at(key, rem(i, kl))) | acc])
end
defp do_xor_key(<<>>, _, _, _, acc), do: acc |> :lists.reverse() |> IO.iodata_to_binary()
@doc "Rotate each byte left by amount bits."
def process_rotate_left(data, amount) when is_binary(data) and is_integer(amount) do
amount = rem(amount, 8)
for <<b <- data>>, into: <<>>, do: <<band(bor(bsl(b, amount), bsr(b, 8 - amount)), 0xFF)>>
end
@doc "Decode a binary from the given encoding to a UTF-8 string."
def decode_string(data, nil), do: data
def decode_string(data, encoding) do
enc = String.upcase(to_string(encoding))
case enc do
"UTF-8" ->
data
"ASCII" ->
data
"UTF-16LE" ->
:unicode.characters_to_binary(data, {:utf16, :little}, :utf8)
"UTF-16BE" ->
:unicode.characters_to_binary(data, {:utf16, :big}, :utf8)
"SJIS" ->
decode_sjis(data)
"IBM437" ->
decode_ibm437(data)
_ ->
data
end
end
@doc "Read a null-terminated string from binary with encoding support."
def read_strz_enc(data, encoding, consume \\ true, include \\ false) do
enc = if encoding, do: String.upcase(to_string(encoding)), else: nil
# For UTF-16 encodings, the terminator is two null bytes
if enc in ["UTF-16LE", "UTF-16BE"] do
{str_bytes, rest} = find_utf16_terminator(data, consume, include)
{decode_string(str_bytes, encoding), rest}
else
case :binary.match(data, <<0>>) do
{pos, 1} ->
str =
if include do
binary_part(data, 0, pos + 1)
else
binary_part(data, 0, pos)
end
rest =
if consume do
binary_part(data, pos + 1, byte_size(data) - pos - 1)
else
binary_part(data, pos, byte_size(data) - pos)
end
{decode_string(str, encoding), rest}
:nomatch ->
{decode_string(data, encoding), <<>>}
end
end
end
defp find_utf16_terminator(data, consume, include) do
find_utf16_terminator(data, 0, consume, include)
end
defp find_utf16_terminator(data, pos, consume, include) when pos + 1 < byte_size(data) do
if :binary.at(data, pos) == 0 and :binary.at(data, pos + 1) == 0 do
str =
if include do
binary_part(data, 0, pos + 2)
else
binary_part(data, 0, pos)
end
rest =
if consume do
binary_part(data, pos + 2, byte_size(data) - pos - 2)
else
binary_part(data, pos, byte_size(data) - pos)
end
{str, rest}
else
find_utf16_terminator(data, pos + 2, consume, include)
end
end
defp find_utf16_terminator(data, _pos, _consume, _include) do
{data, <<>>}
end
defp decode_sjis(data) do
decode_sjis_chars(data, [])
end
defp decode_sjis_chars(<<>>, acc), do: IO.iodata_to_binary(Enum.reverse(acc))
defp decode_sjis_chars(<<b, rest::binary>>, acc) when b < 0x80 do
decode_sjis_chars(rest, [<<b>> | acc])
end
defp decode_sjis_chars(<<b, rest::binary>>, acc) when b >= 0xA1 and b <= 0xDF do
# Half-width katakana
unicode = 0xFF61 + (b - 0xA1)
decode_sjis_chars(rest, [<<unicode::utf8>> | acc])
end
defp decode_sjis_chars(<<b1, b2, rest::binary>>, acc)
when (b1 >= 0x81 and b1 <= 0x9F) or (b1 >= 0xE0 and b1 <= 0xEF) do
unicode = sjis_to_unicode(b1, b2)
decode_sjis_chars(rest, [<<unicode::utf8>> | acc])
end
defp decode_sjis_chars(<<_b, rest::binary>>, acc) do
decode_sjis_chars(rest, [<<0xEF, 0xBF, 0xBD>> | acc])
end
defp sjis_to_unicode(b1, b2) do
# Convert SJIS to JIS X 0208 row/col
{row, col} = sjis_to_jis(b1, b2)
jis_to_unicode(row, col)
end
defp sjis_to_jis(b1, b2) do
row_offset = if b1 < 0xA0, do: 0x70, else: 0xB0
row = (b1 - row_offset) * 2 - 1
{row, col} =
if b2 >= 0x9F do
{row + 1, b2 - 0x7E}
else
col = if b2 > 0x7F, do: b2 - 0x40, else: b2 - 0x3F
{row, col + 0x20}
end
{row, col}
end
defp jis_to_unicode(row, col) do
cond do
# Hiragana
row == 0x24 -> 0x3020 + col
# Katakana
row == 0x25 -> 0x3080 + col
# Symbols row 1
row == 0x21 -> jis_symbols_row1(col)
# Full-width ASCII
row == 0x23 -> jis_fullwidth_ascii(col)
true -> 0xFFFD
end
end
defp jis_symbols_row1(col) do
# Common JIS X 0208 row 1 symbols
table = %{
0x21 => 0x3000,
0x22 => 0x3001,
0x23 => 0x3002,
0x24 => 0xFF0C,
0x25 => 0xFF0E,
0x26 => 0x30FB,
0x27 => 0xFF1A,
0x28 => 0xFF1B,
0x29 => 0xFF1F,
0x2A => 0xFF01,
0x3C => 0xFF0D,
0x5C => 0x30FC
}
Map.get(table, col, 0xFFFD)
end
defp jis_fullwidth_ascii(col) do
cond do
# 0-9
col >= 0x30 and col <= 0x39 -> 0xFF10 + (col - 0x30)
# A-Z
col >= 0x41 and col <= 0x5A -> 0xFF21 + (col - 0x41)
# a-z
col >= 0x61 and col <= 0x7A -> 0xFF41 + (col - 0x61)
true -> 0xFFFD
end
end
defp decode_ibm437(data) do
# IBM437 (CP437) to UTF-8 - map high bytes to Unicode
data
|> :binary.bin_to_list()
|> Enum.map(fn
b when b < 128 -> <<b::utf8>>
b -> <<ibm437_to_unicode(b)::utf8>>
end)
|> IO.iodata_to_binary()
end
# IBM437 high byte to Unicode mapping
defp ibm437_to_unicode(b) do
table = %{
128 => 0x00C7,
129 => 0x00FC,
130 => 0x00E9,
131 => 0x00E2,
132 => 0x00E4,
133 => 0x00E0,
134 => 0x00E5,
135 => 0x00E7,
136 => 0x00EA,
137 => 0x00EB,
138 => 0x00E8,
139 => 0x00EF,
140 => 0x00EE,
141 => 0x00EC,
142 => 0x00C4,
143 => 0x00C5,
144 => 0x00C9,
145 => 0x00E6,
146 => 0x00C6,
147 => 0x00F4,
148 => 0x00F6,
149 => 0x00F2,
150 => 0x00FB,
151 => 0x00F9,
152 => 0x00FF,
153 => 0x00D6,
154 => 0x00DC,
155 => 0x00A2,
156 => 0x00A3,
157 => 0x00A5,
158 => 0x20A7,
159 => 0x0192,
160 => 0x00E1,
161 => 0x00ED,
162 => 0x00F3,
163 => 0x00FA,
164 => 0x00F1,
165 => 0x00D1,
166 => 0x00AA,
167 => 0x00BA,
168 => 0x00BF,
169 => 0x2310,
170 => 0x00AC,
171 => 0x00BD,
172 => 0x00BC,
173 => 0x00A1,
174 => 0x00AB,
175 => 0x00BB,
176 => 0x2591,
177 => 0x2592,
178 => 0x2593,
179 => 0x2502,
180 => 0x2524,
181 => 0x2561,
182 => 0x2562,
183 => 0x2556,
184 => 0x2555,
185 => 0x2563,
186 => 0x2551,
187 => 0x2557,
188 => 0x255D,
189 => 0x255C,
190 => 0x255B,
191 => 0x2510,
192 => 0x2514,
193 => 0x2534,
194 => 0x252C,
195 => 0x251C,
196 => 0x2500,
197 => 0x253C,
198 => 0x255E,
199 => 0x255F,
200 => 0x255A,
201 => 0x2554,
202 => 0x2569,
203 => 0x2566,
204 => 0x2560,
205 => 0x2550,
206 => 0x256C,
207 => 0x2567,
208 => 0x2568,
209 => 0x2564,
210 => 0x2565,
211 => 0x2559,
212 => 0x2558,
213 => 0x2552,
214 => 0x2553,
215 => 0x256B,
216 => 0x256A,
217 => 0x2518,
218 => 0x250C,
219 => 0x2588,
220 => 0x2584,
221 => 0x258C,
222 => 0x2590,
223 => 0x2580,
224 => 0x03B1,
225 => 0x00DF,
226 => 0x0393,
227 => 0x03C0,
228 => 0x03A3,
229 => 0x03C3,
230 => 0x00B5,
231 => 0x03C4,
232 => 0x03A6,
233 => 0x0398,
234 => 0x03A9,
235 => 0x03B4,
236 => 0x221E,
237 => 0x03C6,
238 => 0x03B5,
239 => 0x2229,
240 => 0x2261,
241 => 0x00B1,
242 => 0x2265,
243 => 0x2264,
244 => 0x2320,
245 => 0x2321,
246 => 0x00F7,
247 => 0x2248,
248 => 0x00B0,
249 => 0x2219,
250 => 0x00B7,
251 => 0x221A,
252 => 0x207F,
253 => 0x00B2,
254 => 0x25A0,
255 => 0x00A0
}
Map.get(table, b, 0xFFFD)
end
@doc "KSY add operator - string concat or arithmetic add."
def kaitai_add(a, b) when is_binary(a) and is_binary(b), do: a <> b
def kaitai_add(a, b) when is_binary(a), do: a <> to_string(b)
def kaitai_add(a, b) when is_binary(b), do: to_string(a) <> b
def kaitai_add(a, b) when is_list(a) and is_list(b), do: a ++ b
def kaitai_add(a, b), do: a + b
@doc "Zlib decompress."
def process_zlib(data) when is_binary(data) do
:zlib.uncompress(data)
end
end