Implement Seriatim subprocess adapter
This commit is contained in:
274
internal/adapters/seriatim/subprocess.go
Normal file
274
internal/adapters/seriatim/subprocess.go
Normal file
@@ -0,0 +1,274 @@
|
||||
package seriatim
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/adapters/subprocess"
|
||||
)
|
||||
|
||||
// EnvConfig defines optional Seriatim environment tuning values.
|
||||
type EnvConfig struct {
|
||||
OverlapWordRunGap *float64
|
||||
OverlapWordRunReorderWindow *float64
|
||||
BackchannelMaxDuration *float64
|
||||
FillerMaxDuration *float64
|
||||
}
|
||||
|
||||
// SubprocessRunnerConfig defines deterministic settings for Seriatim CLI execution.
|
||||
type SubprocessRunnerConfig struct {
|
||||
Binary string
|
||||
Timeout time.Duration
|
||||
OutputSchema string
|
||||
CoalesceGap *float64
|
||||
Report bool
|
||||
Env EnvConfig
|
||||
}
|
||||
|
||||
// SubprocessRunner invokes Seriatim via subprocess.
|
||||
type SubprocessRunner struct {
|
||||
binary string
|
||||
timeout time.Duration
|
||||
outputSchema string
|
||||
coalesceGap *float64
|
||||
report bool
|
||||
env EnvConfig
|
||||
}
|
||||
|
||||
// NewSubprocessRunnerFromConfigValues parses config-derived values once.
|
||||
func NewSubprocessRunnerFromConfigValues(
|
||||
binary string,
|
||||
timeout string,
|
||||
outputSchema string,
|
||||
coalesceGap *float64,
|
||||
report bool,
|
||||
env EnvConfig,
|
||||
) (*SubprocessRunner, error) {
|
||||
if strings.TrimSpace(timeout) == "" {
|
||||
return nil, fmt.Errorf("seriatim timeout is required")
|
||||
}
|
||||
parsedTimeout, err := time.ParseDuration(timeout)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse seriatim timeout %q: %w", timeout, err)
|
||||
}
|
||||
return NewSubprocessRunner(SubprocessRunnerConfig{
|
||||
Binary: binary,
|
||||
Timeout: parsedTimeout,
|
||||
OutputSchema: outputSchema,
|
||||
CoalesceGap: coalesceGap,
|
||||
Report: report,
|
||||
Env: env,
|
||||
})
|
||||
}
|
||||
|
||||
// NewSubprocessRunner constructs a validated Seriatim subprocess runner.
|
||||
func NewSubprocessRunner(cfg SubprocessRunnerConfig) (*SubprocessRunner, error) {
|
||||
if strings.TrimSpace(cfg.Binary) == "" {
|
||||
return nil, fmt.Errorf("seriatim binary is required")
|
||||
}
|
||||
if cfg.Timeout <= 0 {
|
||||
return nil, fmt.Errorf("seriatim timeout must be > 0")
|
||||
}
|
||||
if strings.TrimSpace(cfg.OutputSchema) == "" {
|
||||
return nil, fmt.Errorf("seriatim output schema is required")
|
||||
}
|
||||
switch cfg.OutputSchema {
|
||||
case "seriatim-minimal", "seriatim-intermediate", "seriatim-full":
|
||||
default:
|
||||
return nil, fmt.Errorf("seriatim output schema %q is unsupported", cfg.OutputSchema)
|
||||
}
|
||||
if cfg.CoalesceGap != nil && *cfg.CoalesceGap < 0 {
|
||||
return nil, fmt.Errorf("seriatim coalesce gap must be >= 0")
|
||||
}
|
||||
|
||||
return &SubprocessRunner{
|
||||
binary: cfg.Binary,
|
||||
timeout: cfg.Timeout,
|
||||
outputSchema: cfg.OutputSchema,
|
||||
coalesceGap: cfg.CoalesceGap,
|
||||
report: cfg.Report,
|
||||
env: cfg.Env,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Run executes Seriatim merge with deterministic flags and validates output artifacts.
|
||||
func (r *SubprocessRunner) Run(ctx context.Context, req MergeRequest) (MergeResult, error) {
|
||||
if r == nil {
|
||||
return MergeResult{}, fmt.Errorf("seriatim subprocess runner is nil")
|
||||
}
|
||||
if strings.TrimSpace(req.OutputMergedTranscriptPath) == "" {
|
||||
return MergeResult{}, fmt.Errorf("seriatim merge output path is required")
|
||||
}
|
||||
if len(req.InputTranscriptPaths) == 0 {
|
||||
return MergeResult{}, fmt.Errorf("seriatim merge requires at least one input transcript")
|
||||
}
|
||||
if r.report && strings.TrimSpace(req.ReportPath) == "" {
|
||||
return MergeResult{}, fmt.Errorf("seriatim report is enabled but report path is missing")
|
||||
}
|
||||
|
||||
args := r.buildArgs(req)
|
||||
env := r.buildEnvOverrides()
|
||||
|
||||
if req.GeneratedConfigPath != "" {
|
||||
if err := r.writeInvocationConfig(req, args); err != nil {
|
||||
return MergeResult{}, fmt.Errorf("write seriatim invocation config %q: %w", req.GeneratedConfigPath, err)
|
||||
}
|
||||
}
|
||||
|
||||
runRes, err := subprocess.Run(ctx, subprocess.RunRequest{
|
||||
Executable: r.binary,
|
||||
Args: args,
|
||||
Timeout: r.timeout,
|
||||
EnvOverrides: env,
|
||||
StdoutLogPath: req.StdoutLogPath,
|
||||
StderrLogPath: req.StderrLogPath,
|
||||
})
|
||||
if err != nil {
|
||||
return MergeResult{
|
||||
MergedTranscriptPath: req.OutputMergedTranscriptPath,
|
||||
ReportPath: req.ReportPath,
|
||||
StdoutLogPath: req.StdoutLogPath,
|
||||
StderrLogPath: req.StderrLogPath,
|
||||
GeneratedConfigPath: req.GeneratedConfigPath,
|
||||
ExitCode: runRes.ExitCode,
|
||||
Duration: runRes.Duration,
|
||||
InvokedBinary: r.binary,
|
||||
OutputSchema: r.outputSchema,
|
||||
}, fmt.Errorf("run seriatim merge (binary=%q): %w", r.binary, err)
|
||||
}
|
||||
|
||||
if err := validateJSONFile(req.OutputMergedTranscriptPath); err != nil {
|
||||
return MergeResult{
|
||||
MergedTranscriptPath: req.OutputMergedTranscriptPath,
|
||||
ReportPath: req.ReportPath,
|
||||
StdoutLogPath: req.StdoutLogPath,
|
||||
StderrLogPath: req.StderrLogPath,
|
||||
GeneratedConfigPath: req.GeneratedConfigPath,
|
||||
ExitCode: runRes.ExitCode,
|
||||
Duration: runRes.Duration,
|
||||
InvokedBinary: r.binary,
|
||||
OutputSchema: r.outputSchema,
|
||||
}, fmt.Errorf("validate seriatim merged output %q: %w", req.OutputMergedTranscriptPath, err)
|
||||
}
|
||||
|
||||
if r.report {
|
||||
if err := validateJSONFile(req.ReportPath); err != nil {
|
||||
return MergeResult{
|
||||
MergedTranscriptPath: req.OutputMergedTranscriptPath,
|
||||
ReportPath: req.ReportPath,
|
||||
StdoutLogPath: req.StdoutLogPath,
|
||||
StderrLogPath: req.StderrLogPath,
|
||||
GeneratedConfigPath: req.GeneratedConfigPath,
|
||||
ExitCode: runRes.ExitCode,
|
||||
Duration: runRes.Duration,
|
||||
InvokedBinary: r.binary,
|
||||
OutputSchema: r.outputSchema,
|
||||
}, fmt.Errorf("validate seriatim report output %q: %w", req.ReportPath, err)
|
||||
}
|
||||
}
|
||||
|
||||
return MergeResult{
|
||||
MergedTranscriptPath: req.OutputMergedTranscriptPath,
|
||||
ReportPath: req.ReportPath,
|
||||
StdoutLogPath: req.StdoutLogPath,
|
||||
StderrLogPath: req.StderrLogPath,
|
||||
GeneratedConfigPath: req.GeneratedConfigPath,
|
||||
ExitCode: runRes.ExitCode,
|
||||
Duration: runRes.Duration,
|
||||
InvokedBinary: r.binary,
|
||||
OutputSchema: r.outputSchema,
|
||||
Metadata: map[string]any{
|
||||
"adapter": "seriatim_subprocess",
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (r *SubprocessRunner) buildArgs(req MergeRequest) []string {
|
||||
args := []string{"merge"}
|
||||
|
||||
for _, path := range req.InputTranscriptPaths {
|
||||
args = append(args, "--input-file", path)
|
||||
}
|
||||
args = append(args, "--output-file", req.OutputMergedTranscriptPath)
|
||||
|
||||
if r.report {
|
||||
args = append(args, "--report-file", req.ReportPath)
|
||||
}
|
||||
if strings.TrimSpace(req.SpeakersPath) != "" {
|
||||
args = append(args, "--speakers", req.SpeakersPath)
|
||||
}
|
||||
if strings.TrimSpace(req.AutocorrectPath) != "" {
|
||||
args = append(args, "--autocorrect", req.AutocorrectPath)
|
||||
}
|
||||
|
||||
args = append(args, "--output-schema", r.outputSchema)
|
||||
if r.coalesceGap != nil {
|
||||
args = append(args, "--coalesce-gap", strconv.FormatFloat(*r.coalesceGap, 'f', -1, 64))
|
||||
}
|
||||
|
||||
return args
|
||||
}
|
||||
|
||||
func (r *SubprocessRunner) buildEnvOverrides() map[string]string {
|
||||
out := map[string]string{}
|
||||
if r.env.OverlapWordRunGap != nil {
|
||||
out["SERIATIM_OVERLAP_WORD_RUN_GAP"] = strconv.FormatFloat(*r.env.OverlapWordRunGap, 'f', -1, 64)
|
||||
}
|
||||
if r.env.OverlapWordRunReorderWindow != nil {
|
||||
out["SERIATIM_OVERLAP_WORD_RUN_REORDER_WINDOW"] = strconv.FormatFloat(*r.env.OverlapWordRunReorderWindow, 'f', -1, 64)
|
||||
}
|
||||
if r.env.BackchannelMaxDuration != nil {
|
||||
out["SERIATIM_BACKCHANNEL_MAX_DURATION"] = strconv.FormatFloat(*r.env.BackchannelMaxDuration, 'f', -1, 64)
|
||||
}
|
||||
if r.env.FillerMaxDuration != nil {
|
||||
out["SERIATIM_FILLER_MAX_DURATION"] = strconv.FormatFloat(*r.env.FillerMaxDuration, 'f', -1, 64)
|
||||
}
|
||||
if len(out) == 0 {
|
||||
return nil
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (r *SubprocessRunner) writeInvocationConfig(req MergeRequest, args []string) error {
|
||||
payload := map[string]any{
|
||||
"schema": "seriatim.generated.v1",
|
||||
"binary": r.binary,
|
||||
"args": args,
|
||||
"timeout": r.timeout.String(),
|
||||
"output_schema": r.outputSchema,
|
||||
"report_enabled": r.report,
|
||||
"input_transcript_paths": req.InputTranscriptPaths,
|
||||
"output_path": req.OutputMergedTranscriptPath,
|
||||
}
|
||||
if req.ReportPath != "" {
|
||||
payload["report_path"] = req.ReportPath
|
||||
}
|
||||
if req.SpeakersPath != "" {
|
||||
payload["speakers_path"] = req.SpeakersPath
|
||||
}
|
||||
if req.AutocorrectPath != "" {
|
||||
payload["autocorrect_path"] = req.AutocorrectPath
|
||||
}
|
||||
if r.coalesceGap != nil {
|
||||
payload["coalesce_gap"] = *r.coalesceGap
|
||||
}
|
||||
|
||||
return subprocess.WriteYAMLAtomic(req.GeneratedConfigPath, payload, 0o644)
|
||||
}
|
||||
|
||||
func validateJSONFile(path string) error {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read file: %w", err)
|
||||
}
|
||||
var v any
|
||||
if err := json.Unmarshal(data, &v); err != nil {
|
||||
return fmt.Errorf("parse json: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user