協調型状況認識¶

// 協調型状況認識システム（RxJS + Angular）
import { Injectable } from '@angular/core';
import { BehaviorSubject, Observable, Subject, combineLatest, merge } from 'rxjs';
import { map, filter, distinctUntilChanged, debounceTime, share, scan } from 'rxjs/operators';
import { webSocket } from 'rxjs/webSocket';

// データ型定義
interface SituationUpdate {
  id: string;
  timestamp: Date;
  source: ParticipantInfo;
  type: 'perception' | 'analysis' | 'prediction' | 'recommendation';
  content: any;
  confidence: number;
  priority: 'low' | 'medium' | 'high' | 'critical';
}

interface ParticipantInfo {
  id: string;
  name: string;
  role: string;
  expertise: string[];
  location: string;
  status: 'active' | 'busy' | 'offline';
}

interface CollaborativeSAState {
  participants: ParticipantInfo[];
  sharedSituation: SharedSituationModel;
  activeDiscussions: Discussion[];
  consensusItems: ConsensusItem[];
  alerts: Alert[];
}

interface SharedSituationModel {
  currentSituation: SituationElement[];
  trends: TrendAnalysis[];
  predictions: PredictionItem[];
  riskAssessment: RiskAssessment;
  actionPriorities: ActionPriority[];
}

@Injectable({
  providedIn: 'root'
})
export class CollaborativeSAService {
  private wsConnection$ = webSocket<any>({
    url: 'ws://localhost:8080/collaborative-sa',
    openObserver: {
      next: () => console.log('協調SA接続確立')
    }
  });

  // 状態管理
  private participants$ = new BehaviorSubject<ParticipantInfo[]>([]);
  private situationUpdates$ = new Subject<SituationUpdate>();
  private consensusRequests$ = new Subject<ConsensusRequest>();
  private userActions$ = new Subject<UserAction>();

  // 統合状況認識ストリーム
  private collaborativeSAState$: Observable<CollaborativeSAState>;

  constructor() {
    this.initializeCollaborativeStreams();
  }

  private initializeCollaborativeStreams(): void {
    // WebSocket経由の更新を処理
    this.wsConnection$.subscribe(message => {
      this.handleIncomingMessage(message);
    });

    // 統合状況認識ストリームの構築
    this.collaborativeSAState$ = combineLatest([
      this.participants$,
      this.buildSharedSituationStream(),
      this.buildDiscussionStream(),
      this.buildConsensusStream(),
      this.buildAlertStream()
    ]).pipe(
      map(([participants, sharedSituation, discussions, consensus, alerts]) => ({
        participants,
        sharedSituation,
        activeDiscussions: discussions,
        consensusItems: consensus,
        alerts
      })),
      share()
    );
  }

  // 共有状況認識ストリームの構築
  private buildSharedSituationStream(): Observable<SharedSituationModel> {
    return this.situationUpdates$.pipe(
      // 優先度によるフィルタリング
      filter(update => update.priority !== 'low'),

      // 時間窓での集約（5秒間）
      debounceTime(5000),

      // 状況要素の統合
      scan((accumulated: SharedSituationModel, update: SituationUpdate) => {
        return this.integrateSituationUpdate(accumulated, update);
      }, this.getInitialSituationModel()),

      // 重複除去
      distinctUntilChanged((a, b) => this.compareSituationModels(a, b)),

      share()
    );
  }

  // リアルタイム参加者管理
  manageParticipants(): Observable<ParticipantInfo[]> {
    return merge(
      // 新規参加者
      this.wsConnection$.pipe(
        filter(msg => msg.type === 'participant_joined'),
        map(msg => msg.participant)
      ),

      // 参加者離脱
      this.wsConnection$.pipe(
        filter(msg => msg.type === 'participant_left'),
        map(msg => msg.participantId)
      ),

      // ステータス更新
      this.wsConnection$.pipe(
        filter(msg => msg.type === 'participant_status_changed'),
        map(msg => ({ id: msg.participantId, status: msg.newStatus }))
      )
    ).pipe(
      scan((participants: ParticipantInfo[], event: any) => {
        return this.updateParticipantList(participants, event);
      }, []),
      share()
    );
  }

  // 状況更新の投稿
  postSituationUpdate(update: Omit<SituationUpdate, 'id' | 'timestamp'>): void {
    const fullUpdate: SituationUpdate = {
      ...update,
      id: this.generateUpdateId(),
      timestamp: new Date()
    };

    // ローカルストリームに追加
    this.situationUpdates$.next(fullUpdate);

    // 他の参加者に送信
    this.wsConnection$.next({
      type: 'situation_update',
      update: fullUpdate
    });
  }

  // 合意形成プロセスの開始
  initiateConsensusProcess(topic: string, options: string[], participants: string[]): Observable<ConsensusResult> {
    const consensusId = this.generateConsensusId();

    const consensusRequest: ConsensusRequest = {
      id: consensusId,
      topic,
      options,
      participants,
      initiator: this.getCurrentUserId(),
      deadline: new Date(Date.now() + 30 * 60 * 1000), // 30分後
      votingMethod: 'majority'
    };

    // 合意プロセス開始を送信
    this.wsConnection$.next({
      type: 'consensus_request',
      request: consensusRequest
    });

    // 合意結果を監視
    return this.wsConnection$.pipe(
      filter(msg => msg.type === 'consensus_result' && msg.consensusId === consensusId),
      map(msg => msg.result),
      share()
    );
  }

  // インタラクティブディスカッション
  createDiscussionThread(topic: string, initialContent: string): Observable<Discussion> {
    const discussionId = this.generateDiscussionId();

    const discussion: Discussion = {
      id: discussionId,
      topic,
      participants: [],
      messages: [{
        id: this.generateMessageId(),
        author: this.getCurrentUserId(),
        content: initialContent,
        timestamp: new Date(),
        type: 'text'
      }],
      status: 'active',
      priority: 'medium'
    };

    this.wsConnection$.next({
      type: 'discussion_created',
      discussion
    });

    // ディスカッション更新を監視
    return this.wsConnection$.pipe(
      filter(msg => msg.type === 'discussion_update' && msg.discussionId === discussionId),
      map(msg => msg.discussion),
      share()
    );
  }

  // 状況認識品質メトリクス
  calculateCollaborationQuality(): Observable<CollaborationQualityMetrics> {
    return this.collaborativeSAState$.pipe(
      map(state => {
        const diversity = this.calculateInformationDiversity(state);
        const timeliness = this.calculateInformationTimeliness(state);
        const consensus = this.calculateConsensusLevel(state);
        const participation = this.calculateParticipationLevel(state);

        return {
          informationDiversity: diversity,
          informationTimeliness: timeliness,
          consensusLevel: consensus,
          participationLevel: participation,
          overallQuality: (diversity + timeliness + consensus + participation) / 4
        };
      }),
      share()
    );
  }

  // 状況認識の可視化データ生成
  generateCollaborationVisualization(): Observable<CollaborationVisualizationData> {
    return this.collaborativeSAState$.pipe(
      map(state => ({
        participantNetwork: this.buildParticipantNetworkData(state),
        informationFlow: this.buildInformationFlowData(state),
        consensusProgress: this.buildConsensusProgressData(state),
        situationCoverage: this.buildSituationCoverageData(state)
      })),
      share()
    );
  }

  // プライベートメソッド
  private integrateSituationUpdate(
    current: SharedSituationModel, 
    update: SituationUpdate
  ): SharedSituationModel {
    // 更新内容の統合ロジック
    const updatedSituation = { ...current };

    switch (update.type) {
      case 'perception':
        updatedSituation.currentSituation = this.mergePerceptionData(
          current.currentSituation, 
          update.content
        );
        break;
      case 'analysis':
        updatedSituation.trends = this.mergeTrendAnalysis(
          current.trends, 
          update.content
        );
        break;
      case 'prediction':
        updatedSituation.predictions = this.mergePredictions(
          current.predictions, 
          update.content
        );
        break;
      case 'recommendation':
        updatedSituation.actionPriorities = this.mergeRecommendations(
          current.actionPriorities, 
          update.content
        );
        break;
    }

    // リスク評価の再計算
    updatedSituation.riskAssessment = this.recalculateRiskAssessment(updatedSituation);

    return updatedSituation;
  }

  private calculateInformationDiversity(state: CollaborativeSAState): number {
    // 情報ソースの多様性を計算
    const sources = new Set(
      state.sharedSituation.currentSituation.map(element => element.source)
    );
    const participants = state.participants.length;

    return participants > 0 ? sources.size / participants : 0;
  }

  private calculateConsensusLevel(state: CollaborativeSAState): number {
    // 合意レベルを計算
    const completedConsensus = state.consensusItems.filter(item => item.status === 'completed');
    const avgAgreement = completedConsensus.reduce((sum, item) => sum + item.agreementLevel, 0);

    return completedConsensus.length > 0 ? avgAgreement / completedConsensus.length : 0;
  }

  // Getter for external access
  getCollaborativeSAState(): Observable<CollaborativeSAState> {
    return this.collaborativeSAState$;
  }
}

// Angular Component例
@Component({
  selector: 'app-collaborative-sa-dashboard',
  template: `
    <div class="collaborative-sa-dashboard">
      <!-- 参加者パネル -->
      <div class="participants-panel">
        <h3>参加者 ({{(participants$ | async)?.length}})</h3>
        <div class="participant-list">
          <div *ngFor="let participant of participants$ | async" 
               class="participant-item" 
               [class]="participant.status">
            <span class="name">{{participant.name}}</span>
            <span class="role">{{participant.role}}</span>
            <span class="status">{{participant.status}}</span>
          </div>
        </div>
      </div>

      <!-- 状況概要パネル -->
      <div class="situation-overview">
        <h3>統合状況認識</h3>
        <div class="situation-summary" *ngIf="sharedSituation$ | async as situation">
          <div class="current-situation">
            <h4>現在の状況</h4>
            <div *ngFor="let element of situation.currentSituation" class="situation-element">
              {{element.description}}
            </div>
          </div>

          <div class="risk-assessment">
            <h4>リスク評価</h4>
            <div class="risk-level" [class]="situation.riskAssessment.level">
              {{situation.riskAssessment.level}}: {{situation.riskAssessment.description}}
            </div>
          </div>
        </div>
      </div>

      <!-- アクティブディスカッション -->
      <div class="discussions-panel">
        <h3>アクティブディスカッション</h3>
        <div class="discussion-list">
          <div *ngFor="let discussion of activeDiscussions$ | async" class="discussion-item">
            <h4>{{discussion.topic}}</h4>
            <div class="message-count">{{discussion.messages.length}} メッセージ</div>
            <button (click)="joinDiscussion(discussion.id)">参加</button>
          </div>
        </div>
      </div>

      <!-- 合意プロセス -->
      <div class="consensus-panel">
        <h3>合意プロセス</h3>
        <div class="consensus-list">
          <div *ngFor="let consensus of consensusItems$ | async" class="consensus-item">
            <h4>{{consensus.topic}}</h4>
            <div class="progress-bar">
              <div class="progress" [style.width.%]="consensus.progress"></div>
            </div>
            <div class="voting-status">{{consensus.votedCount}}/{{consensus.totalParticipants}} 投票済み</div>
          </div>
        </div>
      </div>
    </div>
  `,
  styleUrls: ['./collaborative-sa-dashboard.component.scss']
})
export class CollaborativeSADashboardComponent implements OnInit {
  participants$: Observable<ParticipantInfo[]>;
  sharedSituation$: Observable<SharedSituationModel>;
  activeDiscussions$: Observable<Discussion[]>;
  consensusItems$: Observable<ConsensusItem[]>;

  constructor(private collaborativeSA: CollaborativeSAService) {}

  ngOnInit(): void {
    const state$ = this.collaborativeSA.getCollaborativeSAState();

    this.participants$ = state$.pipe(map(state => state.participants));
    this.sharedSituation$ = state$.pipe(map(state => state.sharedSituation));
    this.activeDiscussions$ = state$.pipe(map(state => state.activeDiscussions));
    this.consensusItems$ = state$.pipe(map(state => state.consensusItems));
  }

  joinDiscussion(discussionId: string): void {
    // ディスカッション参加ロジック
  }
}

// 結果インターフェース
interface CollaborationQualityMetrics {
  informationDiversity: number;
  informationTimeliness: number;
  consensusLevel: number;
  participationLevel: number;
  overallQuality: number;
}

interface ConsensusRequest {
  id: string;
  topic: string;
  options: string[];
  participants: string[];
  initiator: string;
  deadline: Date;
  votingMethod: 'majority' | 'unanimous' | 'weighted';
}

// 分散状況認識同期システム
class DistributedSASyncSystem {
  private peerConnections: Map<string, RTCPeerConnection> = new Map();
  private localSituationState$ = new BehaviorSubject<SituationState>(null);
  private remoteSituationStates$ = new Map<string, BehaviorSubject<SituationState>>();
  private conflictResolver: ConflictResolver;

  constructor() {
    this.conflictResolver = new ConflictResolver();
    this.setupPeerToPeerConnections();
  }

  // P2P接続の設定
  private setupPeerToPeerConnections(): void {
    // WebRTC設定
    const configuration: RTCConfiguration = {
      iceServers: [
        { urls: 'stun:stun.l.google.com:19302' },
        { urls: 'stun:stun1.l.google.com:19302' }
      ]
    };

    // シグナリングサーバーとの接続
    const signalingSocket = io('ws://signaling-server.example.com');

    signalingSocket.on('peer-joined', (peerId: string) => {
      this.initiatePeerConnection(peerId, configuration);
    });
  }

  // ピア接続の開始
  private async initiatePeerConnection(
    peerId: string, 
    configuration: RTCConfiguration
  ): Promise<void> {
    const peerConnection = new RTCPeerConnection(configuration);
    this.peerConnections.set(peerId, peerConnection);

    // データチャンネル設定
    const dataChannel = peerConnection.createDataChannel('situationSync', {
      ordered: true
    });

    dataChannel.onopen = () => {
      console.log(`Data channel opened with ${peerId}`);
      this.syncWithPeer(peerId);
    };

    dataChannel.onmessage = (event) => {
      this.handleRemoteSituationUpdate(peerId, JSON.parse(event.data));
    };

    // ICE候補とオファー/回答の処理
    this.setupICEHandling(peerConnection, peerId);
  }

  // 状況データの分散同期
  syncDistributedSituation(): Observable<ConsolidatedSituationState> {
    return combineLatest([
      this.localSituationState$,
      ...Array.from(this.remoteSituationStates$.values())
    ]).pipe(
      map(situationStates => this.consolidateSituationStates(situationStates)),
      distinctUntilChanged(),
      share()
    );
  }

  // 状況状態の統合
  private consolidateSituationStates(
    states: SituationState[]
  ): ConsolidatedSituationState {
    const validStates = states.filter(state => state !== null);

    if (validStates.length === 0) {
      return this.getEmptyConsolidatedState();
    }

    // 複数の状況認識を統合
    const consolidated = {
      perception: this.mergePerceptions(validStates),
      comprehension: this.mergeComprehensions(validStates),
      projection: this.mergeProjections(validStates),
      confidence: this.calculateConsolidatedConfidence(validStates),
      conflicts: this.detectConflicts(validStates),
      timestamp: new Date()
    };

    return consolidated;
  }

  // コンフリクト解決
  private detectConflicts(states: SituationState[]): ConflictInfo[] {
    const conflicts: ConflictInfo[] = [];

    // ペアワイズ比較でコンフリクトを検出
    for (let i = 0; i < states.length; i++) {
      for (let j = i + 1; j < states.length; j++) {
        const conflict = this.conflictResolver.detectConflict(states[i], states[j]);
        if (conflict) {
          conflicts.push(conflict);
        }
      }
    }

    return conflicts;
  }

  // 合意アルゴリズム（Byzantine Fault Tolerance風）
  reachDistributedConsensus(
    proposedState: SituationState
  ): Observable<ConsensusResult> {
    return new Observable<ConsensusResult>(observer => {
      const consensusRound = this.generateConsensusRoundId();
      const participants = Array.from(this.peerConnections.keys());

      // フェーズ1: Prepare
      const preparePromises = participants.map(peerId => 
        this.sendConsensusMessage(peerId, {
          type: 'prepare',
          round: consensusRound,
          proposedState: proposedState
        })
      );

      Promise.all(preparePromises).then(() => {
        // フェーズ2: Promise
        const promisePhase$ = this.collectPromiseResponses(consensusRound);

        promisePhase$.subscribe(responses => {
          if (this.hasQuorum(responses)) {
            // フェーズ3: Commit
            this.broadcastCommit(consensusRound, proposedState);
            observer.next({
              success: true,
              consensusState: proposedState,
              participantCount: responses.length
            });
          } else {
            observer.next({
              success: false,
              reason: 'No quorum reached',
              participantCount: responses.length
            });
          }
          observer.complete();
        });
      });
    });
  }
}

// コンフリクト解決クラス
class ConflictResolver {
  detectConflict(stateA: SituationState, stateB: SituationState): ConflictInfo | null {
    // 時間ベースの一致確認
    const timeDiff = Math.abs(stateA.timestamp.getTime() - stateB.timestamp.getTime());
    if (timeDiff > 60000) { // 1分以上の差
      return null; // 時間が異なるため比較対象外
    }

    // 認識内容の相違確認
    const perceptionConflict = this.comparePerceptions(stateA.perception, stateB.perception);
    const comprehensionConflict = this.compareComprehensions(stateA.comprehension, stateB.comprehension);
    const projectionConflict = this.compareProjections(stateA.projection, stateB.projection);

    if (perceptionConflict || comprehensionConflict || projectionConflict) {
      return {
        type: 'content_mismatch',
        stateA: stateA.id,
        stateB: stateB.id,
        conflictAreas: [
          ...(perceptionConflict ? ['perception'] : []),
          ...(comprehensionConflict ? ['comprehension'] : []),
          ...(projectionConflict ? ['projection'] : [])
        ],
        severity: this.calculateConflictSeverity(stateA, stateB)
      };
    }

    return null;
  }

  resolveConflict(conflict: ConflictInfo, availableStates: SituationState[]): SituationState {
    switch (conflict.type) {
      case 'content_mismatch':
        return this.resolveContentMismatch(conflict, availableStates);
      case 'temporal_inconsistency':
        return this.resolveTemporalInconsistency(conflict, availableStates);
      default:
        return this.defaultResolution(availableStates);
    }
  }

  private resolveContentMismatch(
    conflict: ConflictInfo, 
    states: SituationState[]
  ): SituationState {
    // 信頼度ベースの重み付け統合
    const weightedStates = states.map(state => ({
      state,
      weight: state.confidence * this.getSourceCredibility(state.source)
    }));

    const totalWeight = weightedStates.reduce((sum, ws) => sum + ws.weight, 0);

    // 重み付き平均による統合
    return this.weightedMerge(weightedStates, totalWeight);
  }
}