File size: 4,222 Bytes
76aca0a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
# Sherpa Onnx Punctuation Preprocessor (CharTokenizer)
#
# Model: sherpa-onnx-punct-ct-transformer
# Tokenizer: character-level for Chinese, word-level for English
# Vocab: tokens.json (272727 entries)
# Padding: to 64 tokens

import json
import os
from typing import List, Tuple
import numpy as np


class CharTokenizer:
    """Character/word tokenizer for the sherpa punct CT Transformer model."""

    def __init__(self, tokens_path: str, unk_symbol: str = "<unk>"):
        if not os.path.exists(tokens_path):
            raise FileNotFoundError(f"tokens.json not found: {tokens_path}")
        with open(tokens_path, "r", encoding="utf-8") as f:
            id2token = json.load(f)
        self.id2token = id2token
        self.token2id = {tok: idx for idx, tok in enumerate(id2token)}
        self.unk_id = self.token2id.get(unk_symbol, 0)

    def tokenize(self, text: str) -> List[int]:
        """Split text into tokens and return token IDs.

        Chinese characters are segmented individually.
        English words are kept as whole tokens.
        """
        # Split on whitespace
        word_list = text.split()

        words = []
        for w in word_list:
            s = ""
            for c in w:
                if len(c.encode()) > 1:
                    # Multi-byte character (Chinese, Japanese, etc.)
                    if s == "":
                        s = c
                    elif len(s[-1].encode()) > 1:
                        s += c
                    else:
                        words.append(s)
                        s = c
                else:
                    # ASCII character
                    if s == "":
                        s = c
                    elif len(s[-1].encode()) > 1:
                        words.append(s)
                        s = c
                    else:
                        s += c
            if s:
                words.append(s)

        ids = []
        for w in words:
            if len(w[0].encode()) > 1:
                # Chinese phrase: tokenize each character
                for c in w:
                    ids.append(self.token2id.get(c, self.unk_id))
            else:
                ids.append(self.token2id.get(w, self.unk_id))
        return ids

    def tokenize_full(self, text: str) -> List[int]:
        """Tokenize full text without truncation or padding."""
        return self.tokenize(text)

    def encode(
        self, text: str, pad_length: int = 64
    ) -> Tuple[np.ndarray, int]:
        """Tokenize and pad to fixed length. Truncates if > pad_length.

        Returns:
            input_array: (1, pad_length) int32 numpy array
            original_length: actual token count before padding
        """
        ids = self.tokenize(text)
        original_len = len(ids)

        # Truncate or pad to pad_length
        if len(ids) > pad_length:
            ids = ids[:pad_length]
            original_len = pad_length

        padded = np.zeros((1, pad_length), dtype=np.int32)
        padded[0, : len(ids)] = ids

        return padded, min(original_len, pad_length)

    def encode_long(
        self, text: str, window_size: int = 64
    ) -> Tuple[List[np.ndarray], List[int], List[int]]:
        """Tokenize long text into sliding windows for batched inference.

        Splits full token sequence into windows of window_size.
        Each window is padded to window_size if shorter.

        Returns:
            windows: list of (1, window_size) int32 arrays
            window_token_ids: list of token ID lists per window
            window_lens: original token lengths per window (before padding)
        """
        ids = self.tokenize(text)
        if not ids:
            return [], [], []

        windows = []
        window_token_ids = []
        window_lens = []

        for start in range(0, len(ids), window_size):
            chunk = ids[start:start + window_size]
            chunk_len = len(chunk)

            padded = np.zeros((1, window_size), dtype=np.int32)
            padded[0, :chunk_len] = chunk

            windows.append(padded)
            window_token_ids.append(chunk)
            window_lens.append(chunk_len)

        return windows, window_token_ids, window_lens