Pattern Recognition Integer Sequence
An algorithm and tool for detecting patterns in data streams using parallel realtime stream sequence detection with a finite automaton.
PRIS is designed to detect patterns or sequences in data streams. Capture character strings or event sequences without caching, states, or overhead. Ideal for game input detection and sequence testing.
- Install
- Quick Start
- Features
- What is PRIS?
- How It Works
- Usage
- Examples
- Functional Positions
- Key IDs
- Understanding Hots, Matches, and Drops
- Hot, Match, Drop Explained
- Related Algorithms
- Algorithm Background
pip install prisUsing pris is a two-step process:
- Create a
Sequences()table - Pump it with
.table_insert_keys(...)for all your incoming bits.
from pris import Sequences
# something to detect
sequence = ('a', 'b', 'c')
sq = Sequences()
sq.input_sequence(sequence, 'alphabet')Simulate a stream:
hots, matches, drops = sq.table_insert_keys(['a', 'b', 'c'])
# Output: (), ('alphabet',), ()- Feedforward sequencing
- Works with streams of unlimited length
- Detects overlapping and repeat sequences
- Functional sinking for inline dynamic paths
- Minimal overhead (1 integer per path)
- Unlimited path length
- Real-time operation
Parallel Sequence Detection on Realtime Streams with a Finite Automaton
A fancy way to say: PRIS detects and matches sequences in streaming data with no cache, states, or memory bloat. Sequences can be strings, iterables, or events.
- Parallel: Detect many sequences in parallel ("wind" in "window" and "sidewinder")
- Sequence: Stepwise matching (e.g., W → I → N → D)
- Detection: Find sequences in a stream (file, events, etc.)
- Realtime: Works with live or unseekable data (media, sockets, etc.)
- Finite Automaton: Internally tracks position using the leanest structure possible (an int per path)
- Add sequences to detect
- Input events (like key presses)
- Capture matches, hot starts, or drops (misses)
Internally, PRIS keeps a table of indices (ints) per sequence. When an input matches, it increments the index; on failure, it resets. Everything is handled live, without caching history.
- O(k) initiation via hot start
- One int per path
- O(k) position checks per event
- O(n) for iterating all live sequences
Still searching for a formal name for this algorithm. If you know it, get in touch!
Basic usage:
from pris import Sequences
sq = Sequences()
sq.input_sequence(('a', 'b', 'c'), 'abc')
hots, matches, drops = sq.table_insert_keys(['a', 'b', 'c'])Possible applications:
- Game input/combos
- Sequence search in files or streams
- Command or protocol detection
from pris import Sequences
KONAMI_CODE = ('up', 'up', 'down', 'down', 'left', 'right', 'left', 'right', 'b', 'a', 'start')
sq = Sequences()
sq.input_sequence(KONAMI_CODE, 'konami')
# Simulate pressing all but last
sq.table_insert_keys(KONAMI_CODE[:-1])
# Press last
hots, matches, drops = sq.table_insert_keys(['start'])
print("Matches", matches) # ('konami',)Sequences can include functions as wildcards:
from pris import Sequences
def is_vowel(v): return v in 'aeiou'
sq = Sequences()
sq.input_sequence(('p', is_vowel, 't'), 'p?t')
inputs = [['p','a','t'], ['p','u','t'], ['p','e','t']]
for iv in inputs:
_, m, _ = sq.table_insert_keys(iv)
print("Matches", m) # ('p?t',)A sequence key can be any value. If None, the key is stringified from the sequence.
from pris import sequences
WORDS = {
'window': ('w', 'i', 'n', 'd', 'o', 'w'),
'windy': 'windy',
}
sq = sequences.Sequences(WORDS)
trip = sq.insert_keys(*'window')Alternatively, use the Sequence class:
from pris import sequences, Sequence
WORDS = [
Sequence('window', 'w', 'i', 'n', 'd', 'o', 'w'),
Sequence('windy'),
]
sq = sequences.Sequences(WORDS)
trip = sq.insert_keys(*'window')Wildcards, functional sinks, and overlaps:
from pris import sequences
def sink(v): return True
def vowel(v): return v in 'aieou'
WORDS = [
('w','i','n','d','o','w'),
'windy',
('q',sink,'d'),
('c',vowel,'t'),
]
sq = sequences.Sequences(WORDS)
trip = sq.insert_keys(*'window')Searching in a long string:
from pris.sequences import Sequences
from collections import Counter
def sink(v): return True
sq = Sequences()
sq.input_sequence('fragil')
sq.input_sequence(('a', sink, 'i'), 'a?i')
incoming = "supercalifragilisticexpialidocious"
counter = Counter()
for ch in incoming:
_, matches, _ = sq.insert_key(ch)
counter.update(matches)
print(counter) # Counter({'a?i': 3, 'fragil': 1})Three event types bubble up when you insert:
- Hots: Sequences now active ("hot")
- Matches: Sequences completed
- Drops: Sequences that failed mid-stream
from pris.sequences import Sequences
sq = Sequences()
sq.input_sequence(('a','b','c'),'A')
sq.input_sequence(('x','y','z'),'B')
h, m, d = sq.table_insert_keys(['a','b'])
print("Hots", h) # ('A',)
h, m, d = sq.table_insert_keys(['x'])
print("Drops", d) # ('A',)
h, m, d = sq.table_insert_keys(['y','z'])
print("Matches", m) # ('B',)- Hot: Sequence has started and is being tracked
- Match: Sequence was detected from start to finish
- Drop: Sequence failed due to a mismatch
- Commentz-Walter
- Boyer-Moore
- Knuth-Morris-Pratt (KMP)
- Rabin-Karp
- Aho-Corasick
- Finite State Machines (FSM)
- Dynamic Programming (LCS, etc)
PRIS was originally built for tracking secure channel switches in WebSocket servers, mapping user actions along a path with no cache. It’s since evolved to fast detection of any sequence (chars, events, objects) in real time, with no stored history and minimal resource usage.
- Non-Interference: Paths don't conflict with themselves. Multiple chars in a sequence are handled cleanly.
- Parallel Paths: Similar initial sequences (e.g. "windy" and "wind") coexist with no drama.
reducing complexity from O(n) to O(k) through hot start reduction.
Still, if you know a precise algorithm this matches, get in touch or PR it in.