import { HttpClient } from '@angular/common/http'; import { Injectable, OnDestroy } from '@angular/core'; import { map, Observable, Subject, take } from 'rxjs'; import { DpService } from 'dp-ui/dp.service'; import { FisAppMessage, MessageHeader, AppMessageType } from 'dp-ui/fisappmessage/apprequestmessagetype'; import { InferenceFrame, DetectionResult, MPOB_CLASSES, HEALTH_ALERT_CLASSES } from './inference.service'; interface PalmVisionConfig { connection: { uacp: string; uacp_ws: string; uacpEmulation: string; }; } export interface EdgeResultPayload { frame: string; filename?: string; batchId?: string; detections: DetectionResult[]; industrial_summary: Record; inference_ms: number; processing_ms?: number; } @Injectable({ providedIn: 'root' }) export class RemoteInferenceService implements OnDestroy { private config: PalmVisionConfig | null = null; private destroyed$ = new Subject(); constructor( private http: HttpClient, private dpService: DpService, ) { this.http.get('./config/config.json') .pipe(take(1)) .subscribe({ next: cfg => (this.config = cfg) }); } analyze(file: File, sourceLabel?: string, batchId?: string): Observable { return new Observable(observer => { const reader = new FileReader(); reader.onload = () => { const frame = reader.result as string; this.send('PalmVision', 'analyze', { frame, sourceLabel, batchId }) .subscribe({ next: raw => observer.next(this.mapAnalysisResponse(raw, batchId)), error: err => observer.error(err), complete: () => observer.complete(), }); }; reader.onerror = () => observer.error(new Error('FileReader failed to read image')); reader.readAsDataURL(file); }); } getHistory(): Observable { return this.send('History', 'getAll', undefined).pipe( map(body => { console.log('[Vault Transport Debug] Raw History Network Payload:', body); if (Array.isArray(body)) { return body; } if (body && typeof body === 'object') { const extractedRecords = body.records || body.data || body.items || body.history; if (Array.isArray(extractedRecords)) { return extractedRecords; } const values = Object.values(body); if (values.length > 0 && values.every(v => v !== null && typeof v === 'object')) { return values as any[]; } } return []; }) ); } deleteRecord(archiveId: string): Observable<{ deleted: boolean }> { return this.send('History', 'delete', { archiveId }); } clearHistory(): Observable<{ deleted: number }> { return this.send('History', 'clearAll', undefined); } getImage(archiveId: string): Observable<{ archiveId: string; image_data: string }> { return this.send('PalmHistory', 'GetImage', { archiveId }); } saveExternalResult(payload: EdgeResultPayload): Observable { const adjustedPayload: any = { ...payload, frame: (payload as any).frame || (payload as any).imageDataUrl || '', }; if (adjustedPayload.imageDataUrl) { delete adjustedPayload.imageDataUrl; } return this.send('PalmHistory', 'SaveExternalResult', adjustedPayload); } ngOnDestroy(): void { this.destroyed$.next(); this.destroyed$.complete(); } /** * Builds a compliant FIS envelope and dispatches it via the framework's official DpService stream. * Leverages core multiplexed transport pipelines rather than direct Socket.io interfaces. */ private send(serviceId: string, operation: string, payload: unknown): Observable { const messageID = crypto.randomUUID(); // Package parameters inside a fully compliant enterprise envelope structure const message: FisAppMessage = { header: { messageID, serviceId, messageName: operation, messageType: AppMessageType.Command } as unknown as MessageHeader, data: payload, }; return new Observable(observer => { // Direct call routing through the shared enterprise stream engine this.dpService.stream(message).subscribe({ next: (res: any) => { // Gracefully intercept and isolate system finalization frames before parsing if (res && res.complete === true) { observer.complete(); return; } // Extract body mapping parameters directly from enterprise results packets const body = typeof res === 'string' ? JSON.parse(res) : (res?.message ? JSON.parse(res.message) : res); if (body?.error) { observer.error(new Error(body.error)); } else { observer.next(body as T); } }, error: err => observer.error(err), complete: () => observer.complete(), }); }); } private mapAnalysisResponse(raw: any, batchId?: string): InferenceFrame { const detections: DetectionResult[] = (raw?.detections ?? []).map((d: any) => ({ bunch_id: d.bunch_id, class: d.class, confidence: d.confidence, is_health_alert: HEALTH_ALERT_CLASSES.includes(d.class), box: d.box, norm_box: d.norm_box, })); const industrial_summary: Record = raw?.industrial_summary ?? raw?.technical_evidence?.industrial_summary ?? {}; return { frameId: raw?.archive_id ?? crypto.randomUUID(), batchId, imageDataUrl: raw?.image_data ?? raw?.imageDataUrl ?? '', detections, inference_ms: raw?.inference_ms ?? 0, processing_ms: raw?.processing_ms ?? 0, total_count: detections.length, industrial_summary, source: 'remote', }; } }