ユースケース：製造業 — 予知保全 IoT プラットフォームの構築

class SensorReading < ApplicationRecord
  belongs_to :device
  belongs_to :production_line
 
  # device_id:       デバイス識別子
  # sensor_type:     vibration / temperature / current / pressure
  # value:           センサー値（float）
  # recorded_at:     計測タイムスタンプ
  # anomaly_detected: 異常検知フラグ
  # anomaly_score:   異常スコア（0.0〜1.0）
 
  enum :sensor_type, {
    vibration: 0,
    temperature: 1,
    current: 2,
    pressure: 3
  }
 
  scope :anomalies, -> { where(anomaly_detected: true) }
  scope :for_device, ->(device_id) { where(device_id: device_id) }
  scope :recent, -> { where(recorded_at: 24.hours.ago..) }
  scope :by_type, ->(type) { where(sensor_type: type) }
 
  validates :value, presence: true, numericality: true
  validates :sensor_type, presence: true
  validates :recorded_at, presence: true
end

class Device < ApplicationRecord
  belongs_to :production_line
  has_many :sensor_readings, dependent: :destroy
  has_many :maintenance_predictions, dependent: :destroy
  has_many :alerts, dependent: :destroy
 
  # name:             設備名（例: "第1プレス機 PP-001"）
  # device_type:      設備種別（press / lathe / grinder / furnace）
  # installed_at:     設置日
  # last_maintained_at: 最終保全日
  # status:           稼働状態（running / warning / stopped / maintenance）
 
  enum :status, {
    running: 0,
    warning: 1,
    stopped: 2,
    maintenance: 3
  }
 
  def hours_since_maintenance
    return 0 unless last_maintained_at
    ((Time.current - last_maintained_at) / 1.hour).round(1)
  end
end

class AnomalyDetector
  THRESHOLDS = {
    vibration:   { warning: 2.0, critical: 3.5 },
    temperature: { warning: 2.5, critical: 4.0 },
    current:     { warning: 1.8, critical: 3.0 },
    pressure:    { warning: 2.0, critical: 3.5 }
  }.freeze
 
  WINDOW_SIZE = 720 # 直近12時間（1分間隔のデータ）
 
  def initialize(device)
    @device = device
    @baseline = BaselineCalculator.new(device)
  end
 
  def analyze(readings)
    readings.map do |reading|
      z_score = calculate_z_score(reading)
      severity = determine_severity(reading.sensor_type, z_score)
 
      {
        reading_id: reading.id,
        z_score: z_score.round(3),
        severity: severity,
        recommended_action: recommend_action(severity, reading.sensor_type),
        confidence: calculate_confidence(readings.size)
      }
    end
  end
 
  private
 
  def calculate_z_score(reading)
    baseline = @baseline.for_sensor(reading.sensor_type)
    return 0.0 if baseline.std_dev.zero?
 
    (reading.value - baseline.mean) / baseline.std_dev
  end
 
  def determine_severity(sensor_type, z_score)
    thresholds = THRESHOLDS[sensor_type.to_sym]
    abs_score = z_score.abs
 
    if abs_score >= thresholds[:critical]
      :critical
    elsif abs_score >= thresholds[:warning]
      :warning
    else
      :normal
    end
  end
 
  def recommend_action(severity, sensor_type)
    case severity
    when :critical
      "即座に設備停止・点検を実施してください（#{sensor_type}異常）"
    when :warning
      "次回の定期保全時に重点点検を推奨します（#{sensor_type}注意）"
    else
      "異常なし"
    end
  end
 
  def calculate_confidence(sample_size)
    # サンプル数が少ないと信頼度が下がる
    [sample_size.to_f / WINDOW_SIZE, 1.0].min.round(2)
  end
end

class MaintenancePrediction
  DEGRADATION_PATTERNS = {
    bearing:  { slope: -0.015, failure_threshold: 0.2 },
    belt:     { slope: -0.008, failure_threshold: 0.3 },
    motor:    { slope: -0.012, failure_threshold: 0.25 },
    hydraulic: { slope: -0.010, failure_threshold: 0.15 }
  }.freeze
 
  def predict_failure(device)
    recent_readings = device.sensor_readings
                            .where(recorded_at: 30.days.ago..)
                            .order(:recorded_at)
 
    return no_prediction if recent_readings.size < 100
 
    health_score = calculate_health_score(recent_readings)
    degradation_rate = calculate_degradation_rate(recent_readings)
    component = identify_degrading_component(recent_readings)
 
    pattern = DEGRADATION_PATTERNS[component]
    return no_prediction unless pattern
 
    remaining_life = estimate_remaining_life(
      health_score, degradation_rate, pattern
    )
 
    {
      device_id: device.id,
      estimated_failure_date: remaining_life[:date],
      confidence: remaining_life[:confidence],
      health_score: health_score,
      degrading_component: component,
      recommended_action: determine_action(remaining_life),
      estimated_cost_if_ignored: calculate_failure_cost(device)
    }
  end
 
  private
 
  def calculate_health_score(readings)
    # 複数センサーの加重平均で総合健全性を算出
    scores = readings.group_by(&:sensor_type).map do |type, type_readings|
      detector = AnomalyDetector.new(type_readings.first.device)
      results = detector.analyze(type_readings.last(10))
      normal_ratio = results.count { |r| r[:severity] == :normal }.to_f / results.size
      [type, normal_ratio]
    end.to_h
 
    weights = { "vibration" => 0.4, "temperature" => 0.25,
                "current" => 0.2, "pressure" => 0.15 }
 
    scores.sum { |type, score| score * weights.fetch(type, 0.25) }.round(3)
  end
 
  def determine_action(remaining_life)
    days = remaining_life[:days]
 
    if days <= 3
      "緊急: #{days}日以内に故障の可能性。即座に保全計画を立ててください"
    elsif days <= 14
      "計画保全: #{days}日以内の保全を推奨。部品の事前手配を開始してください"
    elsif days <= 30
      "経過観察: 約#{days}日の残存寿命。次回定期保全で重点点検してください"
    else
      "正常: 当面の保全不要。定期モニタリングを継続してください"
    end
  end
 
  def no_prediction
    { estimated_failure_date: nil, confidence: 0.0,
      recommended_action: "データ不足: 予測に必要なデータが蓄積されていません" }
  end
end

class AlertDispatcher
  ESCALATION_RULES = {
    normal:    [],
    warning:   [:dashboard, :slack],
    critical:  [:dashboard, :slack, :sms],
    emergency: [:dashboard, :slack, :sms, :line_stop]
  }.freeze
 
  def dispatch(alert)
    channels = ESCALATION_RULES[alert.severity.to_sym] || []
 
    channels.each do |channel|
      send("notify_#{channel}", alert)
    rescue StandardError => e
      Rails.logger.error("[AlertDispatcher] #{channel} 送信失敗: #{e.message}")
      fallback_notify(alert, channel)
    end
 
    schedule_escalation(alert) if alert.critical? || alert.emergency?
  end
 
  private
 
  def notify_dashboard(alert)
    ActionCable.server.broadcast(
      "factory_#{alert.device.production_line.factory_id}",
      { type: "alert", payload: alert.as_json }
    )
  end
 
  def notify_slack(alert)
    SlackNotifier.post(
      channel: "#factory-alerts",
      text: format_alert_message(alert),
      icon_emoji: severity_emoji(alert.severity)
    )
  end
 
  def notify_sms(alert)
    on_call = OnCallSchedule.current_engineer(alert.device.production_line)
    SmsGateway.send(
      to: on_call.phone_number,
      body: "【#{alert.severity.upcase}】#{alert.device.name}: #{alert.message}"
    )
  end
 
  def notify_line_stop(alert)
    # PLC にライン停止信号を送信（安全系統経由）
    PlcGateway.send_stop_signal(
      production_line: alert.device.production_line,
      reason: alert.message,
      initiated_by: "ai_predictive_maintenance"
    )
    Rails.logger.warn("[LINE_STOP] #{alert.device.production_line.name} 自動停止")
  end
 
  def schedule_escalation(alert)
    # 30分以内に対応がなければ上位者にエスカレーション
    AlertEscalationJob.set(wait: 30.minutes).perform_later(alert.id)
  end
end

class EdgeDataBuffer
  MAX_BUFFER_SIZE = 10_000
  SYNC_INTERVAL = 30.seconds
  CRITICAL_PRIORITY = 0
  NORMAL_PRIORITY = 10
 
  def initialize(device_id)
    @device_id = device_id
    @buffer = PriorityQueue.new
    @network_available = true
  end
 
  def push(reading)
    priority = reading[:anomaly_detected] ? CRITICAL_PRIORITY : NORMAL_PRIORITY
 
    if @buffer.size >= MAX_BUFFER_SIZE
      # バッファ満杯時は正常データを捨ててでも異常データを保持
      drop_lowest_priority if priority < @buffer.peek_lowest_priority
    end
 
    @buffer.push(reading, priority)
    attempt_sync if @network_available
  end
 
  def attempt_sync
    return if @buffer.empty?
 
    batch = @buffer.pop_batch(100)
    CloudSyncClient.send_batch(@device_id, batch)
    @network_available = true
  rescue NetworkError => e
    @network_available = false
    @buffer.push_all(batch) # 送信失敗したデータをバッファに戻す
    schedule_retry
    Rails.logger.warn("[EdgeBuffer] ネットワーク障害: #{e.message}")
  end
 
  def schedule_retry
    # 指数バックオフで再試行（最大5分間隔）
    RetryScheduler.exponential_backoff(
      initial: SYNC_INTERVAL,
      max: 5.minutes,
      callback: method(:attempt_sync)
    )
  end
end