UI-Prototype/src/src.palm.vision/services/remote-inference.service.ts

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<string, number>;
  inference_ms: number;
  processing_ms?: number;
}

@Injectable({ providedIn: 'root' })
export class RemoteInferenceService implements OnDestroy {
  private config: PalmVisionConfig | null = null;
  private destroyed$ = new Subject<void>();

  constructor(
    private http: HttpClient,
    private dpService: DpService,
  ) {
    this.http.get<PalmVisionConfig>('./config/config.json')
      .pipe(take(1))
      .subscribe({ next: cfg => (this.config = cfg) });
  }

  analyze(file: File, sourceLabel?: string, batchId?: string): Observable<InferenceFrame> {
    return new Observable<InferenceFrame>(observer => {
      const reader = new FileReader();
      reader.onload = () => {
        const frame = reader.result as string;
        this.send<any>('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<any[]> {
    return this.send<any>('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<any> {
    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<T>(serviceId: string, operation: string, payload: unknown): Observable<T> {
    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<T>(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<string, number> = 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',
    };
  }
}