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, SS2/SS3 single-shifts, 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
• Structured CSI parameters: colon sub-parameters and empty fields are preserved, not flattened
• 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?, ss3?, osc?, invalid?, and so on — plus three groups:

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

CSI parameters

A CSI parameter list is a sequence of groups separated by ;. Each group is one or more sub-parameters separated by :. Any sub-parameter may be empty — the ITU T.416 color forms use an empty field for the colorspace slot, as in \e[38:2::255:128:0m.

The parameter accessors preserve this structure. An empty field is nil, never 0, so a real zero and an omitted field stay distinct.

token = Ansi::Tokenizer.tokenize("\e[1;38:5:200m").first

token.params    # [1, 38] — the head value of each group
token.groups    # two Param values: 1, and 38:5:200

params and each_param give you one value per group — the group's first sub-parameter, or nil if that field is empty. This is the common path for simple sequences like \e[1;31m.

token.each_param { |value| apply(value) }   # value : Int32?

For the extended color forms, reach into a group's sub-parameters with groups or each_group. Each yields a Param:

token = Ansi::Tokenizer.tokenize("\e[38:2::255:128:0m").first
group = token.groups.first

group.head        # 38 — the first sub-parameter, or nil if empty
group.size        # 6 — sub-parameter count
group.to_a        # [38, 2, nil, 255, 128, 0] — the empty colorspace field is nil
group.each { |sub| p sub }   # sub : Int32?

Because the colon form arrives as a single group, an SGR consumer can treat 38:... as one self-contained color operation regardless of whether the source used the colon form, the ITU empty-field form, or the legacy \e[38;2;255;128;0m semicolon form. The examples/sgr_state.cr folder handles all three.

For a fully allocation-free walk that still exposes structure, each_subparam yields every sub-parameter with a flag marking whether it opens a new group:

token.each_subparam do |value, first_in_group|
  # value : Int32?, first_in_group : Bool
end

params, groups, and Param#to_a allocate; each_param, each_group, each_subparam, and Param#each do not. Reach for the block forms in a hot path.

Both : and ; are recognized, so \e[38:2:255m and \e[38;2;255m parse into the shapes described above. An empty trailing group yields nil, matching terminal handling of \e[1;m as two parameters, the second empty.

CSI final and marker

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.payload     # "?25" as Bytes, introducer and final stripped

SS2 / SS3 accessors

Single-shift sequences carry exactly one byte after the introducer — an arrow key in application cursor mode arrives as \eOA, F1 as \eOP.

token = Ansi::Tokenizer.tokenize("\eOA").first

token.final       # 'A'.ord.to_u8 — the shifted byte
token.final_char  # 'A'
token.payload     # "A" as Bytes, introducer stripped

final and final_char are shared with Csi, so one match handles both output cursor reports and input arrow keys:

if token.ss3? || token.csi?
  case token.final_char
  when 'A' then up
  when 'B' then down
  when 'C' then right
  when 'D' then left
  end
end

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, \x8e/\x8f as SS2/SS3. 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

Track SGR state by folding groups over each m-final CSI, so extended colors resolve regardless of encoding:

token.groups.each do |group|
  case group.head
  when 0      then reset_style
  when 1      then bold
  when 30..37 then set_fg(group.head.not_nil! - 30)
  when 38, 48 then set_color(group.to_a)   # [38, 2, nil, r, g, b] or [38, 5, n]
  end
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 }

Examples

The examples/ directory covers the common uses end to end:

Run them all with make examples.

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.