How to Stream OPC UA Data to Kinesis Using AWS SiteWise Edge Gateway

1
cbor2~=5.4.2
2
stream-manager==1.1.1

1
"""
2
Script to use Greengrass Stream Manager to stream data to a Kinesis Data Stream
3
See also https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/blob/main/samples/stream_manager_kinesis.py
4
"""
5

6
import argparse
7
import asyncio
8
import logging
9
import time
10

11
from stream_manager import (
12
    ExportDefinition,
13
    KinesisConfig,
14
    MessageStreamDefinition,
15
    ReadMessagesOptions,
16
    ResourceNotFoundException,
17
    StrategyOnFull,
18
    StreamManagerClient,
19
)
20

21
logging.basicConfig(level=logging.INFO)
22
logger = logging.getLogger()
23

24

25
def main(
26
        stream_name: str,
27
        kinesis_stream_name: str,
28
        batch_size: int = None
29
):
30
    try:
31
        # Create a client for the StreamManager
32
        client = StreamManagerClient()
33

34
        # Try deleting the stream (if it exists) so that we have a fresh start
35
        try:
36
            client.delete_message_stream(stream_name=stream_name)
37
        except ResourceNotFoundException:
38
            pass
39

40
        exports = ExportDefinition(
41
            kinesis=[KinesisConfig(
42
                identifier="KinesisExport" + stream_name,
43
                kinesis_stream_name=kinesis_stream_name,
44
                batch_size=batch_size,
45
            )]
46
        )
47
        client.create_message_stream(
48
            MessageStreamDefinition(
49
                name=stream_name,
50
                strategy_on_full=StrategyOnFull.OverwriteOldestData,
51
                export_definition=exports
52
            )
53
        )
54

55
        while True:
56
            time.sleep(1)
57

58
    except asyncio.TimeoutError:
59
        logger.exception("Timed out while executing")
60
    except Exception:
61
        logger.exception("Exception while running")
62
    finally:
63
        # Always close the client to avoid resource leaks
64
        if client:
65
            client.close()
66

67

68
def parse_args() -> argparse.Namespace:
69
    parser = argparse.ArgumentParser()
70
    parser.add_argument('--greengrass-stream', required=True, default='SiteWise_Stream_Kinesis')
71
    parser.add_argument('--kinesis-stream', required=True)
72
    parser.add_argument('--batch-size', required=False, type=int, default=500)
73
    return parser.parse_args()
74

75

76
if __name__ == '__main__':
77
    args = parse_args()
78
    logger.info(f'args: {args.__dict__}')
79
    main(args.greengrass_stream, args.kinesis_stream, args.batch_size)

1
# Replace $ArtifactsS3Bucket with your value to complete component registration.
2

3
RecipeFormatVersion: 2020-01-25
4

5
ComponentName: jp.co.xyz.StreamManagerKinesis
6
ComponentVersion: 1.0.0
7
ComponentDescription: Streams data in Greengrass stream to a Kinesis Data Stream.
8
ComponentPublisher: self
9
ComponentDependencies:
10
  aws.greengrass.StreamManager:
11
    VersionRequirement: '^2.0.0'
12
ComponentConfiguration:
13
  DefaultConfiguration:
14
    GreengrassStream: SiteWise_Stream_Kinesis
15
    KinesisStream: ''
16
    BatchSize: 100 # minimum 1, maximum 500
17

18
Manifests:
19
  - Platform:
20
      os: linux
21
    Lifecycle:
22
      Install: pip3 install --user -r {artifacts:decompressedPath}/component/requirements.txt
23
      Run: |
24
        export PYTHONPATH=$PYTHONPATH:{artifacts:decompressedPath}/stream_manager_sdk
25
        python3 {artifacts:decompressedPath}/component/kinesis_data_stream.py \
26
          --greengrass-stream {configuration:/GreengrassStream} \
27
          --kinesis-stream {configuration:/KinesisStream} \
28
          --batch-size {configuration:/BatchSize}
29
    Artifacts:
30
      - URI: s3://$ArtifactsS3Bucket/artifacts/jp.co.xyz.StreamManagerKinesis/1.0.0/component.zip
31
        Unarchive: ZIP