スマートグラス Even G2 と Amazon Transcribe Streaming で会話中の 5・7・5 を拾ってみた

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Even G2 streaming transcription and 5-7-5 detection mock

Parameters:
  AllowedSourceIp:
    Type: String
    Description: Global IPv4 address allowed to invoke the API, in /32 CIDR notation
    AllowedPattern: '^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)/32$'

Globals:
  Function:
    Runtime: nodejs22.x
    Architectures:
      - x86_64
    MemorySize: 512
    Timeout: 10

Resources:
  DetectorApi:
    Type: AWS::Serverless::Api
    Properties:
      Name: !Sub '${AWS::StackName}-api'
      StageName: demo
      OpenApiVersion: 3.0.1
      EndpointConfiguration: REGIONAL
      Cors:
        AllowOrigin: "'*'"
        AllowMethods: "'POST,OPTIONS'"
        AllowHeaders: "'Content-Type'"
      Auth:
        ResourcePolicy:
          IpRangeWhitelist:
            - !Ref AllowedSourceIp

  DetectorFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub '${AWS::StackName}-api'
      Description: Creates Transcribe sessions and detects 5-7-5 mora sequences
      CodeUri: ../backend/
      Handler: src/handler.handler
      Policies:
        - AWSLambdaBasicExecutionRole
        - Version: '2012-10-17'
          Statement:
            - Sid: StartTranscribeWebSocket
              Effect: Allow
              Action: transcribe:StartStreamTranscriptionWebSocket
              Resource: '*'
      Events:
        CreateSession:
          Type: Api
          Properties:
            RestApiId: !Ref DetectorApi
            Path: /session
            Method: post
        Detect575:
          Type: Api
          Properties:
            RestApiId: !Ref DetectorApi
            Path: /detect
            Method: post

  DetectorFunctionLogGroup:
    Type: AWS::Logs::LogGroup
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      LogGroupName: !Sub '/aws/lambda/${DetectorFunction}'
      RetentionInDays: 1

Outputs:
  ApiUrl:
    Description: Base URL for the Even Hub app
    Value: !Sub 'https://${DetectorApi}.execute-api.${AWS::Region}.${AWS::URLSuffix}/demo'
  FunctionName:
    Description: Lambda function name
    Value: !Ref DetectorFunction
  LogGroupName:
    Description: CloudWatch Logs log group created by this stack
    Value: !Ref DetectorFunctionLogGroup

function hasKnownReading(token) {
  return (
    typeof token.reading === 'string' &&
    Number.isInteger(token.moras) &&
    token.moras >= 0
  )
}

function joinTokens(tokens, field) {
  return tokens.map(token => token[field]).join('')
}

function candidateFrom(tokens, boundaries, resultId, startIndex, endIndex) {
  const ranges = [
    [startIndex, boundaries[0]],
    [boundaries[0] + 1, boundaries[1]],
    [boundaries[1] + 1, boundaries[2]],
  ]
  const lines = ranges.map(([start, end]) =>
    joinTokens(tokens.slice(start, end + 1), 'surface'),
  )
  const readings = ranges.map(([start, end]) =>
    joinTokens(tokens.slice(start, end + 1), 'reading'),
  )

  return {
    id: `${resultId}:${startIndex}:${endIndex}:${readings.join('|')}`,
    lines,
    readings,
  }
}

function findCandidates(tokens, resultId) {
  const candidates = []

  for (let startIndex = 0; startIndex < tokens.length; startIndex += 1) {
    let total = 0
    const boundaries = []

    for (let endIndex = startIndex; endIndex < tokens.length; endIndex += 1) {
      total += tokens[endIndex].moras

      const boundaryIndex = [5, 12, 17].indexOf(total)
      if (boundaryIndex >= 0) {
        boundaries[boundaryIndex] = endIndex
      }

      if (total === 17) {
        if (boundaries.length === 3 && tokens[endIndex].isNew) {
          candidates.push(
            candidateFrom(tokens, boundaries, resultId, startIndex, endIndex),
          )
        }
        break
      }

      const nextBoundary = [5, 12, 17][boundaries.length]
      if (total > nextBoundary) break
    }
  }

  return candidates
}

function trailingContext(tokens) {
  let startIndex = tokens.length
  let total = 0

  for (let index = tokens.length - 1; index >= 0; index -= 1) {
    const nextTotal = total + tokens[index].moras
    if (nextTotal > 16) break
    total = nextTotal
    startIndex = index
  }

  return tokens
    .slice(startIndex)
    .map(({ isNew: _isNew, ...token }) => token)
}

export function detect575({ context, newTokens, resultId }) {
  const combined = [
    ...context.map(token => ({ ...token, isNew: false })),
    ...newTokens.map(token => ({ ...token, isNew: true })),
  ]
  const lastUnknownIndex = combined.findLastIndex(
    token => !hasKnownReading(token),
  )
  const usable = combined.slice(lastUnknownIndex + 1)

  return {
    context: trailingContext(usable),
    candidates: findCandidates(usable, resultId),
  }
}

import { fileURLToPath } from 'node:url'

const IGNORED_CHARACTERS = /[\s、。！？,.!?]/u
const SMALL_KANA = /[ャュョァィゥェォヮゃゅょぁぃぅぇぉゎ]/u
const KANA_ONLY = /^[ぁ-ゖァ-ヺー]+$/u

export function countMoras(reading) {
  let count = 0

  for (const character of reading) {
    if (!IGNORED_CHARACTERS.test(character) && !SMALL_KANA.test(character)) {
      count += 1
    }
  }

  return count
}

function normalizeReading(reading) {
  return Array.from(reading, character => {
    const codePoint = character.codePointAt(0)
    if (codePoint >= 0x3041 && codePoint <= 0x3096) {
      return String.fromCodePoint(codePoint + 0x60)
    }
    return character
  }).join('')
}

export function toAnalyzedTokens(features) {
  return features.map(feature => {
    const readingSource =
      feature.pronunciation ||
      feature.reading ||
      (KANA_ONLY.test(feature.surface_form) ? feature.surface_form : null)
    const reading = readingSource ? normalizeReading(readingSource) : null

    return {
      surface: feature.surface_form,
      reading,
      moras: reading === null ? null : countMoras(reading),
    }
  })
}

export async function createJapaneseTokenizer(options = {}) {
  const { builder } = await import('@faanau/kuromoji')
  const dicPath =
    options.dicPath ??
    fileURLToPath(
      new URL('../node_modules/@faanau/kuromoji/dict/', import.meta.url),
    )

  const tokenizer = await new Promise((resolve, reject) => {
    builder({ dicPath }).build((error, builtTokenizer) => {
      if (error) reject(error)
      else resolve(builtTokenizer)
    })
  })

  return text => toAnalyzedTokens(tokenizer.tokenize(text))
}

import { EventStreamCodec } from '@smithy/core/event-streams'
import { fromUtf8, toUtf8 } from '@smithy/core/serde'

export interface TranscriptResult {
  resultId: string
  isPartial: boolean
  transcript: string
}

export type TranscribeMessage =
  | { kind: 'transcript'; results: TranscriptResult[] }
  | { kind: 'exception'; type: string; message: string }

const codec = new EventStreamCodec(toUtf8, fromUtf8)

export function encodeAudioEvent(pcm: Uint8Array): Uint8Array {
  return codec.encode({
    headers: {
      ':message-type': { type: 'string', value: 'event' },
      ':event-type': { type: 'string', value: 'AudioEvent' },
      ':content-type': { type: 'string', value: 'application/octet-stream' },
    },
    body: pcm,
  })
}

function stringHeader(
  headers: ReturnType<EventStreamCodec['decode']>['headers'],
  name: string,
): string | undefined {
  const header = headers[name]
  return header?.type === 'string' ? header.value : undefined
}

export function decodeTranscribeMessage(data: ArrayBufferView): TranscribeMessage {
  const decoded = codec.decode(data)
  const messageType = stringHeader(decoded.headers, ':message-type')
  const body = JSON.parse(toUtf8(decoded.body)) as unknown

  if (messageType === 'exception') {
    const value = body as { Message?: unknown; message?: unknown }
    const message = value.Message ?? value.message
    return {
      kind: 'exception',
      type: stringHeader(decoded.headers, ':exception-type') ?? 'TranscribeException',
      message: typeof message === 'string' ? message : 'Amazon Transcribeでエラーが発生しました。',
    }
  }

  const payload = body as {
    Transcript?: {
      Results?: Array<{
        ResultId?: unknown
        IsPartial?: unknown
        Alternatives?: Array<{ Transcript?: unknown }>
      }>
    }
  }
  const results = payload.Transcript?.Results ?? []

  return {
    kind: 'transcript',
    results: results.flatMap((result) => {
      const transcript = result.Alternatives?.[0]?.Transcript
      if (
        typeof result.ResultId !== 'string' ||
        typeof result.IsPartial !== 'boolean' ||
        typeof transcript !== 'string'
      ) {
        return []
      }
      return [{ resultId: result.ResultId, isPartial: result.IsPartial, transcript }]
    }),
  }
}

export class PcmChunker {
  private pending = new Uint8Array()

  constructor(private readonly chunkSize = 3200) {
    if (chunkSize <= 0 || chunkSize % 2 !== 0) {
      throw new Error('chunkSizeは正の偶数である必要があります。')
    }
  }

  push(input: Uint8Array): Uint8Array[] {
    if (input.byteLength % 2 !== 0) {
      throw new Error('16bit PCMの入力は偶数バイトである必要があります。')
    }

    const joined = new Uint8Array(this.pending.byteLength + input.byteLength)
    joined.set(this.pending)
    joined.set(input, this.pending.byteLength)

    const chunks: Uint8Array[] = []
    let offset = 0
    while (joined.byteLength - offset >= this.chunkSize) {
      chunks.push(joined.slice(offset, offset + this.chunkSize))
      offset += this.chunkSize
    }
    this.pending = joined.slice(offset)
    return chunks
  }

  flush(): Uint8Array {
    const remainder = this.pending
    this.pending = new Uint8Array()
    return remainder
  }
}

import type { TranscribeSession } from '../api/detector-api'
import { decodeTranscribeMessage, encodeAudioEvent, type TranscriptResult } from './eventstream'
import { PcmChunker } from './pcm-chunker'

const OPEN = 1
const CLOSED = 3

export interface WebSocketLike {
  binaryType: BinaryType
  readyState: number
  onopen: (() => void) | null
  onmessage: ((event: { data: ArrayBuffer }) => void) | null
  onerror: (() => void) | null
  onclose: (() => void) | null
  send(data: Uint8Array): void
  close(): void
}

interface TranscribeClientOptions {
  createSession: () => Promise<TranscribeSession>
  onResult: (result: TranscriptResult) => void | Promise<void>
  onFatal: (error: Error) => void
  onFinishTimeout?: () => void
  socketFactory?: (url: string) => WebSocketLike
}

type State = 'idle' | 'connecting' | 'open' | 'finishing' | 'closed'

export class TranscribeClient {
  private readonly chunker = new PcmChunker(3200)
  private readonly socketFactory: (url: string) => WebSocketLike
  private socket?: WebSocketLike
  private state: State = 'idle'
  private finishPromise?: Promise<void>
  private resolveFinish?: () => void
  private finishTimer?: ReturnType<typeof setTimeout>
  private failed = false
  private readonly pendingResultTasks = new Set<Promise<void>>()

  constructor(private readonly options: TranscribeClientOptions) {
    this.socketFactory =
      options.socketFactory ?? ((url) => new WebSocket(url) as unknown as WebSocketLike)
  }

  async start(): Promise<void> {
    if (this.state !== 'idle') throw new Error('文字起こしはすでに開始されています。')
    this.state = 'connecting'
    const session = await this.options.createSession()
    if (Date.parse(session.connectBefore) <= Date.now()) {
      this.state = 'closed'
      throw new Error('文字起こし接続URLの有効期限が切れています。')
    }

    const socket = this.socketFactory(session.url)
    this.socket = socket
    socket.binaryType = 'arraybuffer'

    return new Promise<void>((resolve, reject) => {
      socket.onopen = () => {
        this.state = 'open'
        resolve()
      }
      socket.onmessage = (event) => void this.handleMessage(event.data)
      socket.onerror = () => {
        const error = new Error('Amazon Transcribeとの接続でエラーが発生しました。')
        if (this.state === 'connecting') reject(error)
        this.fail(error)
      }
      socket.onclose = () => {
        const previousState = this.state
        this.state = 'closed'
        this.clearFinishTimer()
        void this.resolveFinishAfterPendingResults()
        if (previousState === 'connecting') {
          reject(new Error('Amazon Transcribeへ接続できませんでした。'))
        } else if (previousState === 'open' && !this.failed) {
          this.fail(new Error('Amazon Transcribeとの接続が途中で閉じられました。'))
        }
      }
    })
  }

  sendPcm(pcm: Uint8Array): void {
    if (this.state !== 'open' || this.socket?.readyState !== OPEN) return
    for (const chunk of this.chunker.push(pcm)) {
      this.socket.send(encodeAudioEvent(chunk))
    }
  }

  finish(): Promise<void> {
    if (this.finishPromise !== undefined) return this.finishPromise
    if (this.state !== 'open' || this.socket?.readyState !== OPEN) {
      return Promise.resolve()
    }

    this.state = 'finishing'
    this.finishPromise = new Promise<void>((resolve) => {
      this.resolveFinish = resolve
      this.finishTimer = setTimeout(() => {
        this.options.onFinishTimeout?.()
        this.closeSocket()
      }, 5000)
    })
    const remainder = this.chunker.flush()
    if (remainder.byteLength > 0) this.socket.send(encodeAudioEvent(remainder))
    this.socket.send(encodeAudioEvent(new Uint8Array()))
    return this.finishPromise
  }

  private async handleMessage(data: ArrayBuffer): Promise<void> {
    try {
      const message = decodeTranscribeMessage(new Uint8Array(data))
      if (message.kind === 'exception') {
        this.fail(new Error(`${message.type}: ${message.message}`))
        return
      }

      await Promise.all(message.results.map((result) => this.trackResult(result)))
      if (this.state === 'finishing' && message.results.some((result) => !result.isPartial)) {
        this.closeSocket()
      }
    } catch {
      this.fail(new Error('Amazon Transcribeの応答を解析できませんでした。'))
    }
  }

  private trackResult(result: TranscriptResult): Promise<void> {
    const task = Promise.resolve(this.options.onResult(result)).catch((error: unknown) => {
      this.fail(error instanceof Error ? error : new Error('確定結果の処理に失敗しました。'))
    })
    this.pendingResultTasks.add(task)
    void task.finally(() => this.pendingResultTasks.delete(task))
    return task
  }

  private async resolveFinishAfterPendingResults(): Promise<void> {
    await Promise.all([...this.pendingResultTasks])
    this.resolveFinish?.()
  }

  private fail(error: Error): void {
    if (!this.failed) {
      this.failed = true
      this.options.onFatal(error)
    }
    this.closeSocket()
  }

  private closeSocket(): void {
    this.clearFinishTimer()
    if (this.socket !== undefined && this.socket.readyState !== CLOSED) {
      this.socket.close()
    } else {
      this.resolveFinish?.()
    }
  }

  private clearFinishTimer(): void {
    if (this.finishTimer !== undefined) {
      clearTimeout(this.finishTimer)
      this.finishTimer = undefined
    }
  }
}