openvibers/openvibe.link Commits
CLI and service: pair, run, install, uninstall, status, stop, resume, plugins
Changes
nnew file mode 100644@@@ -0,0 +1,498 @@++// Command openvibe-node connects a computer, server, Raspberry Pi or robot to OpenVibe.++//++// openvibe-node pair <CODE> redeem a one-time pairing code and store the device credential++// openvibe-node run run in the foreground (this is also what the service runs)++// openvibe-node install install and start the system service++// openvibe-node uninstall stop and remove the system service++// openvibe-node status show the link, the plugins, the stop latch and video++// openvibe-node stop local kill switch: stop every actuator and hold it stopped++// openvibe-node resume release the local kill switch (and a server e-stop)++// openvibe-node plugins list plugins and their capabilities++// openvibe-node version++package main++++import (++ "context"++ "encoding/json"++ "errors"++ "flag"++ "fmt"++ "io"++ "log/slog"++ "os"++ "path/filepath"++ "runtime"++ "sort"++ "strings"++ "time"++++ "github.com/OpenVibers/OpenVibe.Node/internal/config"++ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"++ "github.com/OpenVibers/OpenVibe.Node/internal/link"++ "github.com/OpenVibers/OpenVibe.Node/internal/localctl"++ "github.com/OpenVibers/OpenVibe.Node/internal/node"++ "github.com/OpenVibers/OpenVibe.Node/internal/plugins"++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"++ "github.com/OpenVibers/OpenVibe.Node/internal/safety"++ "github.com/OpenVibers/OpenVibe.Node/internal/service"++)++++// version is set at build time: -ldflags "-X main.version=v0.1.0".++var version = "dev"++++const usage = `openvibe-node %s: connect this device to OpenVibe.++++Usage:++ openvibe-node pair <CODE> [--server URL] [--kind onboard|bridge] [--force]++ openvibe-node run [--dry-run]++ openvibe-node install [--user NAME]++ openvibe-node uninstall++ openvibe-node status [--json]++ openvibe-node stop stop every actuator now and hold them stopped++ openvibe-node resume release the stop++ openvibe-node plugins [--json]++ openvibe-node version++++Global flags (before or after the command):++ --home DIR keep config and state in DIR (also OPENVIBE_NODE_HOME)++ --log-level LVL debug, info, warn or error (default info)++`++++func main() {++ os.Exit(run(os.Args[1:], os.Stdout, os.Stderr))++}++++type globals struct {++ home string++ logLevel string++ paths config.Paths++ log *slog.Logger++}++++func run(args []string, stdout, stderr io.Writer) int {++ if len(args) == 0 || args[0] == "-h" || args[0] == "--help" || args[0] == "help" {++ fmt.Fprintf(stdout, usage, version)++ return 0++ }++ cmd, rest := args[0], args[1:]++ fs := flag.NewFlagSet(cmd, flag.ContinueOnError)++ fs.SetOutput(stderr)++ g := &globals{}++ fs.StringVar(&g.home, "home", "", "config and state directory")++ fs.StringVar(&g.logLevel, "log-level", "info", "log level")++ var (++ server = fs.String("server", "", "OpenVibe.Bot origin (pair)")++ kind = fs.String("kind", "", "onboard or bridge (pair)")++ force = fs.Bool("force", false, "pair again even if already paired")++ dryRun = fs.Bool("dry-run", false, "run only the dry-run plugin and the test pattern (run)")++ asJSON = fs.Bool("json", false, "print JSON (status, plugins)")++ svcUser = fs.String("user", "", "run the service as this account (install)")++ )++ // Accept flags anywhere: split positionals out.++ var positional []string++ for len(rest) > 0 {++ if err := fs.Parse(rest); err != nil {++ return 2++ }++ rest = fs.Args()++ if len(rest) > 0 {++ positional = append(positional, rest[0])++ rest = rest[1:]++ }++ }++ g.paths = config.DefaultPaths(g.home)++ g.log = newLogger(stderr, g.logLevel)++++ var err error++ switch cmd {++ case "pair":++ if len(positional) != 1 {++ fmt.Fprintln(stderr, "usage: openvibe-node pair <CODE>")++ return 2++ }++ err = cmdPair(g, positional[0], *server, *kind, *force, stdout)++ case "run":++ err = cmdRun(g, *dryRun)++ case "install":++ err = cmdInstall(g, *svcUser, stdout)++ case "uninstall":++ err = service.Control(serviceOptions(g, ""), "stop")++ if uerr := service.Control(serviceOptions(g, ""), "uninstall"); uerr != nil {++ err = uerr++ } else {++ err = nil++ fmt.Fprintln(stdout, "Service removed. Config and credential are kept in", g.paths.ConfigDir)++ }++ case "status":++ err = cmdStatus(g, *asJSON, stdout)++ case "stop":++ err = cmdStop(g, stdout)++ case "resume":++ err = cmdResume(g, stdout)++ case "plugins":++ err = cmdPlugins(g, *asJSON, stdout)++ case "version", "--version":++ fmt.Fprintf(stdout, "openvibe-node %s %s/%s\n", version, runtime.GOOS, runtime.GOARCH)++ default:++ fmt.Fprintf(stderr, "unknown command %q\n\n", cmd)++ fmt.Fprintf(stderr, usage, version)++ return 2++ }++ if err != nil {++ fmt.Fprintln(stderr, "error:", err)++ return 1++ }++ return 0++}++++func newLogger(w io.Writer, level string) *slog.Logger {++ var l slog.Level++ if err := l.UnmarshalText([]byte(level)); err != nil {++ l = slog.LevelInfo++ }++ return slog.New(slog.NewTextHandler(w, &slog.HandlerOptions{Level: l}))++}++++func loadConfig(g *globals) (*config.Config, error) {++ return config.Load(g.paths.ConfigFile())++}++++func pluginSpecs(cfg *config.Config, paths config.Paths) []plugins.Spec {++ var out []plugins.Spec++ for _, p := range cfg.Plugins {++ if p.Disabled {++ continue++ }++ var env []string++ for k, v := range p.Env {++ env = append(env, k+"="+v)++ }++ out = append(out, plugins.Spec{Name: p.Name, Argv: cfg.PluginCommand(p, paths), Env: env, Dir: p.Dir, Config: p.Config})++ }++ return out++}++++// probeAll asks every plugin to describe itself without touching hardware.++func probeAll(cfg *config.Config, paths config.Paths, log *slog.Logger) (map[string]*plugins.Describe, []string) {++ out := map[string]*plugins.Describe{}++ var failed []string++ for _, s := range pluginSpecs(cfg, paths) {++ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)++ d, err := plugins.Probe(ctx, s)++ cancel()++ if err != nil {++ log.Warn("plugin did not describe itself", "plugin", s.Name, "err", err)++ failed = append(failed, s.Name)++ continue++ }++ out[s.Name] = d++ }++ return out, failed++}++++func cmdPair(g *globals, code, server, kind string, force bool, stdout io.Writer) error {++ code, err := link.NormalizeCode(code)++ if err != nil {++ return err++ }++ if err := g.paths.Ensure(); err != nil {++ return fmt.Errorf("cannot create %s (run with sudo, or use --home): %w", g.paths.ConfigDir, err)++ }++ if old, err := credentials.Load(g.paths.CredentialFile(), nil); err == nil && !force {++ return fmt.Errorf("already paired as %s; use --force to pair again (the old credential stops working once the owner removes it)", old.DeviceID)++ }++ cfg, err := loadConfig(g)++ if err != nil {++ return err++ }++ _, statErr := os.Stat(g.paths.ConfigFile())++ if server != "" {++ cfg.Server = server++ }++ if kind != "" {++ cfg.DeviceKind = kind++ }++ if err := cfg.Validate(); err != nil {++ return err++ }++ descs, _ := probeAll(cfg, g.paths, g.log)++ req := protocol.PairRequest{Code: code, AgentVersion: version, DeviceKind: cfg.DeviceKind, Drivers: []string{},++ Capabilities: map[string]any{}, OS: runtime.GOOS, Arch: runtime.GOARCH}++ req.Hostname, _ = os.Hostname()++ names := make([]string, 0, len(descs))++ for n := range descs {++ names = append(names, n)++ }++ sort.Strings(names)++ for _, n := range names {++ req.Drivers = append(req.Drivers, descs[n].Driver)++ req.Capabilities[n] = descs[n].Capabilities++ }++ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)++ defer cancel()++ creds, err := link.Pair(ctx, nil, cfg.Server, req)++ if err != nil {++ return err++ }++ if err := credentials.Save(g.paths.CredentialFile(), creds); err != nil {++ return err++ }++ chownLikeDir(g.paths.CredentialFile(), g.paths.ConfigDir)++ if statErr != nil || server != "" || kind != "" {++ if err := config.Save(g.paths.ConfigFile(), cfg); err != nil {++ return err++ }++ chownLikeDir(g.paths.ConfigFile(), g.paths.ConfigDir)++ }++ fmt.Fprintf(stdout, "Paired as %s. Credential saved to %s (mode 600).\n", creds.DeviceID, g.paths.CredentialFile())++ fmt.Fprintln(stdout, "Confirm the device on openvibe.bot, then start it: `openvibe-node run` or `sudo openvibe-node install`.")++ if service.Status(serviceOptions(g, "")) == "running" {++ _ = service.Control(serviceOptions(g, ""), "restart")++ fmt.Fprintln(stdout, "The service was running and has been restarted with the new credential.")++ }++ return nil++}++++func cmdRun(g *globals, dryRun bool) error {++ if err := g.paths.Ensure(); err != nil {++ return fmt.Errorf("cannot create %s (run with sudo, or use --home): %w", g.paths.StateDir, err)++ }++ cfg, err := loadConfig(g)++ if err != nil {++ return err++ }++ if dryRun {++ var keep []config.PluginConfig++ for _, p := range cfg.Plugins {++ if p.Name == "dryrun" {++ keep = append(keep, p)++ }++ }++ if len(keep) == 0 {++ keep = []config.PluginConfig{{Name: "dryrun"}}++ }++ cfg.Plugins = keep++ cfg.Video.Source = "test"++ }++ creds, err := credentials.Load(g.paths.CredentialFile(), func(s string) { g.log.Warn(s) })++ if err != nil && !errors.Is(err, credentials.ErrNotPaired) {++ return err++ }++ n, err := node.New(node.Options{Config: cfg, Paths: g.paths, Creds: creds, Log: g.log, Version: version})++ if err != nil {++ return err++ }++ g.log.Info("openvibe-node starting", "version", version, "config", g.paths.ConfigDir, "plugins", len(cfg.Plugins))++ return service.Run(serviceOptions(g, ""), n.Run)++}++++func serviceOptions(g *globals, user string) service.Options {++ args := []string{"run"}++ if g.home != "" {++ abs, _ := filepath.Abs(g.home)++ args = append(args, "--home", abs)++ }++ if g.logLevel != "" && g.logLevel != "info" {++ args = append(args, "--log-level", g.logLevel)++ }++ return service.Options{Arguments: args, UserName: user}++}++++func cmdInstall(g *globals, user string, stdout io.Writer) error {++ if err := g.paths.Ensure(); err != nil {++ return fmt.Errorf("cannot create %s (run with sudo): %w", g.paths.ConfigDir, err)++ }++ if user != "" {++ if err := chownTree(user, g.paths.ConfigDir, g.paths.StateDir); err != nil {++ return err++ }++ }++ opt := serviceOptions(g, user)++ if err := service.Control(opt, "install"); err != nil {++ return fmt.Errorf("install: %w (are you root?)", err)++ }++ if err := service.Control(opt, "start"); err != nil {++ return fmt.Errorf("installed, but could not start: %w", err)++ }++ fmt.Fprintln(stdout, "Installed and started the openvibe-node service.")++ if _, err := credentials.Load(g.paths.CredentialFile(), nil); err != nil {++ fmt.Fprintln(stdout, "Not paired yet: run `sudo openvibe-node pair <CODE>`.")++ }++ return nil++}++++func cmdStop(g *globals, stdout io.Writer) error {++ _, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdStop}, 2*time.Second)++ if err == nil {++ fmt.Fprintln(stdout, "Stopped. Every actuator is held stopped until `openvibe-node resume`.")++ return nil++ }++ // No running Node answered: latch it in the file, which a running Node polls and a starting Node obeys.++ if ferr := safety.WriteLocalFile(g.paths.LatchFile(), true); ferr != nil {++ return fmt.Errorf("could not reach the node (%v) nor write %s (%v); run with sudo", err, g.paths.LatchFile(), ferr)++ }++ fmt.Fprintln(stdout, "The node did not answer; the stop is latched in", g.paths.LatchFile())++ fmt.Fprintln(stdout, "A running node obeys it within a quarter second; a starting node starts stopped. Release with `openvibe-node resume`.")++ return nil++}++++func cmdResume(g *globals, stdout io.Writer) error {++ _, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdResume}, 2*time.Second)++ if err != nil {++ if ferr := safety.WriteLocalFile(g.paths.LatchFile(), false); ferr != nil {++ return fmt.Errorf("could not reach the node (%v) nor write %s (%v); run with sudo", err, g.paths.LatchFile(), ferr)++ }++ }++ fmt.Fprintln(stdout, "Resumed: the local stop and any server e-stop are released. Operators can drive again.")++ return nil++}++++func cmdStatus(g *globals, asJSON bool, stdout io.Writer) error {++ resp, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdStatus}, 2*time.Second)++ if err != nil {++ return offlineStatus(g, asJSON, stdout)++ }++ if asJSON {++ _, err := stdout.Write(append(resp.Data, '\n'))++ return err++ }++ var s node.Status++ if err := json.Unmarshal(resp.Data, &s); err != nil {++ return err++ }++ fmt.Fprintf(stdout, "OpenVibe Node %s, running for %s (service: %s)\n", s.Version, s.Uptime, service.Status(serviceOptions(g, "")))++ if s.Paired {++ fmt.Fprintf(stdout, "Device: %s on %s\n", s.DeviceID, s.Server)++ } else {++ fmt.Fprintln(stdout, "Device: not paired (openvibe-node pair <CODE>)")++ }++ link := "down"++ if s.Link.Connected {++ link = fmt.Sprintf("up (rtt %d ms)", s.Link.RTTMS)++ } else if s.Link.LastError != "" {++ link = "down: " + s.Link.LastError++ }++ fmt.Fprintf(stdout, "Link: %s, %d reconnects\n", link, s.Link.Reconnects)++ fmt.Fprintf(stdout, "Stop: %s\n", latchText(s.Latch))++ fmt.Fprintf(stdout, "Limits: speed %.2f, turn %.2f, command deadline ≤ %d ms\n", s.Limits.MaxSpeed, s.Limits.MaxTurn, s.Limits.MaxCommandMS)++ fmt.Fprintf(stdout, "Video: %s", s.Video.State)++ if s.Video.Source != "" {++ fmt.Fprintf(stdout, " from %s, %d frames", s.Video.Source, s.Video.Frames)++ }++ if s.Video.LastError != "" && s.Video.State != "live" {++ fmt.Fprintf(stdout, " (%s)", s.Video.LastError)++ }++ fmt.Fprintln(stdout)++ fmt.Fprintln(stdout, "Plugins:")++ for _, p := range s.Plugins {++ fmt.Fprintf(stdout, " %-15s %-9s %s %s", p.Name, p.State, p.Driver, p.Version)++ if p.Restarts > 0 {++ fmt.Fprintf(stdout, ", %d restarts", p.Restarts)++ }++ if p.Fault != nil {++ fmt.Fprintf(stdout, ", fault %s: %s", p.Fault.Code, p.Fault.Message)++ }++ fmt.Fprintln(stdout)++ }++ if b := s.Telemetry.Battery; b != nil && b.Percent != nil {++ fmt.Fprintf(stdout, "Battery: %.0f%%\n", *b.Percent)++ }++ return nil++}++++func latchText(l safety.LatchState) string {++ switch {++ case l.Local && l.Remote:++ return "STOPPED locally and by the server (" + l.RemoteReason + "); `openvibe-node resume` releases both"++ case l.Local:++ return "STOPPED locally; `openvibe-node resume` releases it"++ case l.Remote:++ return "E-STOPPED by the server (" + l.RemoteReason + "); the owner clears it on openvibe.bot"++ }++ return "clear"++}++++func offlineStatus(g *globals, asJSON bool, stdout io.Writer) error {++ type offline struct {++ Running bool `json:"running"`++ Service string `json:"service"`++ Paired bool `json:"paired"`++ DeviceID string `json:"device_id,omitempty"`++ Server string `json:"server,omitempty"`++ Latch safety.LatchState `json:"latch"`++ Config string `json:"config_dir"`++ }++ o := offline{Service: service.Status(serviceOptions(g, "")), Config: g.paths.ConfigDir}++ if c, err := credentials.Load(g.paths.CredentialFile(), nil); err == nil {++ o.Paired, o.DeviceID, o.Server = true, c.DeviceID, c.Server++ }++ if l, _ := safety.OpenLatch(g.paths.LatchFile()); l != nil {++ o.Latch = l.State()++ }++ if asJSON {++ return json.NewEncoder(stdout).Encode(o)++ }++ fmt.Fprintf(stdout, "OpenVibe Node %s is not running (service: %s).\n", version, o.Service)++ if o.Paired {++ fmt.Fprintf(stdout, "Device: %s on %s\n", o.DeviceID, o.Server)++ } else {++ fmt.Fprintln(stdout, "Device: not paired (openvibe-node pair <CODE>)")++ }++ fmt.Fprintf(stdout, "Stop: %s\n", latchText(o.Latch))++ fmt.Fprintf(stdout, "Config: %s\n", o.Config)++ return nil++}++++func cmdPlugins(g *globals, asJSON bool, stdout io.Writer) error {++ type entry struct {++ Name string `json:"name"`++ State string `json:"state"`++ Driver string `json:"driver,omitempty"`++ Version string `json:"version,omitempty"`++ Capabilities map[string]any `json:"capabilities,omitempty"`++ }++ var list []entry++ if resp, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdPlugins}, 2*time.Second); err == nil {++ var s node.Status++ if err := json.Unmarshal(resp.Data, &s); err != nil {++ return err++ }++ for _, p := range s.Plugins {++ list = append(list, entry{p.Name, p.State, p.Driver, p.Version, p.Capabilities})++ }++ } else {++ cfg, err := loadConfig(g)++ if err != nil {++ return err++ }++ descs, _ := probeAll(cfg, g.paths, g.log)++ for _, s := range pluginSpecs(cfg, g.paths) {++ e := entry{Name: s.Name, State: "not running"}++ if d := descs[s.Name]; d != nil {++ e.Driver, e.Version, e.Capabilities = d.Driver, d.Version, d.Capabilities++ } else {++ e.State = "not installed or broken (" + strings.Join(s.Argv, " ") + ")"++ }++ list = append(list, e)++ }++ }++ if asJSON {++ return json.NewEncoder(stdout).Encode(list)++ }++ for _, e := range list {++ fmt.Fprintf(stdout, "%s [%s] %s %s\n", e.Name, e.State, e.Driver, e.Version)++ keys := make([]string, 0, len(e.Capabilities))++ for k := range e.Capabilities {++ keys = append(keys, k)++ }++ sort.Strings(keys)++ for _, k := range keys {++ b, _ := json.Marshal(e.Capabilities[k])++ fmt.Fprintf(stdout, " %-9s %s\n", k, b)++ }++ }++ if len(list) == 0 {++ fmt.Fprintln(stdout, "No plugins configured in", g.paths.ConfigFile())++ }++ return nil++}nnew file mode 100644@@@ -0,0 +1,219 @@++package main++++import (++ "bytes"++ "encoding/json"++ "os"++ "os/exec"++ "path/filepath"++ "runtime"++ "strings"++ "syscall"++ "testing"++ "time"++++ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"++ "github.com/OpenVibers/OpenVibe.Node/internal/fakebot"++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"++ "github.com/OpenVibers/OpenVibe.Node/internal/safety"++ "github.com/OpenVibers/OpenVibe.Node/internal/video"++)++++// With OPENVIBE_NODE_MAIN=1 the test binary is openvibe-node itself.++func TestMain(m *testing.M) {++ if os.Getenv("OPENVIBE_NODE_MAIN") == "1" {++ os.Exit(run(os.Args[1:], os.Stdout, os.Stderr))++ }++ os.Exit(m.Run())++}++++func cli(t *testing.T, args ...string) (string, string, int) {++ t.Helper()++ var out, errb bytes.Buffer++ code := run(args, &out, &errb)++ return out.String(), errb.String(), code++}++++func shortHome(t *testing.T) string {++ d, err := os.MkdirTemp("", "ovn")++ if err != nil {++ t.Fatal(err)++ }++ t.Cleanup(func() { os.RemoveAll(d) })++ return d++}++++func TestHelpAndVersion(t *testing.T) {++ if out, _, code := cli(t, "help"); code != 0 || !strings.Contains(out, "openvibe-node pair <CODE>") {++ t.Fatalf("%d %s", code, out)++ }++ if out, _, code := cli(t, "version"); code != 0 || !strings.Contains(out, runtime.GOOS) {++ t.Fatalf("%d %s", code, out)++ }++ if _, _, code := cli(t, "fly"); code != 2 {++ t.Fatal("unknown command accepted")++ }++}++++func pythonPath(t *testing.T) string {++ root, _ := filepath.Abs("../..")++ return filepath.Join(root, "plugins/sdk") + string(os.PathListSeparator) + filepath.Join(root, "plugins/dryrun")++}++++func TestPair(t *testing.T) {++ srv := fakebot.New()++ defer srv.Close()++ srv.AddCode("ABCD2345")++ home := shortHome(t)++ t.Setenv("PYTHONPATH", pythonPath(t))++ if _, errOut, code := cli(t, "pair", "nope", "--home", home); code == 0 || !strings.Contains(errOut, "8 letters") {++ t.Fatalf("bad code accepted: %s", errOut)++ }++ out, errOut, code := cli(t, "pair", "abcd-2345", "--home", home, "--server", srv.URL())++ if code != 0 {++ t.Fatalf("pair failed: %s", errOut)++ }++ c, err := credentials.Load(filepath.Join(home, "credential.json"), nil)++ if err != nil {++ t.Fatal(err)++ }++ if !strings.Contains(out, c.DeviceID) || strings.Contains(out+errOut, c.Credential.Reveal()) {++ t.Fatalf("output: %s %s", out, errOut)++ }++ st, _ := os.Stat(filepath.Join(home, "credential.json"))++ if st.Mode().Perm() != 0o600 {++ t.Fatalf("mode %o", st.Mode().Perm())++ }++ if _, err := os.Stat(filepath.Join(home, "config.json")); err != nil {++ t.Fatal("config not written")++ }++ reqs := srv.PairRequests()++ if len(reqs) != 1 || reqs[0].Code != "ABCD2345" || reqs[0].AgentVersion != version {++ t.Fatalf("%+v", reqs)++ }++ if _, err := exec.LookPath("python3"); err == nil && len(reqs[0].Drivers) != 1 {++ t.Fatalf("dry-run plugin not probed: %+v", reqs[0])++ }++ if _, errOut, code := cli(t, "pair", "ABCD2345", "--home", home); code == 0 || !strings.Contains(errOut, "already paired") {++ t.Fatalf("second pair: %s", errOut)++ }++ // status without a running node reads the files and never prints the secret.++ out, _, code = cli(t, "status", "--home", home)++ if code != 0 || !strings.Contains(out, "not running") || !strings.Contains(out, c.DeviceID) || strings.Contains(out, c.Credential.Reveal()) {++ t.Fatalf("%d %s", code, out)++ }++}++++func TestStopWithoutRunningNodeLatchesFile(t *testing.T) {++ home := shortHome(t)++ if out, errOut, code := cli(t, "stop", "--home", home); code != 0 || !strings.Contains(out, "latched") {++ t.Fatalf("%s %s", out, errOut)++ }++ l, _ := safety.OpenLatch(filepath.Join(home, "latch.json"))++ if !l.State().Local {++ t.Fatal("not latched")++ }++ cli(t, "resume", "--home", home)++ l, _ = safety.OpenLatch(filepath.Join(home, "latch.json"))++ if l.State().Stopped() {++ t.Fatal("not resumed")++ }++}++++// TestRunDryRun is the acceptance path: `openvibe-node run --dry-run` pairs with the fake server, publishes the test++// pattern, obeys commands, obeys the local kill switch and stops cleanly on SIGTERM.++func TestRunDryRun(t *testing.T) {++ if runtime.GOOS == "windows" {++ t.Skip("unix only")++ }++ if _, err := exec.LookPath("python3"); err != nil {++ t.Skip("python3 not installed")++ }++ srv := fakebot.New()++ defer srv.Close()++ srv.AddCode("WXYZ6789")++ home := shortHome(t)++ t.Setenv("PYTHONPATH", pythonPath(t))++ if _, errOut, code := cli(t, "pair", "WXYZ-6789", "--home", home, "--server", srv.URL()); code != 0 {++ t.Fatal(errOut)++ }++ c, _ := credentials.Load(filepath.Join(home, "credential.json"), nil)++ whip := &video.WHIPReceiver{PublishKey: c.PublishKey.Reveal()}++ srv.SetWHIP(whip)++ defer whip.Close()++++ exe, _ := os.Executable()++ cmd := exec.Command(exe, "run", "--dry-run", "--home", home)++ cmd.Env = append(os.Environ(), "OPENVIBE_NODE_MAIN=1")++ var logs bytes.Buffer++ cmd.Stdout, cmd.Stderr = &logs, &logs++ if err := cmd.Start(); err != nil {++ t.Fatal(err)++ }++ exited := make(chan error, 1)++ go func() { exited <- cmd.Wait() }()++ defer func() {++ if cmd.ProcessState == nil {++ cmd.Process.Kill()++ }++ }()++++ conn, err := srv.NextConn(20 * time.Second)++ if err != nil {++ t.Fatalf("%v\n%s", err, logs.String())++ }++ if _, err := conn.Expect(protocol.TypeStatus, 20*time.Second, func(m protocol.Message) bool {++ s := m.(protocol.Status)++ return len(s.Drivers) == 1 && s.Drivers[0].State == "ready"++ }); err != nil {++ t.Fatalf("%v\n%s", err, logs.String())++ }++ send := func(id, kind, value string) protocol.Message {++ t.Helper()++ conn.Send(protocol.Command{ID: id, Kind: kind, Value: json.RawMessage(value), DeadlineMS: 300})++ r, err := conn.Reply(id, 5*time.Second)++ if err != nil {++ t.Fatal(err)++ }++ return r++ }++ if _, ok := send("c1", "drive", `{"throttle":0.4}`).(protocol.Ack); !ok {++ t.Fatal("drive not acked")++ }++ out, errOut, code := cli(t, "status", "--home", home)++ if code != 0 || !strings.Contains(out, "Link: up") || !strings.Contains(out, "dryrun") || strings.Contains(out, c.Credential.Reveal()) {++ t.Fatalf("status: %s %s", out, errOut)++ }++ if out, _, code := cli(t, "stop", "--home", home); code != 0 || !strings.Contains(out, "held stopped") {++ t.Fatalf("stop: %s", out)++ }++ if n, ok := send("c2", "drive", `{"throttle":0.4}`).(protocol.Nack); !ok || n.FaultCode != protocol.FaultLocalStop {++ t.Fatal("drive accepted after the local stop")++ }++ cli(t, "resume", "--home", home)++ if _, ok := send("c3", "drive", `{"throttle":0.4}`).(protocol.Ack); !ok {++ t.Fatal("drive refused after resume")++ }++ deadline := time.Now().Add(20 * time.Second)++ for whip.Packets.Load() < 10 {++ if time.Now().After(deadline) {++ t.Fatalf("no test pattern received\n%s", logs.String())++ }++ time.Sleep(50 * time.Millisecond)++ }++ cmd.Process.Signal(syscall.SIGTERM)++ select {++ case err := <-exited:++ if err != nil {++ t.Fatalf("exit: %v\n%s", err, logs.String())++ }++ case <-time.After(15 * time.Second):++ t.Fatal("did not stop on SIGTERM")++ }++ if strings.Contains(logs.String(), c.Credential.Reveal()) || strings.Contains(logs.String(), c.PublishKey.Reveal()) {++ t.Fatal("a secret reached the logs")++ }++ if !strings.Contains(logs.String(), "stopping every actuator") {++ t.Fatalf("no shutdown stop in logs:\n%s", logs.String())++ }++}nnew file mode 100644@@@ -0,0 +1,45 @@++//go:build !windows++++package main++++import (++ "fmt"++ "os"++ "os/user"++ "path/filepath"++ "strconv"++ "syscall"++)++++// chownLikeDir gives a file written by root (pair) to the account that owns its directory (the service account).++func chownLikeDir(path, dir string) {++ st, err := os.Stat(dir)++ if err != nil {++ return++ }++ if sys, ok := st.Sys().(*syscall.Stat_t); ok && os.Geteuid() == 0 {++ _ = os.Chown(path, int(sys.Uid), int(sys.Gid))++ }++}++++// chownTree hands the config and state directories to the service account.++func chownTree(name string, dirs ...string) error {++ u, err := user.Lookup(name)++ if err != nil {++ return fmt.Errorf("no account %q (create it first, e.g. with install.sh): %w", name, err)++ }++ uid, _ := strconv.Atoi(u.Uid)++ gid, _ := strconv.Atoi(u.Gid)++ for _, d := range dirs {++ err := filepath.Walk(d, func(p string, _ os.FileInfo, err error) error {++ if err != nil {++ return err++ }++ return os.Lchown(p, uid, gid)++ })++ if err != nil {++ return err++ }++ }++ return nil++}nnew file mode 100644@@@ -0,0 +1,11 @@++//go:build windows++++package main++++import "errors"++++func chownLikeDir(path, dir string) {}++++func chownTree(name string, dirs ...string) error {++ return errors.New("--user is not supported on Windows; the service runs as LocalSystem")++}@@@ -4,6 +4,7 @@ go 1.24.7require (github.com/gorilla/websocket v1.5.3++ github.com/kardianos/service v1.3.0github.com/pion/ice/v4 v4.4.4github.com/pion/interceptor v0.1.49github.com/pion/rtcp v1.2.18@@@ -2,6 +2,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=++github.com/kardianos/service v1.3.0 h1:/LGy+xPP2TM+GLTiCZ2di7cy0Jd/qrawlTUfqKYFdTI=++github.com/kardianos/service v1.3.0/go.mod h1:E4V9ufUuY82F7Ztlu1eN9VXWIQxg8NoLQlmFe0MtrXc=github.com/pion/datachannel v1.6.3 h1:HZAYLpunVI0xi057ZiANl558d0uSpomcWw4UEPz3qXs=github.com/pion/datachannel v1.6.3/go.mod h1:oRmkXVVXxkUbqa8MatgdNHW29kvtQY3HUdjJQsfv2GE=github.com/pion/dtls/v3 v3.1.9 h1:rpeycmLIkc4krpk1IxP7+39o11QdCXbmV9+FGi9yZJ8=nnew file mode 100644@@@ -0,0 +1,137 @@++// Package service installs and runs the Node as a system service (systemd, launchd or the Windows service manager)++// through kardianos/service. The same code path runs `openvibe-node run` in a terminal: Ctrl-C stops it cleanly.++package service++++import (++ "context"++ "errors"++ "sync"++ "time"++++ ks "github.com/kardianos/service"++)++++const (++ Name = "openvibe-node"++ DisplayName = "OpenVibe Node"++ Description = "Connects this device to OpenVibe (robots, cameras, computer control) over one outbound link."++)++++// Program is the body of the service: run until ctx ends.++type Program func(ctx context.Context) error++++type program struct {++ run Program++ mu sync.Mutex++ cancel context.CancelFunc++ done chan struct{}++ err error++}++++func (p *program) Start(s ks.Service) error {++ ctx, cancel := context.WithCancel(context.Background())++ p.mu.Lock()++ p.cancel, p.done = cancel, make(chan struct{})++ p.mu.Unlock()++ go func() {++ err := p.run(ctx)++ p.mu.Lock()++ p.err = err++ p.mu.Unlock()++ close(p.done)++ if err != nil && ctx.Err() == nil {++ // The Node ended on its own (for example a second instance): let the service manager restart it.++ _ = s.Stop()++ }++ }()++ return nil++}++++// Stop cancels the Node and waits for it to stop every actuator (at most 15 s).++func (p *program) Stop(ks.Service) error {++ p.mu.Lock()++ cancel, done := p.cancel, p.done++ p.mu.Unlock()++ if cancel == nil {++ return nil++ }++ cancel()++ select {++ case <-done:++ case <-time.After(15 * time.Second):++ return errors.New("timed out waiting for the node to stop")++ }++ return nil++}++++// Options for the service definition.++type Options struct {++ // Arguments passed to the executable by the service manager (["run"] plus --home when set).++ Arguments []string++ // UserName runs the service as this account (Linux and macOS); empty = the service manager's default (root).++ UserName string++}++++func config(opt Options) *ks.Config {++ return &ks.Config{++ Name: Name, DisplayName: DisplayName, Description: Description, Arguments: opt.Arguments, UserName: opt.UserName,++ Option: ks.KeyValue{++ "Restart": "always", // systemd++ "SuccessExitStatus": "0",++ "KeepAlive": true, // launchd++ "RunAtLoad": true,++ "OnFailure": "restart", // Windows++ "OnFailureDelayDuration": "5s",++ "DelayedAutoStart": false,++ },++ }++}++++// Run runs program under the service manager, or in the foreground when started from a terminal.++func Run(opt Options, run Program) error {++ p := &program{run: run}++ s, err := ks.New(p, config(opt))++ if err != nil {++ return err++ }++ if err := s.Run(); err != nil {++ return err++ }++ p.mu.Lock()++ defer p.mu.Unlock()++ return p.err++}++++// Control performs install, uninstall, start, stop or restart on the service definition.++func Control(opt Options, action string) error {++ s, err := ks.New(&program{}, config(opt))++ if err != nil {++ return err++ }++ return ks.Control(s, action)++}++++// Status reports running, stopped, not installed or unknown.++func Status(opt Options) string {++ s, err := ks.New(&program{}, config(opt))++ if err != nil {++ return "unknown"++ }++ st, err := s.Status()++ if errors.Is(err, ks.ErrNotInstalled) {++ return "not installed"++ }++ if err != nil {++ return "unknown"++ }++ switch st {++ case ks.StatusRunning:++ return "running"++ case ks.StatusStopped:++ return "stopped"++ }++ return "unknown"++}++++// Interactive reports whether the process runs in a terminal rather than under a service manager.++func Interactive() bool { return ks.Interactive() }nnew file mode 100644@@@ -0,0 +1,35 @@++package service++++import (++ "context"++ "testing"++ "time"++)++++func TestProgramStartStop(t *testing.T) {++ stopped := make(chan struct{})++ p := &program{run: func(ctx context.Context) error {++ <-ctx.Done()++ time.Sleep(10 * time.Millisecond) // stopping actuators++ close(stopped)++ return nil++ }}++ if err := p.Start(nil); err != nil {++ t.Fatal(err)++ }++ if err := p.Stop(nil); err != nil {++ t.Fatal(err)++ }++ select {++ case <-stopped:++ default:++ t.Fatal("Stop returned before the program finished")++ }++}++++func TestConfig(t *testing.T) {++ c := config(Options{Arguments: []string{"run"}})++ if c.Name != "openvibe-node" || c.Arguments[0] != "run" || c.Option["Restart"] != "always" {++ t.Fatalf("%+v", c)++ }++}
new file mode 100644
new file mode 100644
@@ -0,0 +1,498 @@
@@ -0,0 +1,498 @@
+// Command openvibe-node connects a computer, server, Raspberry Pi or robot to OpenVibe.
+//
+// openvibe-node pair <CODE> redeem a one-time pairing code and store the device credential
+// openvibe-node run run in the foreground (this is also what the service runs)
+// openvibe-node install install and start the system service
+// openvibe-node uninstall stop and remove the system service
+// openvibe-node status show the link, the plugins, the stop latch and video
+// openvibe-node stop local kill switch: stop every actuator and hold it stopped
+// openvibe-node resume release the local kill switch (and a server e-stop)
+// openvibe-node plugins list plugins and their capabilities
+// openvibe-node version
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "flag"
+ "fmt"
+ "io"
+ "log/slog"
+ "os"
+ "path/filepath"
+ "runtime"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/config"
+ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"
+ "github.com/OpenVibers/OpenVibe.Node/internal/link"
+ "github.com/OpenVibers/OpenVibe.Node/internal/localctl"
+ "github.com/OpenVibers/OpenVibe.Node/internal/node"
+ "github.com/OpenVibers/OpenVibe.Node/internal/plugins"
+ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
+ "github.com/OpenVibers/OpenVibe.Node/internal/safety"
+ "github.com/OpenVibers/OpenVibe.Node/internal/service"
+)
+
+// version is set at build time: -ldflags "-X main.version=v0.1.0".
+var version = "dev"
+
+const usage = `openvibe-node %s: connect this device to OpenVibe.
+
+Usage:
+ openvibe-node pair <CODE> [--server URL] [--kind onboard|bridge] [--force]
+ openvibe-node run [--dry-run]
+ openvibe-node install [--user NAME]
+ openvibe-node uninstall
+ openvibe-node status [--json]
+ openvibe-node stop stop every actuator now and hold them stopped
+ openvibe-node resume release the stop
+ openvibe-node plugins [--json]
+ openvibe-node version
+
+Global flags (before or after the command):
+ --home DIR keep config and state in DIR (also OPENVIBE_NODE_HOME)
+ --log-level LVL debug, info, warn or error (default info)
+`
+
+func main() {
+ os.Exit(run(os.Args[1:], os.Stdout, os.Stderr))
+}
+
+type globals struct {
+ home string
+ logLevel string
+ paths config.Paths
+ log *slog.Logger
+}
+
+func run(args []string, stdout, stderr io.Writer) int {
+ if len(args) == 0 || args[0] == "-h" || args[0] == "--help" || args[0] == "help" {
+ fmt.Fprintf(stdout, usage, version)
+ return 0
+ }
+ cmd, rest := args[0], args[1:]
+ fs := flag.NewFlagSet(cmd, flag.ContinueOnError)
+ fs.SetOutput(stderr)
+ g := &globals{}
+ fs.StringVar(&g.home, "home", "", "config and state directory")
+ fs.StringVar(&g.logLevel, "log-level", "info", "log level")
+ var (
+ server = fs.String("server", "", "OpenVibe.Bot origin (pair)")
+ kind = fs.String("kind", "", "onboard or bridge (pair)")
+ force = fs.Bool("force", false, "pair again even if already paired")
+ dryRun = fs.Bool("dry-run", false, "run only the dry-run plugin and the test pattern (run)")
+ asJSON = fs.Bool("json", false, "print JSON (status, plugins)")
+ svcUser = fs.String("user", "", "run the service as this account (install)")
+ )
+ // Accept flags anywhere: split positionals out.
+ var positional []string
+ for len(rest) > 0 {
+ if err := fs.Parse(rest); err != nil {
+ return 2
+ }
+ rest = fs.Args()
+ if len(rest) > 0 {
+ positional = append(positional, rest[0])
+ rest = rest[1:]
+ }
+ }
+ g.paths = config.DefaultPaths(g.home)
+ g.log = newLogger(stderr, g.logLevel)
+
+ var err error
+ switch cmd {
+ case "pair":
+ if len(positional) != 1 {
+ fmt.Fprintln(stderr, "usage: openvibe-node pair <CODE>")
+ return 2
+ }
+ err = cmdPair(g, positional[0], *server, *kind, *force, stdout)
+ case "run":
+ err = cmdRun(g, *dryRun)
+ case "install":
+ err = cmdInstall(g, *svcUser, stdout)
+ case "uninstall":
+ err = service.Control(serviceOptions(g, ""), "stop")
+ if uerr := service.Control(serviceOptions(g, ""), "uninstall"); uerr != nil {
+ err = uerr
+ } else {
+ err = nil
+ fmt.Fprintln(stdout, "Service removed. Config and credential are kept in", g.paths.ConfigDir)
+ }
+ case "status":
+ err = cmdStatus(g, *asJSON, stdout)
+ case "stop":
+ err = cmdStop(g, stdout)
+ case "resume":
+ err = cmdResume(g, stdout)
+ case "plugins":
+ err = cmdPlugins(g, *asJSON, stdout)
+ case "version", "--version":
+ fmt.Fprintf(stdout, "openvibe-node %s %s/%s\n", version, runtime.GOOS, runtime.GOARCH)
+ default:
+ fmt.Fprintf(stderr, "unknown command %q\n\n", cmd)
+ fmt.Fprintf(stderr, usage, version)
+ return 2
+ }
+ if err != nil {
+ fmt.Fprintln(stderr, "error:", err)
+ return 1
+ }
+ return 0
+}
+
+func newLogger(w io.Writer, level string) *slog.Logger {
+ var l slog.Level
+ if err := l.UnmarshalText([]byte(level)); err != nil {
+ l = slog.LevelInfo
+ }
+ return slog.New(slog.NewTextHandler(w, &slog.HandlerOptions{Level: l}))
+}
+
+func loadConfig(g *globals) (*config.Config, error) {
+ return config.Load(g.paths.ConfigFile())
+}
+
+func pluginSpecs(cfg *config.Config, paths config.Paths) []plugins.Spec {
+ var out []plugins.Spec
+ for _, p := range cfg.Plugins {
+ if p.Disabled {
+ continue
+ }
+ var env []string
+ for k, v := range p.Env {
+ env = append(env, k+"="+v)
+ }
+ out = append(out, plugins.Spec{Name: p.Name, Argv: cfg.PluginCommand(p, paths), Env: env, Dir: p.Dir, Config: p.Config})
+ }
+ return out
+}
+
+// probeAll asks every plugin to describe itself without touching hardware.
+func probeAll(cfg *config.Config, paths config.Paths, log *slog.Logger) (map[string]*plugins.Describe, []string) {
+ out := map[string]*plugins.Describe{}
+ var failed []string
+ for _, s := range pluginSpecs(cfg, paths) {
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ d, err := plugins.Probe(ctx, s)
+ cancel()
+ if err != nil {
+ log.Warn("plugin did not describe itself", "plugin", s.Name, "err", err)
+ failed = append(failed, s.Name)
+ continue
+ }
+ out[s.Name] = d
+ }
+ return out, failed
+}
+
+func cmdPair(g *globals, code, server, kind string, force bool, stdout io.Writer) error {
+ code, err := link.NormalizeCode(code)
+ if err != nil {
+ return err
+ }
+ if err := g.paths.Ensure(); err != nil {
+ return fmt.Errorf("cannot create %s (run with sudo, or use --home): %w", g.paths.ConfigDir, err)
+ }
+ if old, err := credentials.Load(g.paths.CredentialFile(), nil); err == nil && !force {
+ return fmt.Errorf("already paired as %s; use --force to pair again (the old credential stops working once the owner removes it)", old.DeviceID)
+ }
+ cfg, err := loadConfig(g)
+ if err != nil {
+ return err
+ }
+ _, statErr := os.Stat(g.paths.ConfigFile())
+ if server != "" {
+ cfg.Server = server
+ }
+ if kind != "" {
+ cfg.DeviceKind = kind
+ }
+ if err := cfg.Validate(); err != nil {
+ return err
+ }
+ descs, _ := probeAll(cfg, g.paths, g.log)
+ req := protocol.PairRequest{Code: code, AgentVersion: version, DeviceKind: cfg.DeviceKind, Drivers: []string{},
+ Capabilities: map[string]any{}, OS: runtime.GOOS, Arch: runtime.GOARCH}
+ req.Hostname, _ = os.Hostname()
+ names := make([]string, 0, len(descs))
+ for n := range descs {
+ names = append(names, n)
+ }
+ sort.Strings(names)
+ for _, n := range names {
+ req.Drivers = append(req.Drivers, descs[n].Driver)
+ req.Capabilities[n] = descs[n].Capabilities
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer cancel()
+ creds, err := link.Pair(ctx, nil, cfg.Server, req)
+ if err != nil {
+ return err
+ }
+ if err := credentials.Save(g.paths.CredentialFile(), creds); err != nil {
+ return err
+ }
+ chownLikeDir(g.paths.CredentialFile(), g.paths.ConfigDir)
+ if statErr != nil || server != "" || kind != "" {
+ if err := config.Save(g.paths.ConfigFile(), cfg); err != nil {
+ return err
+ }
+ chownLikeDir(g.paths.ConfigFile(), g.paths.ConfigDir)
+ }
+ fmt.Fprintf(stdout, "Paired as %s. Credential saved to %s (mode 600).\n", creds.DeviceID, g.paths.CredentialFile())
+ fmt.Fprintln(stdout, "Confirm the device on openvibe.bot, then start it: `openvibe-node run` or `sudo openvibe-node install`.")
+ if service.Status(serviceOptions(g, "")) == "running" {
+ _ = service.Control(serviceOptions(g, ""), "restart")
+ fmt.Fprintln(stdout, "The service was running and has been restarted with the new credential.")
+ }
+ return nil
+}
+
+func cmdRun(g *globals, dryRun bool) error {
+ if err := g.paths.Ensure(); err != nil {
+ return fmt.Errorf("cannot create %s (run with sudo, or use --home): %w", g.paths.StateDir, err)
+ }
+ cfg, err := loadConfig(g)
+ if err != nil {
+ return err
+ }
+ if dryRun {
+ var keep []config.PluginConfig
+ for _, p := range cfg.Plugins {
+ if p.Name == "dryrun" {
+ keep = append(keep, p)
+ }
+ }
+ if len(keep) == 0 {
+ keep = []config.PluginConfig{{Name: "dryrun"}}
+ }
+ cfg.Plugins = keep
+ cfg.Video.Source = "test"
+ }
+ creds, err := credentials.Load(g.paths.CredentialFile(), func(s string) { g.log.Warn(s) })
+ if err != nil && !errors.Is(err, credentials.ErrNotPaired) {
+ return err
+ }
+ n, err := node.New(node.Options{Config: cfg, Paths: g.paths, Creds: creds, Log: g.log, Version: version})
+ if err != nil {
+ return err
+ }
+ g.log.Info("openvibe-node starting", "version", version, "config", g.paths.ConfigDir, "plugins", len(cfg.Plugins))
+ return service.Run(serviceOptions(g, ""), n.Run)
+}
+
+func serviceOptions(g *globals, user string) service.Options {
+ args := []string{"run"}
+ if g.home != "" {
+ abs, _ := filepath.Abs(g.home)
+ args = append(args, "--home", abs)
+ }
+ if g.logLevel != "" && g.logLevel != "info" {
+ args = append(args, "--log-level", g.logLevel)
+ }
+ return service.Options{Arguments: args, UserName: user}
+}
+
+func cmdInstall(g *globals, user string, stdout io.Writer) error {
+ if err := g.paths.Ensure(); err != nil {
+ return fmt.Errorf("cannot create %s (run with sudo): %w", g.paths.ConfigDir, err)
+ }
+ if user != "" {
+ if err := chownTree(user, g.paths.ConfigDir, g.paths.StateDir); err != nil {
+ return err
+ }
+ }
+ opt := serviceOptions(g, user)
+ if err := service.Control(opt, "install"); err != nil {
+ return fmt.Errorf("install: %w (are you root?)", err)
+ }
+ if err := service.Control(opt, "start"); err != nil {
+ return fmt.Errorf("installed, but could not start: %w", err)
+ }
+ fmt.Fprintln(stdout, "Installed and started the openvibe-node service.")
+ if _, err := credentials.Load(g.paths.CredentialFile(), nil); err != nil {
+ fmt.Fprintln(stdout, "Not paired yet: run `sudo openvibe-node pair <CODE>`.")
+ }
+ return nil
+}
+
+func cmdStop(g *globals, stdout io.Writer) error {
+ _, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdStop}, 2*time.Second)
+ if err == nil {
+ fmt.Fprintln(stdout, "Stopped. Every actuator is held stopped until `openvibe-node resume`.")
+ return nil
+ }
+ // No running Node answered: latch it in the file, which a running Node polls and a starting Node obeys.
+ if ferr := safety.WriteLocalFile(g.paths.LatchFile(), true); ferr != nil {
+ return fmt.Errorf("could not reach the node (%v) nor write %s (%v); run with sudo", err, g.paths.LatchFile(), ferr)
+ }
+ fmt.Fprintln(stdout, "The node did not answer; the stop is latched in", g.paths.LatchFile())
+ fmt.Fprintln(stdout, "A running node obeys it within a quarter second; a starting node starts stopped. Release with `openvibe-node resume`.")
+ return nil
+}
+
+func cmdResume(g *globals, stdout io.Writer) error {
+ _, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdResume}, 2*time.Second)
+ if err != nil {
+ if ferr := safety.WriteLocalFile(g.paths.LatchFile(), false); ferr != nil {
+ return fmt.Errorf("could not reach the node (%v) nor write %s (%v); run with sudo", err, g.paths.LatchFile(), ferr)
+ }
+ }
+ fmt.Fprintln(stdout, "Resumed: the local stop and any server e-stop are released. Operators can drive again.")
+ return nil
+}
+
+func cmdStatus(g *globals, asJSON bool, stdout io.Writer) error {
+ resp, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdStatus}, 2*time.Second)
+ if err != nil {
+ return offlineStatus(g, asJSON, stdout)
+ }
+ if asJSON {
+ _, err := stdout.Write(append(resp.Data, '\n'))
+ return err
+ }
+ var s node.Status
+ if err := json.Unmarshal(resp.Data, &s); err != nil {
+ return err
+ }
+ fmt.Fprintf(stdout, "OpenVibe Node %s, running for %s (service: %s)\n", s.Version, s.Uptime, service.Status(serviceOptions(g, "")))
+ if s.Paired {
+ fmt.Fprintf(stdout, "Device: %s on %s\n", s.DeviceID, s.Server)
+ } else {
+ fmt.Fprintln(stdout, "Device: not paired (openvibe-node pair <CODE>)")
+ }
+ link := "down"
+ if s.Link.Connected {
+ link = fmt.Sprintf("up (rtt %d ms)", s.Link.RTTMS)
+ } else if s.Link.LastError != "" {
+ link = "down: " + s.Link.LastError
+ }
+ fmt.Fprintf(stdout, "Link: %s, %d reconnects\n", link, s.Link.Reconnects)
+ fmt.Fprintf(stdout, "Stop: %s\n", latchText(s.Latch))
+ fmt.Fprintf(stdout, "Limits: speed %.2f, turn %.2f, command deadline ≤ %d ms\n", s.Limits.MaxSpeed, s.Limits.MaxTurn, s.Limits.MaxCommandMS)
+ fmt.Fprintf(stdout, "Video: %s", s.Video.State)
+ if s.Video.Source != "" {
+ fmt.Fprintf(stdout, " from %s, %d frames", s.Video.Source, s.Video.Frames)
+ }
+ if s.Video.LastError != "" && s.Video.State != "live" {
+ fmt.Fprintf(stdout, " (%s)", s.Video.LastError)
+ }
+ fmt.Fprintln(stdout)
+ fmt.Fprintln(stdout, "Plugins:")
+ for _, p := range s.Plugins {
+ fmt.Fprintf(stdout, " %-15s %-9s %s %s", p.Name, p.State, p.Driver, p.Version)
+ if p.Restarts > 0 {
+ fmt.Fprintf(stdout, ", %d restarts", p.Restarts)
+ }
+ if p.Fault != nil {
+ fmt.Fprintf(stdout, ", fault %s: %s", p.Fault.Code, p.Fault.Message)
+ }
+ fmt.Fprintln(stdout)
+ }
+ if b := s.Telemetry.Battery; b != nil && b.Percent != nil {
+ fmt.Fprintf(stdout, "Battery: %.0f%%\n", *b.Percent)
+ }
+ return nil
+}
+
+func latchText(l safety.LatchState) string {
+ switch {
+ case l.Local && l.Remote:
+ return "STOPPED locally and by the server (" + l.RemoteReason + "); `openvibe-node resume` releases both"
+ case l.Local:
+ return "STOPPED locally; `openvibe-node resume` releases it"
+ case l.Remote:
+ return "E-STOPPED by the server (" + l.RemoteReason + "); the owner clears it on openvibe.bot"
+ }
+ return "clear"
+}
+
+func offlineStatus(g *globals, asJSON bool, stdout io.Writer) error {
+ type offline struct {
+ Running bool `json:"running"`
+ Service string `json:"service"`
+ Paired bool `json:"paired"`
+ DeviceID string `json:"device_id,omitempty"`
+ Server string `json:"server,omitempty"`
+ Latch safety.LatchState `json:"latch"`
+ Config string `json:"config_dir"`
+ }
+ o := offline{Service: service.Status(serviceOptions(g, "")), Config: g.paths.ConfigDir}
+ if c, err := credentials.Load(g.paths.CredentialFile(), nil); err == nil {
+ o.Paired, o.DeviceID, o.Server = true, c.DeviceID, c.Server
+ }
+ if l, _ := safety.OpenLatch(g.paths.LatchFile()); l != nil {
+ o.Latch = l.State()
+ }
+ if asJSON {
+ return json.NewEncoder(stdout).Encode(o)
+ }
+ fmt.Fprintf(stdout, "OpenVibe Node %s is not running (service: %s).\n", version, o.Service)
+ if o.Paired {
+ fmt.Fprintf(stdout, "Device: %s on %s\n", o.DeviceID, o.Server)
+ } else {
+ fmt.Fprintln(stdout, "Device: not paired (openvibe-node pair <CODE>)")
+ }
+ fmt.Fprintf(stdout, "Stop: %s\n", latchText(o.Latch))
+ fmt.Fprintf(stdout, "Config: %s\n", o.Config)
+ return nil
+}
+
+func cmdPlugins(g *globals, asJSON bool, stdout io.Writer) error {
+ type entry struct {
+ Name string `json:"name"`
+ State string `json:"state"`
+ Driver string `json:"driver,omitempty"`
+ Version string `json:"version,omitempty"`
+ Capabilities map[string]any `json:"capabilities,omitempty"`
+ }
+ var list []entry
+ if resp, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdPlugins}, 2*time.Second); err == nil {
+ var s node.Status
+ if err := json.Unmarshal(resp.Data, &s); err != nil {
+ return err
+ }
+ for _, p := range s.Plugins {
+ list = append(list, entry{p.Name, p.State, p.Driver, p.Version, p.Capabilities})
+ }
+ } else {
+ cfg, err := loadConfig(g)
+ if err != nil {
+ return err
+ }
+ descs, _ := probeAll(cfg, g.paths, g.log)
+ for _, s := range pluginSpecs(cfg, g.paths) {
+ e := entry{Name: s.Name, State: "not running"}
+ if d := descs[s.Name]; d != nil {
+ e.Driver, e.Version, e.Capabilities = d.Driver, d.Version, d.Capabilities
+ } else {
+ e.State = "not installed or broken (" + strings.Join(s.Argv, " ") + ")"
+ }
+ list = append(list, e)
+ }
+ }
+ if asJSON {
+ return json.NewEncoder(stdout).Encode(list)
+ }
+ for _, e := range list {
+ fmt.Fprintf(stdout, "%s [%s] %s %s\n", e.Name, e.State, e.Driver, e.Version)
+ keys := make([]string, 0, len(e.Capabilities))
+ for k := range e.Capabilities {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ for _, k := range keys {
+ b, _ := json.Marshal(e.Capabilities[k])
+ fmt.Fprintf(stdout, " %-9s %s\n", k, b)
+ }
+ }
+ if len(list) == 0 {
+ fmt.Fprintln(stdout, "No plugins configured in", g.paths.ConfigFile())
+ }
+ return nil
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,219 @@
@@ -0,0 +1,219 @@
+package main
+
+import (
+ "bytes"
+ "encoding/json"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "runtime"
+ "strings"
+ "syscall"
+ "testing"
+ "time"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"
+ "github.com/OpenVibers/OpenVibe.Node/internal/fakebot"
+ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
+ "github.com/OpenVibers/OpenVibe.Node/internal/safety"
+ "github.com/OpenVibers/OpenVibe.Node/internal/video"
+)
+
+// With OPENVIBE_NODE_MAIN=1 the test binary is openvibe-node itself.
+func TestMain(m *testing.M) {
+ if os.Getenv("OPENVIBE_NODE_MAIN") == "1" {
+ os.Exit(run(os.Args[1:], os.Stdout, os.Stderr))
+ }
+ os.Exit(m.Run())
+}
+
+func cli(t *testing.T, args ...string) (string, string, int) {
+ t.Helper()
+ var out, errb bytes.Buffer
+ code := run(args, &out, &errb)
+ return out.String(), errb.String(), code
+}
+
+func shortHome(t *testing.T) string {
+ d, err := os.MkdirTemp("", "ovn")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(d) })
+ return d
+}
+
+func TestHelpAndVersion(t *testing.T) {
+ if out, _, code := cli(t, "help"); code != 0 || !strings.Contains(out, "openvibe-node pair <CODE>") {
+ t.Fatalf("%d %s", code, out)
+ }
+ if out, _, code := cli(t, "version"); code != 0 || !strings.Contains(out, runtime.GOOS) {
+ t.Fatalf("%d %s", code, out)
+ }
+ if _, _, code := cli(t, "fly"); code != 2 {
+ t.Fatal("unknown command accepted")
+ }
+}
+
+func pythonPath(t *testing.T) string {
+ root, _ := filepath.Abs("../..")
+ return filepath.Join(root, "plugins/sdk") + string(os.PathListSeparator) + filepath.Join(root, "plugins/dryrun")
+}
+
+func TestPair(t *testing.T) {
+ srv := fakebot.New()
+ defer srv.Close()
+ srv.AddCode("ABCD2345")
+ home := shortHome(t)
+ t.Setenv("PYTHONPATH", pythonPath(t))
+ if _, errOut, code := cli(t, "pair", "nope", "--home", home); code == 0 || !strings.Contains(errOut, "8 letters") {
+ t.Fatalf("bad code accepted: %s", errOut)
+ }
+ out, errOut, code := cli(t, "pair", "abcd-2345", "--home", home, "--server", srv.URL())
+ if code != 0 {
+ t.Fatalf("pair failed: %s", errOut)
+ }
+ c, err := credentials.Load(filepath.Join(home, "credential.json"), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !strings.Contains(out, c.DeviceID) || strings.Contains(out+errOut, c.Credential.Reveal()) {
+ t.Fatalf("output: %s %s", out, errOut)
+ }
+ st, _ := os.Stat(filepath.Join(home, "credential.json"))
+ if st.Mode().Perm() != 0o600 {
+ t.Fatalf("mode %o", st.Mode().Perm())
+ }
+ if _, err := os.Stat(filepath.Join(home, "config.json")); err != nil {
+ t.Fatal("config not written")
+ }
+ reqs := srv.PairRequests()
+ if len(reqs) != 1 || reqs[0].Code != "ABCD2345" || reqs[0].AgentVersion != version {
+ t.Fatalf("%+v", reqs)
+ }
+ if _, err := exec.LookPath("python3"); err == nil && len(reqs[0].Drivers) != 1 {
+ t.Fatalf("dry-run plugin not probed: %+v", reqs[0])
+ }
+ if _, errOut, code := cli(t, "pair", "ABCD2345", "--home", home); code == 0 || !strings.Contains(errOut, "already paired") {
+ t.Fatalf("second pair: %s", errOut)
+ }
+ // status without a running node reads the files and never prints the secret.
+ out, _, code = cli(t, "status", "--home", home)
+ if code != 0 || !strings.Contains(out, "not running") || !strings.Contains(out, c.DeviceID) || strings.Contains(out, c.Credential.Reveal()) {
+ t.Fatalf("%d %s", code, out)
+ }
+}
+
+func TestStopWithoutRunningNodeLatchesFile(t *testing.T) {
+ home := shortHome(t)
+ if out, errOut, code := cli(t, "stop", "--home", home); code != 0 || !strings.Contains(out, "latched") {
+ t.Fatalf("%s %s", out, errOut)
+ }
+ l, _ := safety.OpenLatch(filepath.Join(home, "latch.json"))
+ if !l.State().Local {
+ t.Fatal("not latched")
+ }
+ cli(t, "resume", "--home", home)
+ l, _ = safety.OpenLatch(filepath.Join(home, "latch.json"))
+ if l.State().Stopped() {
+ t.Fatal("not resumed")
+ }
+}
+
+// TestRunDryRun is the acceptance path: `openvibe-node run --dry-run` pairs with the fake server, publishes the test
+// pattern, obeys commands, obeys the local kill switch and stops cleanly on SIGTERM.
+func TestRunDryRun(t *testing.T) {
+ if runtime.GOOS == "windows" {
+ t.Skip("unix only")
+ }
+ if _, err := exec.LookPath("python3"); err != nil {
+ t.Skip("python3 not installed")
+ }
+ srv := fakebot.New()
+ defer srv.Close()
+ srv.AddCode("WXYZ6789")
+ home := shortHome(t)
+ t.Setenv("PYTHONPATH", pythonPath(t))
+ if _, errOut, code := cli(t, "pair", "WXYZ-6789", "--home", home, "--server", srv.URL()); code != 0 {
+ t.Fatal(errOut)
+ }
+ c, _ := credentials.Load(filepath.Join(home, "credential.json"), nil)
+ whip := &video.WHIPReceiver{PublishKey: c.PublishKey.Reveal()}
+ srv.SetWHIP(whip)
+ defer whip.Close()
+
+ exe, _ := os.Executable()
+ cmd := exec.Command(exe, "run", "--dry-run", "--home", home)
+ cmd.Env = append(os.Environ(), "OPENVIBE_NODE_MAIN=1")
+ var logs bytes.Buffer
+ cmd.Stdout, cmd.Stderr = &logs, &logs
+ if err := cmd.Start(); err != nil {
+ t.Fatal(err)
+ }
+ exited := make(chan error, 1)
+ go func() { exited <- cmd.Wait() }()
+ defer func() {
+ if cmd.ProcessState == nil {
+ cmd.Process.Kill()
+ }
+ }()
+
+ conn, err := srv.NextConn(20 * time.Second)
+ if err != nil {
+ t.Fatalf("%v\n%s", err, logs.String())
+ }
+ if _, err := conn.Expect(protocol.TypeStatus, 20*time.Second, func(m protocol.Message) bool {
+ s := m.(protocol.Status)
+ return len(s.Drivers) == 1 && s.Drivers[0].State == "ready"
+ }); err != nil {
+ t.Fatalf("%v\n%s", err, logs.String())
+ }
+ send := func(id, kind, value string) protocol.Message {
+ t.Helper()
+ conn.Send(protocol.Command{ID: id, Kind: kind, Value: json.RawMessage(value), DeadlineMS: 300})
+ r, err := conn.Reply(id, 5*time.Second)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return r
+ }
+ if _, ok := send("c1", "drive", `{"throttle":0.4}`).(protocol.Ack); !ok {
+ t.Fatal("drive not acked")
+ }
+ out, errOut, code := cli(t, "status", "--home", home)
+ if code != 0 || !strings.Contains(out, "Link: up") || !strings.Contains(out, "dryrun") || strings.Contains(out, c.Credential.Reveal()) {
+ t.Fatalf("status: %s %s", out, errOut)
+ }
+ if out, _, code := cli(t, "stop", "--home", home); code != 0 || !strings.Contains(out, "held stopped") {
+ t.Fatalf("stop: %s", out)
+ }
+ if n, ok := send("c2", "drive", `{"throttle":0.4}`).(protocol.Nack); !ok || n.FaultCode != protocol.FaultLocalStop {
+ t.Fatal("drive accepted after the local stop")
+ }
+ cli(t, "resume", "--home", home)
+ if _, ok := send("c3", "drive", `{"throttle":0.4}`).(protocol.Ack); !ok {
+ t.Fatal("drive refused after resume")
+ }
+ deadline := time.Now().Add(20 * time.Second)
+ for whip.Packets.Load() < 10 {
+ if time.Now().After(deadline) {
+ t.Fatalf("no test pattern received\n%s", logs.String())
+ }
+ time.Sleep(50 * time.Millisecond)
+ }
+ cmd.Process.Signal(syscall.SIGTERM)
+ select {
+ case err := <-exited:
+ if err != nil {
+ t.Fatalf("exit: %v\n%s", err, logs.String())
+ }
+ case <-time.After(15 * time.Second):
+ t.Fatal("did not stop on SIGTERM")
+ }
+ if strings.Contains(logs.String(), c.Credential.Reveal()) || strings.Contains(logs.String(), c.PublishKey.Reveal()) {
+ t.Fatal("a secret reached the logs")
+ }
+ if !strings.Contains(logs.String(), "stopping every actuator") {
+ t.Fatalf("no shutdown stop in logs:\n%s", logs.String())
+ }
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,45 @@
@@ -0,0 +1,45 @@
+//go:build !windows
+
+package main
+
+import (
+ "fmt"
+ "os"
+ "os/user"
+ "path/filepath"
+ "strconv"
+ "syscall"
+)
+
+// chownLikeDir gives a file written by root (pair) to the account that owns its directory (the service account).
+func chownLikeDir(path, dir string) {
+ st, err := os.Stat(dir)
+ if err != nil {
+ return
+ }
+ if sys, ok := st.Sys().(*syscall.Stat_t); ok && os.Geteuid() == 0 {
+ _ = os.Chown(path, int(sys.Uid), int(sys.Gid))
+ }
+}
+
+// chownTree hands the config and state directories to the service account.
+func chownTree(name string, dirs ...string) error {
+ u, err := user.Lookup(name)
+ if err != nil {
+ return fmt.Errorf("no account %q (create it first, e.g. with install.sh): %w", name, err)
+ }
+ uid, _ := strconv.Atoi(u.Uid)
+ gid, _ := strconv.Atoi(u.Gid)
+ for _, d := range dirs {
+ err := filepath.Walk(d, func(p string, _ os.FileInfo, err error) error {
+ if err != nil {
+ return err
+ }
+ return os.Lchown(p, uid, gid)
+ })
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,11 @@
@@ -0,0 +1,11 @@
+//go:build windows
+
+package main
+
+import "errors"
+
+func chownLikeDir(path, dir string) {}
+
+func chownTree(name string, dirs ...string) error {
+ return errors.New("--user is not supported on Windows; the service runs as LocalSystem")
+}
@@ -4,6 +4,7 @@ go 1.24.7
@@ -4,6 +4,7 @@ go 1.24.7
require (
require (
github.com/gorilla/websocket v1.5.3
github.com/gorilla/websocket v1.5.3
+ github.com/kardianos/service v1.3.0
github.com/pion/ice/v4 v4.4.4
github.com/pion/ice/v4 v4.4.4
github.com/pion/interceptor v0.1.49
github.com/pion/interceptor v0.1.49
github.com/pion/rtcp v1.2.18
github.com/pion/rtcp v1.2.18
@@ -2,6 +2,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
@@ -2,6 +2,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/kardianos/service v1.3.0 h1:/LGy+xPP2TM+GLTiCZ2di7cy0Jd/qrawlTUfqKYFdTI=
+github.com/kardianos/service v1.3.0/go.mod h1:E4V9ufUuY82F7Ztlu1eN9VXWIQxg8NoLQlmFe0MtrXc=
github.com/pion/datachannel v1.6.3 h1:HZAYLpunVI0xi057ZiANl558d0uSpomcWw4UEPz3qXs=
github.com/pion/datachannel v1.6.3 h1:HZAYLpunVI0xi057ZiANl558d0uSpomcWw4UEPz3qXs=
github.com/pion/datachannel v1.6.3/go.mod h1:oRmkXVVXxkUbqa8MatgdNHW29kvtQY3HUdjJQsfv2GE=
github.com/pion/datachannel v1.6.3/go.mod h1:oRmkXVVXxkUbqa8MatgdNHW29kvtQY3HUdjJQsfv2GE=
github.com/pion/dtls/v3 v3.1.9 h1:rpeycmLIkc4krpk1IxP7+39o11QdCXbmV9+FGi9yZJ8=
github.com/pion/dtls/v3 v3.1.9 h1:rpeycmLIkc4krpk1IxP7+39o11QdCXbmV9+FGi9yZJ8=
new file mode 100644
new file mode 100644
@@ -0,0 +1,137 @@
@@ -0,0 +1,137 @@
+// Package service installs and runs the Node as a system service (systemd, launchd or the Windows service manager)
+// through kardianos/service. The same code path runs `openvibe-node run` in a terminal: Ctrl-C stops it cleanly.
+package service
+
+import (
+ "context"
+ "errors"
+ "sync"
+ "time"
+
+ ks "github.com/kardianos/service"
+)
+
+const (
+ Name = "openvibe-node"
+ DisplayName = "OpenVibe Node"
+ Description = "Connects this device to OpenVibe (robots, cameras, computer control) over one outbound link."
+)
+
+// Program is the body of the service: run until ctx ends.
+type Program func(ctx context.Context) error
+
+type program struct {
+ run Program
+ mu sync.Mutex
+ cancel context.CancelFunc
+ done chan struct{}
+ err error
+}
+
+func (p *program) Start(s ks.Service) error {
+ ctx, cancel := context.WithCancel(context.Background())
+ p.mu.Lock()
+ p.cancel, p.done = cancel, make(chan struct{})
+ p.mu.Unlock()
+ go func() {
+ err := p.run(ctx)
+ p.mu.Lock()
+ p.err = err
+ p.mu.Unlock()
+ close(p.done)
+ if err != nil && ctx.Err() == nil {
+ // The Node ended on its own (for example a second instance): let the service manager restart it.
+ _ = s.Stop()
+ }
+ }()
+ return nil
+}
+
+// Stop cancels the Node and waits for it to stop every actuator (at most 15 s).
+func (p *program) Stop(ks.Service) error {
+ p.mu.Lock()
+ cancel, done := p.cancel, p.done
+ p.mu.Unlock()
+ if cancel == nil {
+ return nil
+ }
+ cancel()
+ select {
+ case <-done:
+ case <-time.After(15 * time.Second):
+ return errors.New("timed out waiting for the node to stop")
+ }
+ return nil
+}
+
+// Options for the service definition.
+type Options struct {
+ // Arguments passed to the executable by the service manager (["run"] plus --home when set).
+ Arguments []string
+ // UserName runs the service as this account (Linux and macOS); empty = the service manager's default (root).
+ UserName string
+}
+
+func config(opt Options) *ks.Config {
+ return &ks.Config{
+ Name: Name, DisplayName: DisplayName, Description: Description, Arguments: opt.Arguments, UserName: opt.UserName,
+ Option: ks.KeyValue{
+ "Restart": "always", // systemd
+ "SuccessExitStatus": "0",
+ "KeepAlive": true, // launchd
+ "RunAtLoad": true,
+ "OnFailure": "restart", // Windows
+ "OnFailureDelayDuration": "5s",
+ "DelayedAutoStart": false,
+ },
+ }
+}
+
+// Run runs program under the service manager, or in the foreground when started from a terminal.
+func Run(opt Options, run Program) error {
+ p := &program{run: run}
+ s, err := ks.New(p, config(opt))
+ if err != nil {
+ return err
+ }
+ if err := s.Run(); err != nil {
+ return err
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.err
+}
+
+// Control performs install, uninstall, start, stop or restart on the service definition.
+func Control(opt Options, action string) error {
+ s, err := ks.New(&program{}, config(opt))
+ if err != nil {
+ return err
+ }
+ return ks.Control(s, action)
+}
+
+// Status reports running, stopped, not installed or unknown.
+func Status(opt Options) string {
+ s, err := ks.New(&program{}, config(opt))
+ if err != nil {
+ return "unknown"
+ }
+ st, err := s.Status()
+ if errors.Is(err, ks.ErrNotInstalled) {
+ return "not installed"
+ }
+ if err != nil {
+ return "unknown"
+ }
+ switch st {
+ case ks.StatusRunning:
+ return "running"
+ case ks.StatusStopped:
+ return "stopped"
+ }
+ return "unknown"
+}
+
+// Interactive reports whether the process runs in a terminal rather than under a service manager.
+func Interactive() bool { return ks.Interactive() }
new file mode 100644
new file mode 100644
@@ -0,0 +1,35 @@
@@ -0,0 +1,35 @@
+package service
+
+import (
+ "context"
+ "testing"
+ "time"
+)
+
+func TestProgramStartStop(t *testing.T) {
+ stopped := make(chan struct{})
+ p := &program{run: func(ctx context.Context) error {
+ <-ctx.Done()
+ time.Sleep(10 * time.Millisecond) // stopping actuators
+ close(stopped)
+ return nil
+ }}
+ if err := p.Start(nil); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.Stop(nil); err != nil {
+ t.Fatal(err)
+ }
+ select {
+ case <-stopped:
+ default:
+ t.Fatal("Stop returned before the program finished")
+ }
+}
+
+func TestConfig(t *testing.T) {
+ c := config(Options{Arguments: []string{"run"}})
+ if c.Name != "openvibe-node" || c.Arguments[0] != "run" || c.Option["Restart"] != "always" {
+ t.Fatalf("%+v", c)
+ }
+}