diff --git a/apps/control-station/src/core/fleet/useFleet.ts b/apps/control-station/src/core/fleet/useFleet.ts index 18cc2c2..32b0882 100644 --- a/apps/control-station/src/core/fleet/useFleet.ts +++ b/apps/control-station/src/core/fleet/useFleet.ts @@ -37,14 +37,25 @@ export function useFleet() { }, []); useEffect(() => { let active = true; - let timer: ReturnType; - async function poll() { + let fallback: ReturnType | undefined; + const read = async () => { try { const value = await fleetRequest<{ items: Vehicle[] }>(); if (active) { setItems(value.items); setError(""); } } catch { if (active) setError("Реестр недоступен. Показаны последние полученные сведения; связь сейчас не подтверждена."); } - finally { if (active) timer = setTimeout(poll, 5000); } - } - void poll(); - return () => { active = false; clearTimeout(timer); }; + }; + void read(); + const events = new EventSource("/api/v1/fleet/events"); + events.onmessage = event => { + if (!active) return; + try { const value = JSON.parse(event.data); setItems(value.items); setError(""); if (fallback) { clearInterval(fallback); fallback = undefined; } } + catch { unavailable(); } + }; + const unavailable = () => { + if (!active) return; + setError("Связь обновляется. Показаны последние полученные сведения."); + if (!fallback) fallback = setInterval(() => void read(), 5000); + }; + events.onerror = unavailable; + return () => { active = false; events.close(); if (fallback) clearInterval(fallback); }; }, []); return { items, error, refresh }; } diff --git a/apps/control-station/src/workspaces/fleet/VehicleSensors.tsx b/apps/control-station/src/workspaces/fleet/VehicleSensors.tsx index 5dbbd7f..6bcbdd7 100644 --- a/apps/control-station/src/workspaces/fleet/VehicleSensors.tsx +++ b/apps/control-station/src/workspaces/fleet/VehicleSensors.tsx @@ -4,7 +4,8 @@ import type {SensorInventory,SensorTransport} from '../../../../../packages/sens import {fleetRequest} from '../../core/fleet/useFleet'; export function VehicleSensors({vehicleID,enabled,onDetailChange}:{vehicleID:string;enabled:boolean;onDetailChange:(open:boolean)=>void}){ const transport=useMemo(()=>({ - inventory:async()=>{const fleet=await fleetRequest<{items:{id:string;sensor_state:SensorInventory}[]}>();const value=fleet.items.find(v=>v.id===vehicleID);if(!value)throw new Error('Аппарат не найден.');return value.sensor_state;}, + inventory:async()=>{const fleet=await fleetRequest<{items:{id:string;connectivity:string;sensor_state:SensorInventory}[]}>();const value=fleet.items.find(v=>v.id===vehicleID);if(!value)throw new Error('Аппарат не найден.');return {...value.sensor_state,fresh:value.connectivity==='online'};}, + subscribe:(receive,unavailable)=>{const events=new EventSource('/api/v1/fleet/events');events.onmessage=e=>{try{const value=JSON.parse(e.data).items.find((v:{id:string})=>v.id===vehicleID);if(value)receive({...value.sensor_state,fresh:value.connectivity==='online'});else unavailable();}catch{unavailable();}};events.onerror=unavailable;return()=>events.close();}, submit:value=>fleetRequest(`/${encodeURIComponent(vehicleID)}/devices/operations`,'POST',value), operation:id=>fleetRequest(`/${encodeURIComponent(vehicleID)}/devices/operations/${encodeURIComponent(id)}`), }),[vehicleID]); diff --git a/apps/node-agent/cmd/node-agent/main.go b/apps/node-agent/cmd/node-agent/main.go index 14baa38..68fa9f2 100644 --- a/apps/node-agent/cmd/node-agent/main.go +++ b/apps/node-agent/cmd/node-agent/main.go @@ -115,6 +115,7 @@ func run() error { go func() { errs <- private.Serve(unix) }() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + go app.Sensors.WatchUSB(ctx) go pairing.Run(ctx) log.Print("Mission Core Node " + version + " listening on loopback") select { diff --git a/apps/node-agent/internal/node/pairing_transport.go b/apps/node-agent/internal/node/pairing_transport.go index 4f0fffd..e9348ab 100644 --- a/apps/node-agent/internal/node/pairing_transport.go +++ b/apps/node-agent/internal/node/pairing_transport.go @@ -238,6 +238,12 @@ func (p *Pairing) send(ctx context.Context, b CoreBinding, path string, payload } func (p *Pairing) channel(ctx context.Context) { instance := "agent_" + token() + var changed <-chan struct{} + if p.Sensors != nil { + var unsubscribe func() + changed, unsubscribe = p.Sensors.events.subscribe() + defer unsubscribe() + } timer := time.NewTicker(5 * time.Second) defer timer.Stop() defer func() { @@ -335,6 +341,12 @@ func (p *Pairing) channel(ctx context.Context) { case <-ctx.Done(): return case <-timer.C: + case <-changed: + select { + case <-ctx.Done(): + return + case <-time.After(250 * time.Millisecond): + } } } } diff --git a/apps/node-agent/internal/node/sensor_events.go b/apps/node-agent/internal/node/sensor_events.go new file mode 100644 index 0000000..b524570 --- /dev/null +++ b/apps/node-agent/internal/node/sensor_events.go @@ -0,0 +1,137 @@ +package node + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "os/exec" + "strings" + "sync" + "time" +) + +// Events are hints to reconcile OS/SDK state, never commands or trusted inventory. +// A bounded latest-state signal prevents a slow viewer from blocking discovery. +type sensorEvents struct { + mu sync.Mutex + listeners map[chan struct{}]bool +} + +func (e *sensorEvents) subscribe() (<-chan struct{}, func()) { + e.mu.Lock() + defer e.mu.Unlock() + if e.listeners == nil { + e.listeners = map[chan struct{}]bool{} + } + c := make(chan struct{}, 1) + e.listeners[c] = true + return c, func() { e.mu.Lock(); delete(e.listeners, c); e.mu.Unlock() } +} +func (e *sensorEvents) notify() { + e.mu.Lock() + defer e.mu.Unlock() + for c := range e.listeners { + select { + case c <- struct{}{}: + default: + } + } +} + +func usbEvents(input io.Reader, changed func()) { + scanner := bufio.NewScanner(input) + fields := map[string]string{} + for scanner.Scan() { + line := scanner.Text() + if line == "" { + if fields["SUBSYSTEM"] == "usb" && fields["DEVTYPE"] == "usb_device" && + (fields["ACTION"] == "add" || fields["ACTION"] == "remove" || fields["ACTION"] == "change") { + changed() + } + fields = map[string]string{} + } else if key, value, ok := strings.Cut(line, "="); ok && len(fields) < 128 { + fields[key] = value + } + } +} + +func (s *Sensors) WatchUSB(ctx context.Context) { + for ctx.Err() == nil { + cmd := exec.CommandContext(ctx, "/usr/bin/udevadm", "monitor", "--udev", "--subsystem-match=usb", "--property") + cmd.Env = []string{"PATH=/usr/bin:/bin", "LANG=C"} + pipe, err := cmd.StdoutPipe() + if err == nil && cmd.Start() == nil { + usbEvents(pipe, s.events.notify) + _ = cmd.Wait() + } + // A failed monitor cannot disable the existing heartbeat reconciliation. + select { + case <-ctx.Done(): + return + case <-time.After(30 * time.Second): + } + } +} + +func (s *Sensors) stream(w http.ResponseWriter, r *http.Request, server *Server) { + if !server.authorized(w, r) { + return + } + if _, ok := w.(http.Flusher); !ok { + http.Error(w, "Streaming unavailable", 503) + return + } + changed, unsubscribe := s.events.subscribe() + defer unsubscribe() + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("X-Accel-Buffering", "no") + controller := http.NewResponseController(w) + timer := time.NewTimer(0) + defer timer.Stop() + for { + select { + case <-r.Context().Done(): + return + case <-changed: + // USB devices expose several interfaces; coalesce the enumeration burst. + select { + case <-r.Context().Done(): + return + case <-time.After(250 * time.Millisecond): + } + case <-timer.C: + } + if !server.authorized(w, r) { + return + } + value := s.Inventory() + data, err := json.Marshal(value) + if err != nil { + return + } + _ = controller.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if _, err = fmt.Fprintf(w, "retry: 3000\ndata: %s\n\n", data); err != nil { + return + } + if controller.Flush() != nil { + return + } + delay := 15 * time.Second + for _, raw := range value["operations"].([]any) { + if raw.(map[string]any)["state"] == "running" { + delay = 2 * time.Second + break + } + } + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(delay) + } +} diff --git a/apps/node-agent/internal/node/sensor_events_test.go b/apps/node-agent/internal/node/sensor_events_test.go new file mode 100644 index 0000000..ba2e9e5 --- /dev/null +++ b/apps/node-agent/internal/node/sensor_events_test.go @@ -0,0 +1,36 @@ +package node + +import ( + "strings" + "testing" +) + +func TestUSBEventsAreHintsAndDoNotMultiplyInterfaces(t *testing.T) { + input := "UDEV [1] add /devices/example (usb)\nACTION=add\nSUBSYSTEM=usb\nDEVTYPE=usb_device\n\n" + + "ACTION=add\nSUBSYSTEM=usb\nDEVTYPE=usb_interface\n\n" + + "ACTION=remove\nSUBSYSTEM=usb\nDEVTYPE=usb_device\n\n" + + "ACTION=add\nSUBSYSTEM=net\nDEVTYPE=usb_device\n\n" + e := sensorEvents{} + first, closeFirst := e.subscribe() + second, closeSecond := e.subscribe() + defer closeSecond() + count := 0 + usbEvents(strings.NewReader(input), func() { count++; e.notify() }) + if count != 2 || len(first) != 1 || len(second) != 1 { + t.Fatalf("events=%d first=%d second=%d", count, len(first), len(second)) + } + <-first + closeFirst() + e.notify() + if len(first) != 0 { + t.Fatal("closed viewer still receives events") + } +} + +func TestSensorEventStreamRequiresLocalSession(t *testing.T) { + s := newTestServer(t) + s.Sensors, _ = OpenSensors(t.TempDir(), "node_test") + if response := call(s, "GET", "/api/devices/events", "", nil); response.Code != 401 { + t.Fatal("unauthenticated device stream", response.Code) + } +} diff --git a/apps/node-agent/internal/node/sensors.go b/apps/node-agent/internal/node/sensors.go index 48c2f9b..1cf1100 100644 --- a/apps/node-agent/internal/node/sensors.go +++ b/apps/node-agent/internal/node/sensors.go @@ -45,14 +45,16 @@ type SensorOperation struct { Updated int64 `json:"updated_at"` } type Sensors struct { - mu sync.Mutex - prepareMu sync.Mutex - root string - nodeID string - instance string - client *http.Client - operations map[string]*SensorOperation - names map[string]string + events sensorEvents + mu sync.Mutex + prepareMu sync.Mutex + root string + nodeID string + instance string + client *http.Client + operations map[string]*SensorOperation + names map[string]string + initialized map[string]bool } var sensorID = regexp.MustCompile(`^rsd455_[0-9a-f]{32}$`) @@ -63,7 +65,7 @@ func OpenSensors(root, nodeID string) (*Sensors, error) { if e := os.MkdirAll(dir, 0700); e != nil { return nil, e } - s := &Sensors{root: dir, nodeID: nodeID, instance: "discovery_" + digest(token())[:24], operations: map[string]*SensorOperation{}, names: map[string]string{}} + s := &Sensors{root: dir, nodeID: nodeID, instance: "discovery_" + digest(token())[:24], operations: map[string]*SensorOperation{}, names: map[string]string{}, initialized: map[string]bool{}} s.client = &http.Client{Timeout: 25 * time.Second, Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { return (&net.Dialer{}).DialContext(ctx, "unix", "/run/mission-core-sensors/driver.sock") }}} @@ -85,6 +87,19 @@ func OpenSensors(root, nodeID string) (*Sensors, error) { } data, _ := os.ReadFile(filepath.Join(dir, "names.json")) _ = json.Unmarshal(data, &s.names) + data, _ = os.ReadFile(filepath.Join(dir, "initialized.json")) + _ = json.Unmarshal(data, &s.initialized) + if s.initialized == nil { + s.initialized = map[string]bool{} + } + for _, op := range s.operations { + if op.Command.Action == "prepare" && op.State == "complete" { + s.initialized[op.Command.Session.DeviceID] = true + } + } + if e := s.write("initialized.json", s.initialized); e != nil { + return nil, e + } return s, nil } func (s *Sensors) write(name string, value any) error { @@ -163,6 +178,7 @@ func (s *Sensors) Inventory() map[string]any { id, _ := item["id"].(string) seen[id] = true s.mu.Lock() + item["configured"] = s.initialized[id] if n := s.names[id]; n != "" { item["name"] = n } @@ -189,15 +205,21 @@ func (s *Sensors) Inventory() map[string]any { if seen[id] { continue } - now := time.Now().UTC().Format(time.RFC3339Nano) - name := "RealSense D455" - s.mu.Lock() - if n := s.names[id]; n != "" { - name = n - } - s.mu.Unlock() - items = append(items, map[string]any{"id": id, "name": name, "model": "RealSense D455", "prepared": false, "verified": false, "online": true, "usb": read("speed") + " Мбит/с", "layers": []any{}, "snapshot": map[string]any{"context": map[string]any{"session_id": s.instance + "_" + id, "device": map[string]any{"device_id": id, "model": map[string]string{"plugin_id": "missioncore.realsense", "plugin_version": "0.6.6", "model_id": "realsense.d455"}, "stability": "stable", "basis": "hardware-identifier"}, "execution": map[string]string{"node_id": s.nodeID, "agent_instance_id": s.instance, "platform": "linux"}, "opened_at": now}, "revision": 0, "enrollment": "empty", "connectivity": "connected", "acquisition": "idle", "observed_at": now}}) + seen[id] = true + items = append(items, s.discovery(id, read("speed")+" Мбит/с", true)) } + s.mu.Lock() + configured := []string{} + for id, ready := range s.initialized { + if ready && sensorID.MatchString(id) && !seen[id] { + configured = append(configured, id) + } + } + s.mu.Unlock() + for _, id := range configured { + items = append(items, s.discovery(id, "—", false)) + } + var preparation any if data, e := os.ReadFile("/var/lib/mission-core-node-drivers/preparation.json"); e == nil && len(data) < 32768 { _ = json.Unmarshal(data, &preparation) @@ -212,6 +234,26 @@ func (s *Sensors) Inventory() map[string]any { s.mu.Unlock() return map[string]any{"schema": "missioncore.node.devices/v1", "items": items, "preparation": preparation, "operations": operations} } +func (s *Sensors) discovery(id, speed string, online bool) map[string]any { + now := time.Now().UTC().Format(time.RFC3339Nano) + s.mu.Lock() + name, configured := s.names[id], s.initialized[id] + s.mu.Unlock() + if name == "" { + name = "RealSense D455" + } + connectivity, enrollment := "offline", "empty" + if online { + connectivity = "connected" + } + if configured { + enrollment = "enrolled" + } + return map[string]any{"id": id, "name": name, "model": "RealSense D455", "configured": configured, "prepared": false, "verified": false, "online": online, "usb": speed, "layers": []any{}, "snapshot": map[string]any{ + "context": map[string]any{"session_id": s.instance + "_" + id, "device": map[string]any{"device_id": id, "model": map[string]string{"plugin_id": "missioncore.realsense", "plugin_version": "0.6.6", "model_id": "realsense.d455"}, "stability": "stable", "basis": "hardware-identifier"}, "execution": map[string]string{"node_id": s.nodeID, "agent_instance_id": s.instance, "platform": "linux"}, "opened_at": now}, + "revision": 0, "enrollment": enrollment, "connectivity": connectivity, "acquisition": "idle", "observed_at": now}} +} + func sensorViewAction(action string) bool { return action == "details" || action == "offer" || action == "close-peer" } @@ -267,10 +309,12 @@ func (s *Sensors) Submit(c SensorCommand, remote bool) (*SensorOperation, error) } s.operations[c.ID] = value copy := *value + s.events.notify() go s.execute(c) return ©, nil } func (s *Sensors) execute(c SensorCommand) { + defer s.events.notify() var result any var err error uncertain := false @@ -313,6 +357,13 @@ func (s *Sensors) execute(c SensorCommand) { } s.mu.Lock() defer s.mu.Unlock() + if err == nil && c.Action == "prepare" { + s.initialized[c.Session.DeviceID] = true + if e := s.write("initialized.json", s.initialized); e != nil { + err = e + uncertain = true + } + } v := s.operations[c.ID] v.Updated = time.Now().Unix() if err != nil { @@ -394,6 +445,7 @@ func (s *Sensors) RemoteResults() []any { return out } func (s *Sensors) Routes(mux *http.ServeMux, server *Server) { + mux.HandleFunc("GET /api/devices/events", func(w http.ResponseWriter, r *http.Request) { s.stream(w, r, server) }) mux.HandleFunc("GET /api/devices", func(w http.ResponseWriter, r *http.Request) { if server.authorized(w, r) { reply(w, 200, s.Inventory()) diff --git a/apps/node-agent/internal/node/sensors_test.go b/apps/node-agent/internal/node/sensors_test.go index ffc7710..fc0ec05 100644 --- a/apps/node-agent/internal/node/sensors_test.go +++ b/apps/node-agent/internal/node/sensors_test.go @@ -84,3 +84,28 @@ func TestSensorPreservesUncertainDriverOutcome(t *testing.T) { }) } } + +func TestSensorConfiguredIdentitySurvivesRestartAndDisconnect(t *testing.T) { + root := t.TempDir() + s, _ := OpenSensors(root, "node_test") + c := sensorTestCommand() + s.initialized[c.Session.DeviceID] = true + if e := s.write("initialized.json", s.initialized); e != nil { + t.Fatal(e) + } + s, e := OpenSensors(root, "node_test") + if e != nil { + t.Fatal(e) + } + s.client = &http.Client{Transport: sensorRoundTrip(func(r *http.Request) (*http.Response, error) { + return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"items":[]}`)), Header: http.Header{}}, nil + })} + items := s.Inventory()["items"].([]any) + if len(items) != 1 { + t.Fatalf("configured camera disappeared: %d", len(items)) + } + item := items[0].(map[string]any) + if item["id"] != c.Session.DeviceID || item["online"] != false || item["configured"] != true { + t.Fatalf("incorrect offline identity: %+v", item) + } +} diff --git a/apps/node-agent/packaging/build.py b/apps/node-agent/packaging/build.py index dc29c65..f293dfd 100644 --- a/apps/node-agent/packaging/build.py +++ b/apps/node-agent/packaging/build.py @@ -11,7 +11,7 @@ import sys from build_deb import build, VERSION, BRAND_SHA256 ROOT = Path(__file__).resolve().parents[1] -DG_COMMIT = "17e150b1c74ab8a345fe34ce51dccd5bb862fa85" +DG_COMMIT = "be6463fd596ae2a1a96409d2cee84edc90e7167f" def guideline_sources(): diff --git a/apps/node-agent/packaging/build_deb.py b/apps/node-agent/packaging/build_deb.py index c3a9eb9..d5d4dbd 100644 --- a/apps/node-agent/packaging/build_deb.py +++ b/apps/node-agent/packaging/build_deb.py @@ -13,7 +13,7 @@ import tarfile ROOT = Path(__file__).resolve().parents[1] -VERSION = "0.6.6" +VERSION = "0.6.8" BRAND_SHA256 = "8bfee8ca9f98e0db48d98aae3af4b32493b8593e18b064a0239d513d824182af" diff --git a/apps/node-agent/sensors/device.py b/apps/node-agent/sensors/device.py index 8324520..de3d487 100644 --- a/apps/node-agent/sensors/device.py +++ b/apps/node-agent/sensors/device.py @@ -93,13 +93,30 @@ class Device: self.verified_this_process = False for path in self.root.glob("recordings/*/manifest.json"): value = json.loads(path.read_text()) - if value.get("state") == "recording": + if value.get("state") in ("recording", "finalizing"): value.update(state="interrupted", recovered_at=utc()) atomic(path, value) def save(self): atomic(self.root / "config.json", self.config) + def disconnect(self): + with self.lock: + if not self.online: + return + self.online = False + self.verified_this_process = False + self.sdk_device = None + # Stored playback is independent from a physical USB camera. + live = self.pipeline is not None and self.playback_id is None + if self.playback_id is None: + self.images = {} + self.motion = {} + self.depth = None + self.message = "Камера отключена от БК." + if live: + self.stop(failed=True) + def refresh(self, dev): with self.lock: self.sdk_device = dev @@ -110,7 +127,7 @@ class Device: self.online = True self.firmware = dev.get_info(rs.camera_info.firmware_version) self.transport = dev.get_info(rs.camera_info.usb_type_descriptor) - if self.pipeline is not None: + if self.pipeline is not None or self.acquisition == "stopping": return profiles, options = [], [] for index, sensor in enumerate(dev.query_sensors()): @@ -254,7 +271,7 @@ class Device: def start(self, selected=None, record=False): with self.lock: - if self.pipeline is not None: + if self.pipeline is not None or self.acquisition == "stopping": raise ValueError("Захват уже запущен. Сначала остановите его.") if not self.online or self.sdk_device is None: raise ValueError("Камера не подключена.") @@ -352,7 +369,7 @@ class Device: if not isinstance(ident, str) or not re.fullmatch(r"capture_[0-9a-f]{32}", ident): raise ValueError("Некорректная запись.") with self.lock: - if self.pipeline is not None: + if self.pipeline is not None or self.acquisition == "stopping": raise ValueError("Сначала остановите текущий захват или просмотр записи.") directory = self.root / "recordings" / ident source, manifest = directory / "source.db3", directory / "manifest.json" @@ -440,26 +457,39 @@ class Device: def stop(self, failed=False, from_capture=False): with self.lock: + if self.acquisition == "stopping": + raise ValueError("Исходная запись ещё сохраняется. Дождитесь завершения.") self.stop_event.set() pipeline, self.pipeline = self.pipeline, None if pipeline is not None: self.acquisition = "stopping" - pipeline.stop() + try: + pipeline.stop() + except RuntimeError: + # Removal may invalidate the SDK handle before STOP reaches it. + failed = True + self.message = "Захват прерван. Проверьте подключение камеры." del pipeline if self.thread and not from_capture: self.thread.join(timeout=3) self.playback_id = None - self.acquisition = "failed" if failed else "idle" + value = dict(self.record) if self.record else None + self.acquisition = "stopping" if value else "failed" if failed else "idle" self.revision += 1 - if self.record: - value = dict(self.record) + if value: value.update( - state="failed" if failed else "complete", + state="finalizing", ended_at=utc(), ended_monotonic_ns=time.monotonic_ns(), frames=dict(self.frames), ) + self.record = value path = self.root / "recordings" / value["id"] + atomic(path / "manifest.json", value) + if value: + # Hashing a large source must not lock inventory or heartbeat. The + # stopping state still prevents START, preparation and package updates. + try: source = path / "source.db3" if source.exists(): digest = hashlib.sha256() @@ -467,12 +497,24 @@ class Device: for chunk in iter(lambda: f.read(1024 * 1024), b""): digest.update(chunk) value.update(sha256=digest.hexdigest(), bytes=source.stat().st_size) + else: + failed = True + value["state"] = "failed" if failed else "complete" atomic(path / "manifest.json", value) - self.record = None - return {"ok": True} + except OSError: + failed = True + self.message = "Не удалось завершить сохранение исходной записи. Проверьте диск БК." + value["state"] = "failed" + atomic(path / "manifest.json", value) + finally: + with self.lock: + self.record = None + self.acquisition = "failed" if failed else "idle" + self.revision += 1 + return {"ok": True} def verify(self): - if self.pipeline is not None: + if self.pipeline is not None or self.acquisition == "stopping": raise ValueError("Остановите захват перед повторной проверкой.") try: self.start() diff --git a/apps/node-agent/sensors/server.py b/apps/node-agent/sensors/server.py index e704c47..0f57106 100644 --- a/apps/node-agent/sensors/server.py +++ b/apps/node-agent/sensors/server.py @@ -64,8 +64,7 @@ class Host: self.devices[ident].refresh(dev) for ident, device in self.devices.items(): if ident not in found: - device.online = False - device.verified_this_process = False + device.disconnect() await asyncio.to_thread(work) diff --git a/apps/node-agent/ui/src/NodeSensors.tsx b/apps/node-agent/ui/src/NodeSensors.tsx index 45667b4..26294e1 100644 --- a/apps/node-agent/ui/src/NodeSensors.tsx +++ b/apps/node-agent/ui/src/NodeSensors.tsx @@ -1,5 +1,5 @@ import {SensorWorkspace} from '../../../../packages/sensor-ui/src/SensorWorkspace'; import type {SensorTransport} from '../../../../packages/sensor-ui/src/contracts'; import {request} from './api'; -const transport:SensorTransport={inventory:()=>request('/api/devices'),submit:value=>request('/api/devices/operations','POST',value),operation:id=>request('/api/devices/operations/'+encodeURIComponent(id))}; +const transport:SensorTransport={subscribe:(receive,unavailable)=>{const events=new EventSource('/api/devices/events');events.onmessage=e=>{try{receive(JSON.parse(e.data));}catch{unavailable();}};events.onerror=unavailable;return()=>events.close();},inventory:()=>request('/api/devices'),submit:value=>request('/api/devices/operations','POST',value),operation:id=>request('/api/devices/operations/'+encodeURIComponent(id))}; export function NodeSensors(){return ;} diff --git a/apps/node-agent/ui/src/api.ts b/apps/node-agent/ui/src/api.ts index 9a05ac8..e567c20 100644 --- a/apps/node-agent/ui/src/api.ts +++ b/apps/node-agent/ui/src/api.ts @@ -28,7 +28,7 @@ export function desktopAction(action: DesktopAction): boolean { } export function desktopLogin(): boolean { return desktopAction("authorize"); } export async function request(path: string, method = "GET", body?: unknown): Promise { - const res = await fetch(path, { method, credentials: "same-origin", cache: "no-store", headers: body === undefined ? {} : {"Content-Type": "application/json"}, body: body === undefined ? undefined : JSON.stringify(body), signal: AbortSignal.timeout(10000) }); + const res = await fetch(path, { method, credentials: "same-origin", cache: "no-store", headers: body === undefined ? {} : {"Content-Type": "application/json"}, body: body === undefined ? undefined : JSON.stringify(body), signal: AbortSignal.timeout(10000) }).catch(() => { throw new Error("БК не ответил вовремя. Обновите сведения; действие могло продолжиться на борту."); }); if (!res.ok) { const error = await res.json().catch(() => ({})); throw new APIError(error.error ?? "Не удалось выполнить запрос к ноде", res.status); diff --git a/apps/node-agent/ui/vite.config.js b/apps/node-agent/ui/vite.config.js index aba17a6..6c16b4f 100644 --- a/apps/node-agent/ui/vite.config.js +++ b/apps/node-agent/ui/vite.config.js @@ -1,6 +1,8 @@ import { defineConfig } from "vite"; export default defineConfig({ + // Shared sensor TSX lives outside this app tsconfig; use the same JSX runtime. + esbuild: { jsx: "automatic" }, // Design Guideline packages are linked during development. Their own React // must never become a second hook dispatcher in the portable production bundle. resolve: { dedupe: ["react", "react-dom", "@nodedc/ui-react"] }, diff --git a/docs/node/05_SENSOR_HOST_AND_SHARED_CONTROL.md b/docs/node/05_SENSOR_HOST_AND_SHARED_CONTROL.md index a04b73b..e4ca7ff 100644 --- a/docs/node/05_SENSOR_HOST_AND_SHARED_CONTROL.md +++ b/docs/node/05_SENSOR_HOST_AND_SHARED_CONTROL.md @@ -110,5 +110,76 @@ UTC/monotonic timestamps, счётчики и SHA-256 завершённого нужно остановить. Первый тест исходной записи обнаружил отказ SDK от расширения .bag; это -исправлено в профиле 0.6.6 и требует повторной аппаратной проверки. SDK +исправлено в профиле 0.6.6 и аппаратно проверено в 0.6.7. SDK [документирует формат .db3](https://github.com/realsenseai/librealsense/blob/master/doc/record-and-playback.md). + +Обновление приёмки: запись .db3 и её RGB/depth-воспроизведение проверены в +общем локальном UI Node. В 0.6.7 исправлен автоматический JSX runtime для +общего модуля вне tsconfig приложения. 0.6.8 выносит подсчёт SHA-256 за mutex +инвентаризации: состояние остаётся stopping/finalizing и запрещает новый захват, +но системные сведения и heartbeat продолжают отвечать. Отдельный тест удерживает +хеширование и проверяет свободный inventory-lock и отказ нового START. + +Последняя визуальная корректировка владельца: после успешной подготовки +отображается только одна лампочка (StatusBadge indicator), без галочки и +дублирующего текста. Доступное название и подсказка сохраняют смысл состояния. + +Полная аппаратная приёмка в нативном GTK/WebKit-окне, работа через tailnet вне +LAN, смена USB-топологии/перезагрузка и чистая система не объявлены пройденными. +На момент этой записи применение нового replay action в работающем Core +ожидает отдельного разрешения на перезапуск: автоматическая проверка +интерпретировала просьбу «не прерывать процесс» как запрет останавливать сервер. + +Корректировка владельца 06.09.2026: кнопка первой подготовки удаляется из +строки после успешной инициализации. Повторное действие находится в настройках: +«Конфигурация на БК → Обновить» и вызывает тот же профиль, без произвольных +команд или сетевого скачивания. Результат первой подготовки сохраняется +независимо от текущей связи/сеанса драйвера. По следующему уточнению владельца отключённый сенсор исчезает из списка +подключённых устройств в обоих UI. Его приватная запись, имя и настройки сохраняются; +действие подготовки требует подключения. Перезапуск службы не возвращает +первичную кнопку. Проверка перехода restart/offline добавлена в Go-тест. + + +## Подключение и отключение USB: уточнение 06.09.2026 + +Оператору не нужен Refresh. Первичное перечисление даёт полный снимок Linux +sysfs и SDK. События udev add/remove/change для USB device запускают сверку: +событие — только сигнал, его поля никогда не являются командой или готовым +описанием устройства. Серия событий интерфейсов объединяется; один физический +D455 имеет одну аппаратную идентичность, USB-адрес/порт могут измениться. + +Node слушает фиксированный `udevadm monitor --udev --subsystem-match=usb +--property` без root, через уже разрешённый AF_NETLINK. Событие будит +существующий исходящий mTLS-канал и локальные подписки. Node и Core отправляют +снимки своим интерфейсам по SSE: одно соединение, новые состояния приходят от +сервера. Core ничего не опрашивает в USB на операторском компьютере. LAN или +Tailscale переносит управляющие сообщения; USB не пробрасывается. SDK драйвера +остаётся единственным владельцем камеры и потоков на БК. + +Heartbeat раз в 5 секунд сохранён для связи, доставки команд и восстановления +пропущенных событий, поэтому обнаружение не зависит исключительно от monitor. +Локальный SSE пересылает контрольный снимок раз в 15 секунд, при операции — раз +в 2 секунды. При потере SSE UI временно использует опрос раз в 3 секунды и +автоматически возвращается на push. Новое соединение всегда получает полный +снимок: потерянная последовательность событий не оставляет фантомные строки. +Нет обещания атомарного обновления двух экранов: локальный экран обновляется +первым, Core — после передачи того же состояния по действующему доверию. + +Подключение: строка появляется автоматически; имя и факт первой подготовки +берутся из сохранённой аппаратной идентичности. Новый сеанс не выдаётся за +новую установку. Лампочка готовности требует актуальных данных, а старый успех +проверки не выдаётся за действующий захват. Сама установка и подготовка не +запускаются просто от появления USB. + +Отключение: строка исчезает из списка подключённых. Открытый экран сохраняет +контекст с сообщением об отключении; старые кадры очищаются. Живой захват +останавливается, незавершённая исходная запись получает failed/interrupted; +она не удаляется и не возобновляется сама. Просмотр уже сохранённого файла +не зависит от USB. Повторное подключение открывает новый сеанс; прежние команды +START с предыдущим сеансом отвергаются. Вторая физическая D455 с другим serial +не наследует имя и подготовку первой. + +Потеря связи с БК не означает отключение USB. Core сохраняет последние сведения, +отмечает недоступность БК и запрещает аппаратные команды до свежего подтверждения. +Физические unplug/replug и смена USB-порта требуют отдельной аппаратной приёмки; +автоматическое пересоздание sandbox IIO при смене путей пока не подтверждено. diff --git a/packages/sensor-ui/src/SensorDetail.tsx b/packages/sensor-ui/src/SensorDetail.tsx index f05565d..d6f862e 100644 --- a/packages/sensor-ui/src/SensorDetail.tsx +++ b/packages/sensor-ui/src/SensorDetail.tsx @@ -2,27 +2,29 @@ import {useEffect,useState} from 'react'; import {ActivityIndicator,Button,Icon,IconButton,Select,SettingsCard,TextField} from '@nodedc/ui-react'; import {perform,type Sensor,type SensorTransport} from './contracts'; import {LiveViewport} from './LiveViewport'; -export function SensorDetail({device,transport,back,refresh,failure}:{device:Sensor;transport:SensorTransport;back:()=>void;refresh:()=>Promise;failure:(e:unknown)=>void}){ +export function SensorDetail({device,transport,back,refresh,failure,enabled=true}:{enabled?:boolean;device:Sensor;transport:SensorTransport;back:()=>void;refresh:()=>Promise;failure:(e:unknown)=>void}){ const [detail,setDetail]=useState(null);const [selected,setSelected]=useState>({});const [layer,setLayer]=useState('color');const [pending,setPending]=useState(false);const [option,setOption]=useState('');const [optionValue,setOptionValue]=useState(''); async function load(){const value=await perform(transport,device,'details');setDetail(value);return value;} useEffect(()=>{let live=true;if(device.prepared)void load().then(value=>{if(live){const selection:Record={};for(const p of value.profiles??[])if(value.defaults?.includes(p.id))selection[p.stream+':'+p.index]=p.id;setSelected(selection);}}).catch(failure);return()=>{live=false;};},[device.id,device.prepared,device.snapshot.context.session_id]); - async function act(action:string,parameters:Record={}){if(pending)return;failure(null);setPending(true);try{await perform(transport,device,action,parameters);await refresh();await load();}catch(e){failure(e);}finally{setPending(false);}} + async function act(action:string,parameters:Record={}){if(pending||!enabled)return;failure(null);setPending(true);try{await perform(transport,device,action,parameters);await refresh();await load();}catch(e){failure(e);}finally{setPending(false);}} const supported=(detail?.profiles??[]).filter(p=>['rgb8','bgr8','z16','y8','motion_xyz32f'].includes(p.format)&&(p.stream!=='infrared'||[1,2].includes(p.index))); const groups=[...new Set(supported.map(p=>p.stream+':'+p.index))]; const videoNames:Record={color:'RGB',depth:'Глубина',infrared1:'ИК · 1',infrared2:'ИК · 2',points:'Точки',motion:'Движение'}; const available=[...new Set(supported.map(p=>p.stream==='infrared'?p.stream+p.index:p.stream==='gyro'||p.stream==='accel'?'motion':p.stream))];if(available.includes('depth'))available.push('points'); - const currentOption=detail?.options?.find(v=>v.id===option);const active=device.snapshot.acquisition==='streaming'||device.snapshot.acquisition==='starting'; + const currentOption=detail?.options?.find(v=>v.id===option);const active=device.snapshot.acquisition==='streaming'||device.snapshot.acquisition==='starting'||device.snapshot.acquisition==='stopping'; return
+ {!device.online&&

Камера отключена от БК.

} {device.snapshot.message&&

{device.snapshot.message}

} -
{pending&&}
+
{pending&&}
{device.playback_id&&

Просмотр исходной записи · повтор. Камера для него не запускается.

} - {device.recording&&

Идёт исходная запись на БК. Она продолжится после закрытия окна.

} + {device.snapshot.acquisition==='stopping'&&

Сохраняем исходную запись и проверяем её целостность.

} + {device.recording&&device.snapshot.acquisition!=='stopping'&&

Идёт исходная запись на БК. Она продолжится после закрытия окна.

} {detail&&<>p.stream+':'+p.index===group&&['rgb8','bgr8','z16','y8','motion_xyz32f'].includes(p.format)).map(p=>({value:p.id,label:`${p.width?p.width+' × '+p.height+' · ':''}${p.fps} Гц · ${p.format}`}))]} onChange={value=>setSelected(v=>({...v,[group]:value}))} disabled={pending||active}/>)}
-
p.stream+':'+p.index===group&&['rgb8','bgr8','z16','y8','motion_xyz32f'].includes(p.format)).map(p=>({value:p.id,label:`${p.width?p.width+' × '+p.height+' · ':''}${p.fps} Гц · ${p.format}`}))]} onChange={value=>setSelected(v=>({...v,[group]:value}))} disabled={!enabled||pending||active}/>)}
 +