ansi

A zero-copy, streaming tokenizer for ANSI/ECMA-48 escape sequences in Crystal.

Splits a byte stream into typed tokens — text runs, C0/C1 controls, CSI sequences, OSC/DCS/APC/SOS/PM strings — without allocating per token and without copying the input.
Sequences split across chunk boundaries are reassembled automatically, so it works directly against a PTY, socket, or subprocess pipe.

• Zero allocation on the streaming path: tokens borrow slices of your buffer
• SWAR text scanning: >13 GB/s on printable runs
• Lossless: concatenating every token reproduces the input byte-for-byte, including malformed sequences
• Lazy accessors: parameters, OSC codes, and payloads are parsed on demand, not eagerly
• No dependencies

Installation

Add to shard.yml:

dependencies:
  ansi:
    github: shpeckman/ansi

Then:

shards install

Requires Crystal >= 1.21.0.

Quick start

require "ansi"

Ansi::Tokenizer.tokenize("\e[1;31mError\e[0m: failed").each do |token|
  puts "#{token.kind}\t#{token.to_s.inspect}"
end

Csi     "\e[1;31m"
Text    "Error"
Csi     "\e[0m"
Text    ": failed"

Tokenizer.tokenize is the batch entry point.
It copies every token into a single arena allocation, so the returned tokens are owned and outlive the source buffer.

Streaming

feed accepts arbitrary chunks and yields tokens as they complete.
Anything incomplete at the end of a chunk is carried into the next one.

tokenizer = Ansi::Tokenizer.new
buffer    = Bytes.new(4096)

while (read = io.read(buffer)) > 0
  tokenizer.feed(buffer[0, read]) { |token| handle(token) }
end

tokenizer.finish { |token| handle(token) }

A \e[3 at the end of one chunk and 1m at the start of the next emit as a single Csi token.
finish flushes whatever remains pending when the stream ends — an unterminated sequence is emitted with its expected kind rather than discarded.

Call reset to clear carried state before reusing a tokenizer on an unrelated stream.

Ownership

Tokens yielded by feed borrow the slice you passed in. Nothing is allocated, but they become invalid as soon as you reuse or free that buffer.

tokenizer.feed(chunk) { |token| tokens << token }        # dangling once chunk is reused
tokenizer.feed(chunk) { |token| tokens << token.clone }  # safe

Use tokens freely inside the block.
To retain one beyond it, call clone.
Cloning an already-owned token is a no-op, so it is cheap to call unconditionally.
owned? reports which kind you hold.

Tokens carried across a chunk boundary are owned already — they are assembled in an internal buffer — as are all tokens from tokenize and finish.

Token

token.kind    # Kind
token.bytes   # Bytes — the raw, complete sequence
token.size    # Int32
token.to_s    # String
token.to_s(io)  # writes raw bytes, no allocation

Kinds

Predicates exist for every kind — text?, csi?, osc?, invalid?, and so on — plus two groups:

token.control?  # anything but Text
token.string?   # Osc, Dcs, Apc, Sos, Pm

CSI accessors

token = Ansi::Tokenizer.tokenize("\e[?25h").first

token.final       # 'h'.ord.to_u8 — the command byte, or nil
token.final_char  # 'h'
token.marker      # '?' — private marker (< = > ?), or nil
token.params      # [25]
token.payload     # "25" as Bytes, introducer and final stripped

params allocates. In a hot path use each_param, which does not:

token.each_param { |param| apply(param) }

Both treat : and ; as separators, so \e[38:2:255m yields 38, 2, 255. An empty trailing parameter yields 0, matching terminal behaviour for \e[1;m.

OSC accessors

token = Ansi::Tokenizer.tokenize("\e]0;window title\a").first

token.code                # 0 — the numeric command, or nil
String.new(token.payload) # "0;window title" — introducer and terminator stripped

Options

Ansi::Tokenizer.new(c1: true, max_sequence: 4096)

c1 (default false) enables 8-bit control recognition: \x9b as CSI, \x9d as OSC, \x9c as ST. Leave it off for UTF-8 input, where those bytes are legitimate continuation bytes and should pass through as text. Enable it only when the source is known to speak 8-bit controls.

max_sequence (default 65536) caps the carry buffer. An unterminated sequence longer than this is force-emitted as Invalid rather than buffering without bound — a guard against hostile or corrupt input.

Recipes

Strip escape sequences:

plain = String.build do |io|
  Ansi::Tokenizer.tokenize(input).each { |token| io << token if token.text? }
end

Decode terminal events by matching the final byte and private marker together:

case {token.final_char, token.marker}
when {'I', nil} then focus_gained
when {'O', nil} then focus_lost
when {'M', '<'} then mouse_press(token.params)
when {'m', '<'} then mouse_release(token.params)
when {'h', '?'} then mode_set(token.params)
end

Rewrite a stream while preserving everything you don't touch:

tokenizer.feed(chunk) do |token|
  if token.csi? && token.final_char == 'm'
    io << rewrite_sgr(token)
  else
    token.to_s(io)
  end
end

Iterate lazily:

tokenizer.each(input).select(&.csi?).each { |token| p token.params }

Development

make spec      # run the spec suite
make bench     # run benchmarks, pinned to an isolated core when possible
make examples  # run every example in examples/
make           # specs, benchmarks, examples

License

MIT. See LICENSE.