DeepStreamとAkka Streamsを使ったEdge AI 用サーバ構築

こんばんは！ Go や Python が勢いがある中、社内でも頑なに Scala を使っている @rkyymmt@github です。ダイバーシティ重要！

今回は、今年、Edge AI な PoC 案件で構築した、Edgeサーバで仕様した技術について少し説明したいと思います。

HW

ハードウェアは次のような構成です。

深度センサー（Intel RealSense）はJetson Nano経由で、他の比較的プリミティブなセンサーはM5CameraのGroveコネクタに繋げてWi-Fi経由で、取得したデータをEdge AIサーバの NVIDIA Jetson AGX Xavier に流し込む構成にしました。
深度センサーの深度情報は一次処理をJetson Nano上で行った後、検知情報のみXavierに送りました。

SW

ソフトウェアは次のような構成としました。
M5Cameraからの映像ストリーミングはマルチプラットフォームのストリーミングフレームワークであるGStreamerベースのSDKであるNVIDIA DeepStreamでパイプ的な加工・フィルタ・AI処理を行った後、Kafkaに流し込み、Alpakka経由でScalaから操作しました。
また、他のセンサー情報はJSONをAkka StreamsのHTTP実装である Akka HTTPで受け取る形を取りました。

例えば、上記のJSONを受け取るWebサーバ部分は、次のようなコードになります。

import java.io._
中略
import akka.event.Logging

final case class Value(
  sensor_data: Int
)

final case class Sensor(
  timestamp: String,
  uid: String,
  `type`: String,
  values: Array[Value]
)

trait JsonSupport extends SprayJsonSupport with DefaultJsonProtocol {
  implicit val valueFormat = jsonFormat1(Value)
  implicit val sensorFormat = jsonFormat4(Sensor)
}

trait Service extends Directives with JsonSupport {
  val route: Route = post {
    extractLog { implicit log =>
      entity(as[Sensor]) { sensor =>
        val timestamp = sensor.timestamp
        val uid = sensor.uid
        val `type` = sensor.`type`
        val values = sensor.values
中略
        complete(StatusCodes.OK)
      }
    }
  }
}

object Main extends Service {
  implicit val system: ActorSystem = ActorSystem()
  implicit val materializer: ActorMaterializer = ActorMaterializer()

  def main(args: Array[String]): Unit = {
    Http().bindAndHandle(route, "localhost", 8080)
  }
}

最後に

Goもよいけど、Scalaもよいぞ。