AWS IoT Core×Unityでオフィストイレのデジタルツインを構築してみた

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Publish",
      "Resource": "arn:aws:iot:ap-northeast-1:XXXXXXXXXXXX:topic/$aws/rules/openclose_timestream/*"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:ap-northeast-1:XXXXXXXXXXXX:client/Room1"
    }
  ]
}

import json
import sys

from awscrt import http, mqtt, exceptions
from awsiot import mqtt_connection_builder

host = "xxxxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com" # AWSアカウントのデバイスデータエンドポイント
rootCAPath = "AmazonRootCA1.pem" # モノ作成時にダウンロードしたルートCA証明書
certificatePath = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx-certificate.pem.crt" # モノ作成時にダウンロードしたデバイス証明書
privateKeyPath = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx-private.pem.key" # モノ作成時にダウンロードしたプライベートキーファイル
port = 8883
clientId = "Room1" # IoTポリシーでConnect権限を付与したクライアントID
topic = "$aws/rules/openclose_timestream/room1/openclose" # IoTルールのトピック

(中略)

def Main(PAL=None):
    if isinstance(PAL, AppPAL):
        sns_data = PAL.GetDataDict()

        arrive_time = sns_data['ArriveTime'].strftime('%Y/%m/%d %H:%M:%S') + '.%06d'%(sns_data['ArriveTime'].microsecond)

        try:
            mqtt_connection = mqtt_connection_builder.mtls_from_path(
                endpoint=host,
                port=port,
                cert_filepath=certificatePath,
                pri_key_filepath=privateKeyPath,
                ca_filepath=rootCAPath,
                client_id=clientId,
                clean_session=False,
                keep_alive_secs=30)

            connect_future = mqtt_connection.connect()
            connect_future.result()

            is_closed = True
            if sns_data["HALLIC"] in [0,128]:
                is_closed = False

            message = {}
            message['LogicalID'] = sns_data['LogicalID']
            message['EndDeviceSID'] = sns_data['EndDeviceSID']
            message['IsClosed'] = is_closed
            message['Power(mV)'] = sns_data['Power']
            message['LQI'] = sns_data['LQI']
            message['ArriveTime'] = arrive_time
            messageJson = json.dumps(message)
            mqtt_connection.publish(
                topic=topic,
                payload=messageJson,
                qos=mqtt.QoS.AT_LEAST_ONCE)
            print('Published topic {}: {}'.format(topic, messageJson))

            disconnect_future = mqtt_connection.disconnect()
            disconnect_future.result()
        except exceptions.AwsCrtError as e:
            print(e)

import { DynamoDBClient, DeleteItemCommand, QueryCommand } from "@aws-sdk/client-dynamodb";
import { ApiGatewayManagementApiClient, PostToConnectionCommand } from "@aws-sdk/client-apigatewaymanagementapi";

const dynamoDBClient = new DynamoDBClient({ region: process.env.AWS_REGION });

export const handler = async event => {
  let connectionData;
  
  const queryInput = {
    TableName: process.env.TABLE_NAME,
    IndexName: process.env.GSI_NAME,
    KeyConditionExpression: 'roomId = :roomId',
    ExpressionAttributeValues: {
      ':roomId': { S: '' + event.RoomID }
    }
  };

  try {
    const queryCommand = new QueryCommand(queryInput);
    connectionData = await dynamoDBClient.send(queryCommand);
  } catch (err) {
    console.log(err);
    return { statusCode: 500, body: err.stack };
  }

  const apiGatewayManagementApiClient = new ApiGatewayManagementApiClient({
    region: process.env.AWS_REGION,
    endpoint: 'https://' + process.env.API_DOMAIN + '/' + process.env.API_STAGE
  });

  const postData = JSON.stringify(event);

  const postCalls = connectionData.Items.map(async ({ connectionId }) => {
    try {
      const postToConnectionCommandInput = {
        Data: postData,
        ConnectionId: connectionId.S,
      };
      const postToConnectionCommandCommand = new PostToConnectionCommand(postToConnectionCommandInput);
      await apiGatewayManagementApiClient.send(postToConnectionCommandCommand);
    } catch (err) {
      if (err.$metadata.httpStatusCode === 410) {
        console.log(`Found stale connection, deleting ${connectionId.S}`);
        const deleteItemInput = {
          TableName: process.env.TABLE_NAME,
          Key: {
            'connectionId': connectionId
          }
        };
        const deleteItemCommand = new DeleteItemCommand(deleteItemInput);
        await dynamoDBClient.send(deleteItemCommand);
      } else {
        throw err;
      }
    }
  });
  
  try {
    await Promise.all(postCalls);
  } catch (err) {
    console.log(err);
    return { statusCode: 500, body: err.stack };
  }

  return { statusCode: 200, body: 'Data sent.' };
};

import { DynamoDBClient, DeleteItemCommand } from "@aws-sdk/client-dynamodb";

const client = new DynamoDBClient({ region: process.env.AWS_REGION });

export const handler = async event => {
  const input = {
    TableName: process.env.TABLE_NAME,
    Key: {
      'connectionId': { S: event.requestContext.connectionId }
    }
  };

  try {
    const command = new DeleteItemCommand(input);
    await client.send(command);
  } catch (err) {
    console.log(err);
    return { statusCode: 500, body: 'Failed to disconnect: ' + JSON.stringify(err) };
  }

  return { statusCode: 200, body: 'Disconnected.' };
};

import { DynamoDBClient, PutItemCommand } from "@aws-sdk/client-dynamodb";
import { LambdaClient, InvokeCommand } from "@aws-sdk/client-lambda";

const client = new DynamoDBClient({ region: process.env.AWS_REGION });

export const handler = async event => {
  const input = {
    TableName: process.env.TABLE_NAME,
    Item: {
      'connectionId': { S: event.requestContext.connectionId },
      'roomId': { S: event.queryStringParameters.roomId }
    }
  };

  try {
    const command = new PutItemCommand(input);
    await client.send(command);
  } catch (err) {
    console.log(err);
    return { statusCode: 500, body: 'Failed to connect: ' + JSON.stringify(err) };
  }

  try {
    let payload = {};
    payload['connectionId'] = event.requestContext.connectionId;
    payload['roomId'] = event.queryStringParameters.roomId;
    
    const lambdaClient = new LambdaClient({ region: process.env.AWS_REGION });
    const invokeCommandInput = {
      FunctionName: process.env.PUSH_INITIAL_DATA_LAMBDA,
      InvocationType: "Event",
      Payload: JSON.stringify(payload),
    };
    const invokeCommand = new InvokeCommand(invokeCommandInput);
    const response = await lambdaClient.send(invokeCommand);
  } catch (err) {
    console.log(err);
    return { statusCode: 500, body: 'Failed to connect: ' + JSON.stringify(err) };
  }

  return { statusCode: 200, body: 'Connected.' };
};

import { TimestreamQueryClient, QueryCommand } from "@aws-sdk/client-timestream-query";
import { ApiGatewayManagementApiClient, PostToConnectionCommand } from "@aws-sdk/client-apigatewaymanagementapi";

export const handler = async event => {
  const connectionId = event.connectionId;
  const roomId = event.roomId;

  const timestreamQueryClient = new TimestreamQueryClient({ region: process.env.AWS_REGION });
  try {
    const input = {
      QueryString: `
        WITH latest_ts AS(
          SELECT
            RoomID,
            LogicalID,
            MAX(time) AS latest_timestamp
          FROM
            iotDatabase.openclose
          WHERE
            RoomID = '` + roomId + `'
          GROUP BY
            RoomID,
            LogicalID
        )
        SELECT
          oc.RoomID AS roomId,
          oc.LogicalID,
          ARBITRARY(
            CASE
              WHEN oc.measure_name = 'ArriveTime' THEN oc.measure_value::varchar
              ELSE NULL
            END
          ) AS ArriveTime,
          ARBITRARY(
            CASE
              WHEN oc.measure_name = 'IsClosed' THEN oc.measure_value::boolean
              ELSE NULL
            END
          ) AS IsClosed
        FROM
          iotDatabase.openclose oc
          JOIN
            latest_ts AS lts
          ON  lts.RoomID = oc.RoomID
          AND lts.LogicalID = oc.LogicalID
          AND lts.latest_timestamp = oc.time
        WHERE
          oc.measure_name IN('ArriveTime', 'IsClosed')
        GROUP BY
          oc.RoomID,
          oc.LogicalID
        ORDER BY
          oc.RoomID,
          oc.LogicalID
      `
    };
    const queryCommand = new QueryCommand(input);
    const response = await timestreamQueryClient.send(queryCommand);

    const columns = response.ColumnInfo.map(c => c.Name);

    const apiGatewayManagementApiClient = new ApiGatewayManagementApiClient({
      region: process.env.AWS_REGION,
      endpoint: 'https://' + process.env.API_DOMAIN + '/' + process.env.API_STAGE
    });
    
    const postCalls = response.Rows.map(async r => {
      let row = {}
      r.Data.forEach((d, i) => {
        try {
          row[columns[i]] = JSON.parse(d.ScalarValue);
        } catch (err) {
          row[columns[i]] = d.ScalarValue;
        }
      });
      
      const postData = JSON.stringify(row);

      const postToConnectionCommandInput = {
        Data: postData,
        ConnectionId: connectionId,
      };
      const postToConnectionCommandCommand = new PostToConnectionCommand(postToConnectionCommandInput);
      await apiGatewayManagementApiClient.send(postToConnectionCommandCommand);

    });
    
    await Promise.all(postCalls);

  } catch (err) {
    console.log(err);
    return { statusCode: 500, body: 'Failed to send initial data: ' + JSON.stringify(err) };
  }
  
  return { statusCode: 200, body: 'Initial data sent.' };
};

using UnityEngine;
using UnityEngine.UI;
using System;
using System.Collections;
using System.Collections.Generic;
using WebSocketSharp;

public class WebSocketController : MonoBehaviour
{
    public GameObject door;
    private WebSocket ws;
    private Queue <string> messageQueue = new Queue <string>();
    private bool isProcessing = false;

    private string receivedMessage;
    public DateTime sensorTime;

    [System.Serializable]
    public class ResponseData
    {
        public string LogicalID;
        public string ArriveTime;
        public bool IsClosed;
    }

    private static WebSocketController instance;
    public static WebSocketController Instance { get { return instance; } }

    private void Awake()
    {
        if (instance != null &&instance != this)
        {
            Destroy(this.gameObject);
        }
        else
        {
            instance = this;
        }
    }
    
    public void StartGame()
    {
        // a. WebSocketのコネクションの確立
        ws = new WebSocket("wss://xxxxxxxxxxxxxxxxx.ap-northeast-1.amazonaws.com/prd?roomId=1");
        ws.OnOpen += (sender, e) =>
        {
            Debug.Log("WebSocket Open");
        };
        ws.OnMessage += OnWebSocketMessage;
        ws.Connect(); 
		
		// 受信データを順番に捌く処理を起動
        StartCoroutine(ProcessMessageQueue());
    }

	// 受信データを順番に捌く関数
    private IEnumerator ProcessMessageQueue()
    {
        while (true)
        {
            if (!isProcessing &&messageQueue.Count > 0)
            {
                isProcessing = true;

                string message = messageQueue.Dequeue();
				// 受信メッセージからドアの状態を取得する処理を起動
                StartCoroutine(GetDoorStatusCoroutine(message));

                yield return new WaitForSeconds(2f);

                isProcessing = false;
            }

            yield return null;
        }
    }

	// 受信メッセージからドアの状態を取得する関数
    private IEnumerator GetDoorStatusCoroutine(string message)
    {
        isProcessing = true;
        bool doorStatus = false;
        string responseText = message;
        Debug.Log("responseText" + responseText);

        // b. 受信したセンサーデータのパース
        ResponseData responseData = JsonUtility.FromJson <ResponseData>(responseText);
        string doorId = responseData.LogicalID; // ドアID
        sensorTime = DateTime.Parse(responseData.ArriveTime); // センサーの時間
        doorStatus = responseData.IsClosed; // ドアのステータス

        // c. データ中のドアIDに応じてドアを開閉する処理の呼び出し
        if (doorId == "1")
        {
            DoorController01.instance.SetDoorStatus(doorStatus,sensorTime);
        } else if (doorId == "2")
        {
            DoorController02.instance.SetDoorStatus(doorStatus,sensorTime);
        } else if (doorId == "3")
        {
            DoorController03.instance.SetDoorStatus(doorStatus,sensorTime);
        } else if (doorId == "4")
        {
            DoorController04.instance.SetDoorStatus(doorStatus,sensorTime);
        } else
        {
            Debug.Log("wrong id");
        }
        yield return null;
        isProcessing = false;
    }

    private void OnWebSocketMessage(object sender, MessageEventArgs e)
    {
        if (e.IsBinary)
        {
            Debug.Log("WebSocket Binary Message Received");
        }
        else
        {
            Debug.Log("WebSocket Text Message Received: " + e.Data);
            messageQueue.Enqueue(e.Data);
        }
    }

        void OnDestroy()
    {
        ws.Close();
        ws = null;
    }
}

using System;
using System.Collections;
using System.Collections.Generic;
using UnityEngine;
using UnityEngine.UI;
using TMPro;

public class DoorController01 : MonoBehaviour
{
    public GameObject door;  
    public Transform doorTransform; 
    public Quaternion closedRotation = Quaternion.Euler(-90, 0, -90); 
    public Quaternion openedRotation = Quaternion.Euler(-90, 0, 0); 
    private bool isDoorClosed;  
    
    public GameObject stopwatch; 
    public DateTime sensorTime;
    private TimeSpan elapsedTime;
    private bool isStopwatchOn;

    public static DoorController01 instance;

    public void Awake()
    {
        if(instance == null)
        {
            instance = this;
        }
    }

    // 起動時に呼び出される関数
    void Start()
    {
        isDoorClosed = true;
        isStopwatchOn = false;
        doorTransform = GetComponent <Transform>();
        doorTransform.rotation = closedRotation;
        stopwatch.SetActive(false);
    }

    // 毎フレーム呼び出される関数
    private void Update()
    {
        // b. データの受信時刻と現在時刻の差分を計算し、ドアが閉まってから現在までの時間を算出する。
        if (isStopwatchOn)
        {
            elapsedTime = DateTime.Now - sensorTime;
            if (elapsedTime.TotalHours >= 1)
            {
                stopwatch.GetComponent <TextMeshPro>().text = elapsedTime.ToString(@"hh\:mm\:ss");
            }
            else
            {
                stopwatch.GetComponent <TextMeshPro>().text = elapsedTime.ToString(@"mm\:ss");
            }
        }
    }

    // a. データの受信時刻とドアの状態を受け取り、ドアのオブジェクトの開閉を制御する。
    public void SetDoorStatus(bool doorStatus, DateTime time)
    {
        
        // "前回データより新しいデータを受信"かつ"ドアの状態に変化があった"場合にドア開閉処理を実行
        if (sensorTime <time &&doorStatus != isDoorClosed)
        {
            sensorTime = time;
            isDoorClosed = doorStatus;
            if (!isDoorClosed)
            {
                OpenDoor();
            }
            else
            {
                CloseDoor();
            }
        }
    }

    //ドアを開く関数
    void OpenDoor()
    {
        Debug.Log("isDoorClosed：" + isDoorClosed);
        if (!isDoorClosed)
        {
            // 開く処理
            StartCoroutine(OpenDoorCoroutine());
            // 開いたら経過時間は表示しない
            isStopwatchOn = false;
            stopwatch.SetActive(false);
        }
    }

    // ドアを開く処理
    IEnumerator OpenDoorCoroutine()
    {
        Quaternion fromRotation = door.transform.rotation;
        Quaternion toRotation = openedRotation;
        float duration = 1.0f; 
        float elapsed = 0.0f;

        while (elapsed <duration)
        {
            door.transform.rotation = Quaternion.Slerp(fromRotation, toRotation, elapsed / duration);
            elapsed += Time.deltaTime;

            float t = elapsed / duration;
            door.transform.rotation = ClampRotation(door.transform.rotation, t);

            yield return null;
        }

        door.transform.rotation = toRotation;
        isDoorClosed = false;
    }

    //ドアを閉じる関数
    void CloseDoor()
    {
        if (isDoorClosed)
        {
            // 閉じる処理
            StartCoroutine(CloseDoorCoroutine());

            // c. ドアが「開」から「閉」になったときだけ空間に時間を表示させる
            isStopwatchOn = true;
            stopwatch.SetActive(true);
        }
    }

    //ドアを閉じる処理
    IEnumerator CloseDoorCoroutine()
    {
        Quaternion fromRotation = door.transform.rotation;
        Quaternion toRotation = closedRotation;
        float duration = 1.0f;
        float elapsed = 0.0f;

        while (elapsed <duration)
        {
            door.transform.rotation = Quaternion.Slerp(fromRotation, toRotation, elapsed / duration);
            elapsed += Time.deltaTime;

            float t = elapsed / duration;
            door.transform.rotation = ClampRotation(door.transform.rotation, t);

            yield return null;
        }

        door.transform.rotation = toRotation;
        isDoorClosed = true;
    }

    // ドアの回転の制御
    Quaternion ClampRotation(Quaternion rotation, float t)
    {
        Quaternion clampedRotation = rotation;
        Vector3 euler = clampedRotation.eulerAngles;

        Quaternion minRotation = Quaternion.Euler(-90f, 0f, 0f);
        Quaternion maxRotation = Quaternion.Euler(-90f, 0f, -90f);
        Quaternion targetRotation = Quaternion.Lerp(minRotation, maxRotation, t);

        if (Quaternion.Angle(rotation, targetRotation) > 0.01f)
        {
            clampedRotation = Quaternion.RotateTowards(rotation, targetRotation, 1f);
        }
        return clampedRotation;
    }