ラズパイで部屋の環境(温度湿度気圧CO2)をモニタリングするIoTデバイスを作る③ ～DynamoDBへのデータ格納～

## はじめに
この記事は下記記事の続きとなっていますので先にこちらをご覧ください。

https://qiita.com/h0uk1st4r/items/42ce9ede4a54e05d9955

本記事ではAWS IoT Coreに送信されたデータを時系列データとして蓄えることができるようにします。

## データを蓄える基盤を作る
### 構成
データを蓄えるためのテーブルとしてDynamoDBを、デバイスシャドウから最新データを取得しDynamoDBにアイテムを作成する機能をLambda関数を追加し図のような構成にします。
![構成図](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3870591/c575e2fa-c30a-e285-b122-bdda83001ce8.png)

### SAMを使用してDynamoDBとLambda関数を構築する
DynamoDB及びLambda関数はSAM(Serverless Application Model)を使用して構築します。
#### サンプルアプリケーションの初期化
1. 今回使用する`sam cli`のバージョンは1.129.0でした
    ```shell-session
    $ sam --version
    SAM CLI, version 1.129.0
    ```
1. `sam init`コマンドで対話しながらアプリケーションを作成します
    ```shell-session
    $ sam init
    ```
1. 「Which template source would you like to use?」は楽をするため「1」を選択
    ```shell-session
    Which template source would you like to use?
            1 - AWS Quick Start Templates
            2 - Custom Template Location
    Choice: 1
    ```
1. 「Choose an AWS Quick Start application template」は最小限の「1 - Hello World Example」を選択
    ```shell-session
    Choose an AWS Quick Start application template
            1 - Hello World Example
            2 - Data processing
            3 - Hello World Example with Powertools for AWS Lambda
            4 - Multi-step workflow
            5 - Scheduled task
            6 - Standalone function
            7 - Serverless API
            8 - Infrastructure event management
            9 - Lambda Response Streaming
            10 - Serverless Connector Hello World Example
            11 - Multi-step workflow with Connectors
            12 - GraphQLApi Hello World Example
            13 - Full Stack
            14 - Lambda EFS example
            15 - DynamoDB Example
            16 - Machine Learning
    Template: 1
    ```
1. 「Use the most popular runtime and package type? (python3.13 and zip) 」は「Enter(N)」を選択し、「runtime」は「14 - python3.9」を「package type」は「1 - Zip」を選択
    ※今回の環境内のPythonが3.9のため
    ```shell-session
    Use the most popular runtime and package type? (python3.13 and zip) [y/N]: 

    Which runtime would you like to use?
            1 - dotnet8
            2 - dotnet6
            3 - go (provided.al2)
            4 - go (provided.al2023)
            5 - graalvm.java11 (provided.al2)
            6 - graalvm.java17 (provided.al2)
            7 - java21
            8 - java17
            9 - java11
            10 - java8.al2
            11 - nodejs20.x
            12 - nodejs18.x
            13 - nodejs16.x
            14 - python3.9
            15 - python3.8
            16 - python3.13
            17 - python3.12
            18 - python3.11
            19 - python3.10
            20 - ruby3.3
            21 - ruby3.2
            22 - rust (provided.al2)
            23 - rust (provided.al2023)
    Runtime: 14

    What package type would you like to use?
            1 - Zip
            2 - Image
    Package type: 1
    ```
1. アプリケーションの分析やロギング関連のオプションはすべて「Enter(N)」
    ```shell-session
    Would you like to enable X-Ray tracing on the function(s) in your application?  [y/N]: 

    Would you like to enable monitoring using CloudWatch Application Insights?
    For more info, please view https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/cloudwatch-application-insights.html [y/N]: 

    Would you like to set Structured Logging in JSON format on your Lambda functions?  [y/N]: 
    ```
1. 「Project name」は「EnvDataStack」としてサンプルアプリケーションの初期化は完了です
    ```shell-session
    Project name [sam-app]: EnvDataStack
    
    -----------------------
    Generating application:
    -----------------------
    Name: EnvDataStack
    Runtime: python3.9
    Architectures: x86_64
    Dependency Manager: pip
    Application Template: hello-world
    Output Directory: .
    Configuration file: EnvDataStack/samconfig.toml
    
    Next steps can be found in the README file at EnvDataStack/README.md
            
    
    Commands you can use next
    =========================
    [*] Create pipeline: cd EnvDataStack && sam pipeline init --bootstrap
    [*] Validate SAM template: cd EnvDataStack && sam validate
    [*] Test Function in the Cloud: cd EnvDataStack && sam sync --stack-name {stack-name} --watch
    ```

#### ソースコードの追加
IoT Coreの「shadow/update/accepted」イベントを受信してDynamoDBに受信したセンサーデータを格納するLambda関数のコードをこのアプリケーションに追加します。
1. `src`フォルダの追加
    ```shell-session
    [EnvDataStack]$ mkdir src
    ```
1. `lambda_function.py`の追加
    `vi`コマンドで次のPythonスクリプトを追加します。
    ```shell-session
    [EnvDataStack]$ vi src/lambda_function.py
    ```
    ```python:lambda_function.py
    import os
    import sys
    from typing import Literal
    from decimal import Decimal
    from datetime import datetime, timezone
    from dateutil.relativedelta import relativedelta
    from logging import Logger, getLogger, Formatter, StreamHandler

    import boto3
    from botocore.config import Config
    from boto3.dynamodb.types import TypeSerializer

    class EnvDataPusher:
        LOGGING_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = os.environ.get("LOGGING_LEVEL", "WARNING")

        TABLE_ARN: str = os.environ["TABLE_ARN"]

        def __init__(self) -> None:
            self._create_logger()
            self._create_dynamodb_client()

        def _create_logger(self) -> None:
            self._logger: Logger = getLogger(self.__class__.__name__)
            self._logger.propagate = False
            for handler in self._logger.handlers:
                self._logger.removeHandler(handler)
            handler: StreamHandler = StreamHandler(sys.stdout)
            handler.setFormatter(Formatter("%(asctime)s [%(levelname)s] %(funcName)s %(message)s"))
            self._logger.addHandler(handler)
            self._logger.setLevel(self.LOGGING_LEVEL)

        def _create_dynamodb_client(self) -> None:
            self._dynamodb_client = boto3.client(
                "dynamodb",
                config=Config(
                    retries={
                        "max_attempts": 5,
                        "mode": "standard"
                    }
                )
            )

        def _put_item_to_table(self, type: str, value: Decimal) -> bool:
            timestamp: datetime = datetime.now(timezone.utc)
            ttl: datetime = timestamp + relativedelta(days=1)
            try:
                self._dynamodb_client.put_item(
                    TableName=self.TABLE_ARN,
                    Item={
                        k: TypeSerializer().serialize(v)
                        for k, v in {
                            "Type": type,
                            "Timestamp": timestamp.strftime("%Y-%m-%dT%H:%M:%SZ"),
                            "Value": value,
                            "TTL": Decimal(str(ttl.timestamp()))
                        }.items()
                    }
                )
            except Exception as e:
                self._logger.error(f"Failed put_item, reason:\n{e}")
                return False
            return True

        def execute(self, event: dict) -> bool:
            temperature: Decimal = Decimal(str(event["state"]["reported"]["temperature"]))
            humidity: Decimal = Decimal(str(event["state"]["reported"]["humidity"]))
            pressure: Decimal = Decimal(str(event["state"]["reported"]["pressure"]))
            concentration: Decimal = Decimal(str(event["state"]["reported"]["concentration"]))

            self._logger.debug(f"{temperature=}")
            self._logger.debug(f"{humidity=}")
            self._logger.debug(f"{pressure=}")
            self._logger.debug(f"{concentration=}")

            is_succeeded: bool = True
            if not self._put_item_to_table("temperature", temperature):
                is_succeeded = False
            if not self._put_item_to_table("humidity", humidity):
                is_succeeded = False
            if not self._put_item_to_table("pressure", pressure):
                is_succeeded = False
            if not self._put_item_to_table("concentration", concentration):
                is_succeeded = False

            return is_succeeded

    pusher: EnvDataPusher = EnvDataPusher()

    def lambda_handler(event: dict, context: dict) -> None:
        is_succeeded: bool = pusher.execute(event)
        if not is_succeeded:
            raise Exception("データ格納処理に失敗")

    ```

    Pythonスクリプト自体は受け取ったイベントデータを元にDynamoDBにタイムスタンプ付きでデータを格納する簡単なものにしています。

#### テンプレートの修正
DynamoDBテーブル、Lambda関数及びLambda関数をトリガーするIoT Ruleを構築することができるようにSAMテンプレート`template.yaml`を修正します。
```shell-session
[EnvDataStack]$ vi template.yaml 
```
```yaml:更新後のtemplate.yaml
AWSTemplateFormatVersion: "2010-09-09"
Transform: "AWS::Serverless-2016-10-31"
Description: "EnvDataStack"

Resources:
  EnvDataTable:  # データ格納用DynamoDBテーブル
    Type: "AWS::DynamoDB::Table"
    Properties:
      AttributeDefinitions:
        - AttributeName: "Type"
          AttributeType: "S"
        - AttributeName: "Timestamp"
          AttributeType: "S"
      BillingMode: "PROVISIONED"
      DeletionProtectionEnabled: true
      KeySchema:
        - AttributeName: "Type"
          KeyType: "HASH"
        - AttributeName: "Timestamp"
          KeyType: "RANGE"
      ProvisionedThroughput:
        ReadCapacityUnits: 1
        WriteCapacityUnits: 1
      TableName: "EnvDataTable"
      TimeToLiveSpecification:
        AttributeName: "TTL"
        Enabled: true

  EnvDataPusher:  # データ格納用Lambda関数
    Type: "AWS::Serverless::Function"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Service:
                - "lambda.amazonaws.com"
            Action:
              - "sts:AssumeRole"
      AutoPublishAlias: "Alias"
      CodeUri: "src/"
      Description: "iot-env-data-collectorのデバイスシャドウ更新に応じて最新のデータをDynamoDBにプッシュする関数"
      Environment:
        Variables:
          TABLE_ARN: !GetAtt "EnvDataTable.Arn"
      Events:
        ShadowUpdateAccepted:
          Type: "IoTRule"
          Properties:
            Sql: "SELECT * FROM \"$aws/things/iot-env-data-collector/shadow/update/accepted\""
      FunctionName: "EnvDataPusher"
      Handler: "lambda_function.lambda_handler"
      MemorySize: 128
      Policies:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Action:
              - "logs:CreateLogGroup"
              - "logs:CreateLogStream"
              - "logs:PutLogEvents"
            Resource: "*"
          - Effect: "Allow"
            Action:
              - "dynamodb:PutItem"
            Resource: !GetAtt "EnvDataTable.Arn"
      Runtime: "python3.9"
      Timeout: 30

```
#### `sam build`コマンドでデプロイの準備をします
`Build Succeeded`となればおｋです。
```shell-session
[EnvDataStack]$ sam build
Starting Build use cache                                                                                                                                                                                                                                                                 
Building codeuri: /tmp/EnvDataStack/src runtime: python3.9 architecture: x86_64 functions: EnvDataPusher                                                                                                                                                                                 
requirements.txt file not found. Continuing the build without dependencies.                                                                                                                                                                                                              
 Running PythonPipBuilder:CopySource                                                                                                                                                                                                                                                     

Build Succeeded

Built Artifacts  : .aws-sam/build
Built Template   : .aws-sam/build/template.yaml

Commands you can use next
=========================
[*] Validate SAM template: sam validate
[*] Invoke Function: sam local invoke
[*] Test Function in the Cloud: sam sync --stack-name {{stack-name}} --watch
[*] Deploy: sam deploy --guided
```
#### アプリケーションをデプロイします
アプリケーションをAWSにデプロイしていきます。
1. `sam deploy --guided`コマンドで対話的に実施します
    ```shell-session
    [EnvDataStack]$ sam deploy --guided
    ```
1. 「Stack Name」はそのままでいいので「Enter」
    ```shell-session
    Stack Name [EnvDataStack]: 
    ```
1. 「AWS Region」は東京リージョンにデプロイするので「Enter」
    ```shell-session
    AWS Region [ap-northeast-1]: 
    ```
1. 「Confirm changes before deploy」はデプロイの度確認不要なので「n」
    ```shell-session
    #Shows you resources changes to be deployed and require a 'Y' to initiate deploy
    Confirm changes before deploy [Y/n]: n
    ```
1. 「Allow SAM CLI IAM role creation」は自動的に必要なロールを作ってもらって問題ないので「Enter(Y)」
    ```shell-session
    #SAM needs permission to be able to create roles to connect to the resources in your template
    Allow SAM CLI IAM role creation [Y/n]: 
    ```
1. 「Disable rollback」はデプロイに失敗したらロールバックしてほしいので「Enter(N)」
    ```shell-session
    #Preserves the state of previously provisioned resources when an operation fails
    Disable rollback [y/N]: 
    ```
1. 今まで答えた事を保存してくれるのですべてそのまま「Enter」
    ```shell-session
    Save arguments to configuration file [Y/n]:
    SAM configuration file [samconfig.toml]:
    SAM configuration environment [default]:
    ```
1. すべての質問に回答し終わるとリソースの構築が開始される
    定期的に表示が更新され`Successfully created/updated stack`と表示されればアプリケーションのデプロイは完了です。
    ```shell-session
    CloudFormation events from stack operations (refresh every 5.0 seconds)
    -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    ResourceStatus                                                        ResourceType                                                          LogicalResourceId                                                     ResourceStatusReason                                                
    -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    CREATE_IN_PROGRESS                                                    AWS::CloudFormation::Stack                                            EnvDataStack                                                          User Initiated                                                      
    CREATE_IN_PROGRESS                                                    AWS::DynamoDB::Table                                                  EnvDataTable                                                          -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::DynamoDB::Table                                                  EnvDataTable                                                          Resource creation Initiated                                         
    CREATE_COMPLETE                                                       AWS::DynamoDB::Table                                                  EnvDataTable                                                          -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::IAM::Role                                                        EnvDataPusherRole                                                     -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::IAM::Role                                                        EnvDataPusherRole                                                     Resource creation Initiated                                         
    CREATE_COMPLETE                                                       AWS::IAM::Role                                                        EnvDataPusherRole                                                     -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::Lambda::Function                                                 EnvDataPusher                                                         -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::Lambda::Function                                                 EnvDataPusher                                                         Resource creation Initiated                                         
    CREATE_COMPLETE                                                       AWS::Lambda::Function                                                 EnvDataPusher                                                         -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::Lambda::Version                                                  EnvDataPusherVersion68a151ade3                                        -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::Lambda::Version                                                  EnvDataPusherVersion68a151ade3                                        Resource creation Initiated                                         
    CREATE_COMPLETE                                                       AWS::Lambda::Version                                                  EnvDataPusherVersion68a151ade3                                        -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::Lambda::Alias                                                    EnvDataPusherAliasAlias                                               -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::Lambda::Alias                                                    EnvDataPusherAliasAlias                                               Resource creation Initiated                                         
    CREATE_COMPLETE                                                       AWS::Lambda::Alias                                                    EnvDataPusherAliasAlias                                               -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::IoT::TopicRule                                                   EnvDataPusherShadowUpdateAccepted                                     -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::IoT::TopicRule                                                   EnvDataPusherShadowUpdateAccepted                                     Resource creation Initiated                                         
    CREATE_COMPLETE                                                       AWS::IoT::TopicRule                                                   EnvDataPusherShadowUpdateAccepted                                     -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::Lambda::Permission                                               EnvDataPusherShadowUpdateAcceptedPermission                           -                                                                   
    CREATE_IN_PROGRESS                                                    AWS::Lambda::Permission                                               EnvDataPusherShadowUpdateAcceptedPermission                           Resource creation Initiated                                         
    CREATE_COMPLETE                                                       AWS::Lambda::Permission                                               EnvDataPusherShadowUpdateAcceptedPermission                           -                                                                   
    CREATE_COMPLETE                                                       AWS::CloudFormation::Stack                                            EnvDataStack                                                          -                                                                   
    -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    
    
    Successfully created/updated stack - EnvDataStack in ap-northeast-1
    ```

### 収集したセンサーデータを確認する
SAMはCloudFormationの拡張したものであるため、今回構築したリソースやスタックのステータスなどはCloudFormationのマネコンからでも確認することができます。
    
マネコンで「CloudFormation > スタック > EnvDataStack > リソースタブ」を開くと作成したリソースが表示されています。
![001.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3870591/1ee9b6db-aea9-145b-cc7f-55177df35312.png)

DynamoDBテーブルである「EnvDataTable」を開き「項目の探索」を行うと1分間隔で各種センサーデータが蓄えられていることが確認できます。
![002.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3870591/8513264c-fe42-88b6-d0d0-10be52a78e25.png)

## さいごに
IoT Coreに送信されたセンサーデータをDynamoDBに時系列データとして蓄えることができるようになりました。
しかしセンサ―データを時系列データとして蓄えることができるようになったことで、欲が出てきてグラフィカルに表示させたくなりますよね。

次の記事では各センサーデータをグラフィカルに表示するWEBアプリを作成して〆たいと思います。

https://qiita.com/h0uk1st4r/items/8f0121f48534cb9df4e1