Doc AWS SDK 예제 GitHub 리포지토리에서 더 많은 SDK 예제를 사용할 수 있습니다. [AWS](https://github.com/awsdocs/aws-doc-sdk-examples)
기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.
# AWS IoT SiteWise SDK for Python(Boto3)을 사용한 예제
다음 코드 예제에서는와 AWS SDK for Python (Boto3) 함께를 사용하여 작업을 수행하고 일반적인 시나리오를 구현하는 방법을 보여줍니다 AWS IoT SiteWise.
*기본 사항*은 서비스 내에서 필수 작업을 수행하는 방법을 보여주는 코드 예제입니다.
*작업*은 대규모 프로그램에서 발췌한 코드이며 컨텍스트에 맞춰 실행해야 합니다. 작업은 개별 서비스 함수를 직접적으로 호출하는 방법을 보여주며 관련 시나리오의 컨텍스트에 맞는 작업을 볼 수 있습니다.
각 예시에는 전체 소스 코드에 대한 링크가 포함되어 있으며, 여기에서 컨텍스트에 맞춰 코드를 설정하고 실행하는 방법에 대한 지침을 찾을 수 있습니다.
**Topics**
+ [시작하기](#get_started)
+ [기본 사항](#basics)
+ [작업](#actions)
## 시작하기
### 안녕하세요 AWS IoT SiteWise
다음 코드 예제에서는 AWS IoT SiteWise를 사용하여 시작하는 방법을 보여 줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
import boto3
def hello_iot_sitewise(iot_sitewise_client):
"""
Use the AWS SDK for Python (Boto3) to create an AWS IoT SiteWise
client and list the asset models in your account.
This example uses the default settings specified in your shared credentials
and config files.
:param iot_sitewise_client: A Boto3 AWS IoT SiteWise Client object. This object wraps
the low-level AWS IoT SiteWise service API.
"""
print("Hello, AWS IoT SiteWise! Let's list some of your asset models:\n")
paginator = iot_sitewise_client.get_paginator("list_asset_models")
page_iterator = paginator.paginate(PaginationConfig={"MaxItems": 10})
asset_model_names: [str] = []
for page in page_iterator:
for asset_model in page["assetModelSummaries"]:
asset_model_names.append(asset_model["name"])
print(f"{len(asset_model_names)} asset model(s) retrieved.")
for asset_model_name in asset_model_names:
print(f"\t{asset_model_name}")
if __name__ == "__main__":
hello_iot_sitewise(boto3.client("iotsitewise"))
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [ListAssetModels](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/ListAssetModels)을 참조하세요.
## 기본 사항
### 기본 사항 알아보기
다음 코드 예제에서는 다음과 같은 작업을 수행하는 방법을 보여줍니다.
+ AWS IoT SiteWise 자산 모델을 생성합니다.
+ AWS IoT SiteWise 자산을 생성합니다.
+ 속성 ID 값을 검색합니다.
+ AWS IoT SiteWise 애셋으로 데이터를 전송합니다.
+ AWS IoT SiteWise Asset 속성의 값을 검색합니다.
+ AWS IoT SiteWise 포털을 생성합니다.
+ AWS IoT SiteWise 게이트웨이를 생성합니다.
+ AWS IoT SiteWise 게이트웨이를 설명합니다.
+ AWS IoT SiteWise 자산을 삭제합니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
명령 프롬프트에서 대화형 시나리오를 실행합니다.
```
class IoTSitewiseGettingStarted:
"""
A scenario that demonstrates how to use Boto3 to manage IoT physical assets using
the AWS IoT SiteWise.
"""
def __init__(
self,
iot_sitewise_wrapper: IoTSitewiseWrapper,
cloud_formation_resource: ServiceResource,
):
self.iot_sitewise_wrapper = iot_sitewise_wrapper
self.cloud_formation_resource = cloud_formation_resource
self.stack = None
self.asset_model_id = None
self.asset_id = None
self.portal_id = None
self.gateway_id = None
def run(self) -> None:
"""
Runs the scenario.
"""
print(
"""
AWS IoT SiteWise is a fully managed software-as-a-service (SaaS) that
makes it easy to collect, store, organize, and monitor data from industrial equipment and processes.
It is designed to help industrial and manufacturing organizations collect data from their equipment and
processes, and use that data to make informed decisions about their operations.
One of the key features of AWS IoT SiteWise is its ability to connect to a wide range of industrial
equipment and systems, including programmable logic controllers (PLCs), sensors, and other
industrial devices. It can collect data from these devices and organize it into a unified data model,
making it easier to analyze and gain insights from the data. AWS IoT SiteWise also provides tools for
visualizing the data, setting up alarms and alerts, and generating reports.
Another key feature of AWS IoT SiteWise is its ability to scale to handle large volumes of data.
It can collect and store data from thousands of devices and process millions of data points per second,
making it suitable for large-scale industrial operations. Additionally, AWS IoT SiteWise is designed
to be secure and compliant, with features like role-based access controls, data encryption,
and integration with other AWS services for additional security and compliance features.
Let's get started...
"""
)
press_enter_to_continue()
print_dashes()
print(f"")
print(
f"Use AWS CloudFormation to create an IAM role that is required for this scenario."
)
template_file = IoTSitewiseGettingStarted.get_template_as_string()
self.stack = self.deploy_cloudformation_stack(
"python-iot-sitewise-basics", template_file
)
outputs = self.stack.outputs
iam_role = None
for output in outputs:
if output.get("OutputKey") == "SitewiseRoleArn":
iam_role = output.get("OutputValue")
if iam_role is None:
error_string = f"Failed to retrieve iam_role from CloudFormation stack."
logger.error(error_string)
raise ValueError(error_string)
print(f"The ARN of the IAM role is {iam_role}")
print_dashes()
print_dashes()
print(f"1. Create an AWS SiteWise Asset Model")
print(
"""
An AWS IoT SiteWise Asset Model is a way to represent the physical assets, such as equipment,
processes, and systems, that exist in an industrial environment. This model provides a structured and
hierarchical representation of these assets, allowing users to define the relationships and values
of each asset.
This scenario creates two asset model values: temperature and humidity.
"""
)
press_enter_to_continue()
asset_model_name = "MyAssetModel1"
temperature_property_name = "temperature"
humidity_property_name = "humidity"
try:
properties = [
{
"name": temperature_property_name,
"dataType": "DOUBLE",
"type": {
"measurement": {},
},
},
{
"name": humidity_property_name,
"dataType": "DOUBLE",
"type": {
"measurement": {},
},
},
]
self.asset_model_id = self.iot_sitewise_wrapper.create_asset_model(
asset_model_name, properties
)
print(
f"Asset Model successfully created. Asset Model ID: {self.asset_model_id}. "
)
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceAlreadyExistsException":
self.asset_model_id = self.get_model_id_for_model_name(asset_model_name)
print(
f"Asset Model {asset_model_name} already exists. Asset Model ID: {self.asset_model_id}. "
)
else:
raise
press_enter_to_continue()
print_dashes()
print(f"2. Create an AWS IoT SiteWise Asset")
print(
"""
The IoT SiteWise model that we just created defines the structure and metadata for your physical assets.
Now we create an asset from the asset model.
"""
)
press_enter_to_continue()
self.asset_id = self.iot_sitewise_wrapper.create_asset(
"MyAsset1", self.asset_model_id
)
print(f"Asset created with ID: {self.asset_id}")
press_enter_to_continue()
print_dashes()
print_dashes()
print(f"3. Retrieve the property ID values")
print(
"""
To send data to an asset, we need to get the property ID values. In this scenario, we access the
temperature and humidity property ID values.
"""
)
press_enter_to_continue()
property_ids = self.iot_sitewise_wrapper.list_asset_model_properties(
self.asset_model_id
)
humidity_property_id = None
temperature_property_id = None
for property_id in property_ids:
if property_id.get("name") == humidity_property_name:
humidity_property_id = property_id.get("id")
elif property_id.get("name") == temperature_property_name:
temperature_property_id = property_id.get("id")
if humidity_property_id is None or temperature_property_id is None:
error_string = f"Failed to retrieve property IDs from Asset Model."
logger.error(error_string)
raise ValueError(error_string)
print(f"The Humidity property Id is {humidity_property_id}")
print(f"The Temperature property Id is {temperature_property_id}")
press_enter_to_continue()
print_dashes()
print_dashes()
print(f"4. Send data to an AWS IoT SiteWise Asset")
print(
"""
By sending data to an IoT SiteWise Asset, you can aggregate data from
multiple sources, normalize the data into a standard format, and store it in a
centralized location. This makes it easier to analyze and gain insights from the data.
In this example, we generate sample temperature and humidity data and send it to the AWS IoT SiteWise asset.
"""
)
press_enter_to_continue()
values = [
{
"propertyId": humidity_property_id,
"valueType": "doubleValue",
"value": 65.0,
},
{
"propertyId": temperature_property_id,
"valueType": "doubleValue",
"value": 23.5,
},
]
self.iot_sitewise_wrapper.batch_put_asset_property_value(self.asset_id, values)
print(f"Data sent successfully.")
press_enter_to_continue()
print_dashes()
print_dashes()
print(f"5. Retrieve the value of the IoT SiteWise Asset property")
print(
"""
IoT SiteWise is an AWS service that allows you to collect, process, and analyze industrial data
from connected equipment and sensors. One of the key benefits of reading an IoT SiteWise property
is the ability to gain valuable insights from your industrial data.
"""
)
press_enter_to_continue()
property_value = self.iot_sitewise_wrapper.get_asset_property_value(
self.asset_id, temperature_property_id
)
print(f"The property name is '{temperature_property_name}'.")
print(
f"The value of this property is: {property_value['value']['doubleValue']}"
)
press_enter_to_continue()
property_value = self.iot_sitewise_wrapper.get_asset_property_value(
self.asset_id, humidity_property_id
)
print(f"The property name is '{humidity_property_name}'.")
print(
f"The value of this property is: {property_value['value']['doubleValue']}"
)
press_enter_to_continue()
print_dashes()
print_dashes()
print(f"6. Create an IoT SiteWise Portal")
print(
"""
An IoT SiteWise Portal allows you to aggregate data from multiple industrial sources,
such as sensors, equipment, and control systems, into a centralized platform.
"""
)
press_enter_to_continue()
contact_email = q.ask("Enter a contact email for the portal:", q.non_empty)
print("Creating the portal. The portal may take a while to become active.")
self.portal_id = self.iot_sitewise_wrapper.create_portal(
"MyPortal1", iam_role, contact_email
)
print(f"Portal created successfully. Portal ID {self.portal_id}")
press_enter_to_continue()
print_dashes()
print_dashes()
print(f"7. Describe the Portal")
print(
"""
In this step, we get a description of the portal and display the portal URL.
"""
)
press_enter_to_continue()
portal_description = self.iot_sitewise_wrapper.describe_portal(self.portal_id)
print(f"Portal URL: {portal_description['portalStartUrl']}")
press_enter_to_continue()
print_dashes()
print_dashes()
print(f"8. Create an IoT SiteWise Gateway")
press_enter_to_continue()
self.gateway_id = self.iot_sitewise_wrapper.create_gateway(
"MyGateway1", "MyThing1"
)
print(f"Gateway creation completed successfully. id is {self.gateway_id}")
print_dashes()
print_dashes()
print(f"9. Describe the IoT SiteWise Gateway")
press_enter_to_continue()
gateway_description = self.iot_sitewise_wrapper.describe_gateway(
self.gateway_id
)
print(f"Gateway Name: {gateway_description['gatewayName']}")
print(f"Gateway ARN: {gateway_description['gatewayArn']}")
print(f"Gateway Platform:\n{gateway_description['gatewayPlatform']}")
print(f"Gateway Creation Date: {gateway_description['gatewayArn']}")
print_dashes()
print_dashes()
print(f"10. Delete the AWS IoT SiteWise Assets")
if q.ask("Would you like to delete the IoT SiteWise Assets? (y/n)", q.is_yesno):
self.cleanup()
else:
print(f"The resources will not be deleted.")
print_dashes()
print_dashes()
print(f"This concludes the AWS IoT SiteWise Scenario")
def cleanup(self) -> None:
"""
Deletes the CloudFormation stack and the resources created for the demo.
"""
if self.gateway_id is not None:
self.iot_sitewise_wrapper.delete_gateway(self.gateway_id)
print(f"Deleted gateway with id {self.gateway_id}.")
self.gateway_id = None
if self.portal_id is not None:
self.iot_sitewise_wrapper.delete_portal(self.portal_id)
print(f"Deleted portal with id {self.portal_id}.")
self.portal_id = None
if self.asset_id is not None:
self.iot_sitewise_wrapper.delete_asset(self.asset_id)
print(f"Deleted asset with id {self.asset_id}.")
self.iot_sitewise_wrapper.wait_asset_deleted(self.asset_id)
self.asset_id = None
if self.asset_model_id is not None:
self.iot_sitewise_wrapper.delete_asset_model(self.asset_model_id)
print(f"Deleted asset model with id {self.asset_model_id}.")
self.asset_model_id = None
if self.stack is not None:
stack = self.stack
self.stack = None
self.destroy_cloudformation_stack(stack)
def deploy_cloudformation_stack(
self, stack_name: str, cfn_template: str
) -> ServiceResource:
"""
Deploys prerequisite resources used by the scenario. The resources are
defined in the associated `SitewiseRoles-template.yaml` AWS CloudFormation script and are deployed
as a CloudFormation stack, so they can be easily managed and destroyed.
:param stack_name: The name of the CloudFormation stack.
:param cfn_template: The CloudFormation template as a string.
:return: The CloudFormation stack resource.
"""
print(f"Deploying CloudFormation stack: {stack_name}.")
stack = self.cloud_formation_resource.create_stack(
StackName=stack_name,
TemplateBody=cfn_template,
Capabilities=["CAPABILITY_NAMED_IAM"],
)
print(f"CloudFormation stack creation started: {stack_name}")
print("Waiting for CloudFormation stack creation to complete...")
waiter = self.cloud_formation_resource.meta.client.get_waiter(
"stack_create_complete"
)
waiter.wait(StackName=stack.name)
stack.load()
print("CloudFormation stack creation complete.")
return stack
def destroy_cloudformation_stack(self, stack: ServiceResource) -> None:
"""
Destroys the resources managed by the CloudFormation stack, and the CloudFormation
stack itself.
:param stack: The CloudFormation stack that manages the example resources.
"""
print(
f"CloudFormation stack '{stack.name}' is being deleted. This may take a few minutes."
)
stack.delete()
waiter = self.cloud_formation_resource.meta.client.get_waiter(
"stack_delete_complete"
)
waiter.wait(StackName=stack.name)
print(f"CloudFormation stack '{stack.name}' has been deleted.")
@staticmethod
def get_template_as_string() -> str:
"""
Returns a string containing this scenario's CloudFormation template.
"""
template_file_path = os.path.join(script_dir, "SitewiseRoles-template.yaml")
file = open(template_file_path, "r")
return file.read()
def get_model_id_for_model_name(self, model_name: str) -> str:
"""
Returns the model ID for the given model name.
:param model_name: The name of the model.
:return: The model ID.
"""
model_id = None
asset_models = self.iot_sitewise_wrapper.list_asset_models()
for asset_model in asset_models:
if asset_model["name"] == model_name:
model_id = asset_model["id"]
break
return model_id
```
AWS IoT SiteWise 작업을 래핑하는 IoTSitewiseWrapper 클래스입니다.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def create_asset_model(
self, asset_model_name: str, properties: List[Dict[str, Any]]
) -> str:
"""
Creates an AWS IoT SiteWise Asset Model.
:param asset_model_name: The name of the asset model to create.
:param properties: The property definitions of the asset model.
:return: The ID of the created asset model.
"""
try:
response = self.iotsitewise_client.create_asset_model(
assetModelName=asset_model_name,
assetModelDescription="This is a sample asset model description.",
assetModelProperties=properties,
)
asset_model_id = response["assetModelId"]
waiter = self.iotsitewise_client.get_waiter("asset_model_active")
waiter.wait(assetModelId=asset_model_id)
return asset_model_id
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceAlreadyExistsException":
logger.error("Asset model %s already exists.", asset_model_name)
else:
logger.error(
"Error creating asset model %s. Here's why %s",
asset_model_name,
err.response["Error"]["Message"],
)
raise
def create_asset(self, asset_name: str, asset_model_id: str) -> str:
"""
Creates an AWS IoT SiteWise Asset.
:param asset_name: The name of the asset to create.
:param asset_model_id: The ID of the asset model to associate with the asset.
:return: The ID of the created asset.
"""
try:
response = self.iotsitewise_client.create_asset(
assetName=asset_name, assetModelId=asset_model_id
)
asset_id = response["assetId"]
waiter = self.iotsitewise_client.get_waiter("asset_active")
waiter.wait(assetId=asset_id)
return asset_id
except ClientError as err:
if err.response["Error"] == "ResourceNotFoundException":
logger.error("Asset model %s does not exist.", asset_model_id)
else:
logger.error(
"Error creating asset %s. Here's why %s",
asset_name,
err.response["Error"]["Message"],
)
raise
def list_asset_models(self) -> List[Dict[str, Any]]:
"""
Lists all AWS IoT SiteWise Asset Models.
:return: A list of dictionaries containing information about each asset model.
"""
try:
asset_models = []
paginator = self.iotsitewise_client.get_paginator("list_asset_models")
pages = paginator.paginate()
for page in pages:
asset_models.extend(page["assetModelSummaries"])
return asset_models
except ClientError as err:
logger.error(
"Error listing asset models. Here's why %s",
err.response["Error"]["Message"],
)
raise
def list_asset_model_properties(self, asset_model_id: str) -> List[Dict[str, Any]]:
"""
Lists all AWS IoT SiteWise Asset Model Properties.
:param asset_model_id: The ID of the asset model to list values for.
:return: A list of dictionaries containing information about each asset model property.
"""
try:
asset_model_properties = []
paginator = self.iotsitewise_client.get_paginator(
"list_asset_model_properties"
)
pages = paginator.paginate(assetModelId=asset_model_id)
for page in pages:
asset_model_properties.extend(page["assetModelPropertySummaries"])
return asset_model_properties
except ClientError as err:
logger.error(
"Error listing asset model values. Here's why %s",
err.response["Error"]["Message"],
)
raise
def batch_put_asset_property_value(
self, asset_id: str, values: List[Dict[str, str]]
) -> None:
"""
Sends data to an AWS IoT SiteWise Asset.
:param asset_id: The asset ID.
:param values: A list of dictionaries containing the values in the form
{propertyId : property_id,
valueType : [stringValue|integerValue|doubleValue|booleanValue],
value : the_value}.
"""
try:
entries = self.properties_to_values(asset_id, values)
self.iotsitewise_client.batch_put_asset_property_value(entries=entries)
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error("Asset %s does not exist.", asset_id)
else:
logger.error(
"Error sending data to asset. Here's why %s",
err.response["Error"]["Message"],
)
raise
def properties_to_values(
self, asset_id: str, values: list[dict[str, Any]]
) -> list[dict[str, Any]]:
"""
Utility function to convert a values list to the entries parameter for batch_put_asset_property_value.
:param asset_id : The asset ID.
:param values : A list of dictionaries containing the values in the form
{propertyId : property_id,
valueType : [stringValue|integerValue|doubleValue|booleanValue],
value : the_value}.
:return: An entries list to pass as the 'entries' parameter to batch_put_asset_property_value.
"""
entries = []
for value in values:
epoch_ns = time.time_ns()
self.entry_id += 1
if value["valueType"] == "stringValue":
property_value = {"stringValue": value["value"]}
elif value["valueType"] == "integerValue":
property_value = {"integerValue": value["value"]}
elif value["valueType"] == "booleanValue":
property_value = {"booleanValue": value["value"]}
elif value["valueType"] == "doubleValue":
property_value = {"doubleValue": value["value"]}
else:
raise ValueError("Invalid valueType: %s", value["valueType"])
entry = {
"entryId": f"{self.entry_id}",
"assetId": asset_id,
"propertyId": value["propertyId"],
"propertyValues": [
{
"value": property_value,
"timestamp": {
"timeInSeconds": int(epoch_ns / 1000000000),
"offsetInNanos": epoch_ns % 1000000000,
},
}
],
}
entries.append(entry)
return entries
def get_asset_property_value(
self, asset_id: str, property_id: str
) -> Dict[str, Any]:
"""
Gets the value of an AWS IoT SiteWise Asset Property.
:param asset_id: The ID of the asset.
:param property_id: The ID of the property.
:return: A dictionary containing the value of the property.
"""
try:
response = self.iotsitewise_client.get_asset_property_value(
assetId=asset_id, propertyId=property_id
)
return response["propertyValue"]
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error(
"Asset %s or property %s does not exist.", asset_id, property_id
)
else:
logger.error(
"Error getting asset property value. Here's why %s",
err.response["Error"]["Message"],
)
raise
def create_portal(
self, portal_name: str, iam_role_arn: str, portal_contact_email: str
) -> str:
"""
Creates an AWS IoT SiteWise Portal.
:param portal_name: The name of the portal to create.
:param iam_role_arn: The ARN of an IAM role.
:param portal_contact_email: The contact email of the portal.
:return: The ID of the created portal.
"""
try:
response = self.iotsitewise_client.create_portal(
portalName=portal_name,
roleArn=iam_role_arn,
portalContactEmail=portal_contact_email,
)
portal_id = response["portalId"]
waiter = self.iotsitewise_client.get_waiter("portal_active")
waiter.wait(portalId=portal_id, WaiterConfig={"MaxAttempts": 40})
return portal_id
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceAlreadyExistsException":
logger.error("Portal %s already exists.", portal_name)
else:
logger.error(
"Error creating portal %s. Here's why %s",
portal_name,
err.response["Error"]["Message"],
)
raise
def describe_portal(self, portal_id: str) -> Dict[str, Any]:
"""
Describes an AWS IoT SiteWise Portal.
:param portal_id: The ID of the portal to describe.
:return: A dictionary containing information about the portal.
"""
try:
response = self.iotsitewise_client.describe_portal(portalId=portal_id)
return response
except ClientError as err:
logger.error(
"Error describing portal %s. Here's why %s",
portal_id,
err.response["Error"]["Message"],
)
raise
def create_gateway(self, gateway_name: str, my_thing: str) -> str:
"""
Creates an AWS IoT SiteWise Gateway.
:param gateway_name: The name of the gateway to create.
:param my_thing: The core device thing name.
:return: The ID of the created gateway.
"""
try:
response = self.iotsitewise_client.create_gateway(
gatewayName=gateway_name,
gatewayPlatform={
"greengrassV2": {"coreDeviceThingName": my_thing},
},
tags={"Environment": "Production"},
)
gateway_id = response["gatewayId"]
return gateway_id
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceAlreadyExistsException":
logger.error("Gateway %s already exists.", gateway_name)
else:
logger.error(
"Error creating gateway %s. Here's why %s",
gateway_name,
err.response["Error"]["Message"],
)
raise
def describe_gateway(self, gateway_id: str) -> Dict[str, Any]:
"""
Describes an AWS IoT SiteWise Gateway.
:param gateway_id: The ID of the gateway to describe.
:return: A dictionary containing information about the gateway.
"""
try:
response = self.iotsitewise_client.describe_gateway(gatewayId=gateway_id)
return response
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error("Gateway %s does not exist.", gateway_id)
else:
logger.error(
"Error describing gateway %s. Here's why %s",
gateway_id,
err.response["Error"]["Message"],
)
raise
def delete_gateway(self, gateway_id: str) -> None:
"""
Deletes an AWS IoT SiteWise Gateway.
:param gateway_id: The ID of the gateway to delete.
"""
try:
self.iotsitewise_client.delete_gateway(gatewayId=gateway_id)
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error("Gateway %s does not exist.", gateway_id)
else:
logger.error(
"Error deleting gateway %s. Here's why %s",
gateway_id,
err.response["Error"]["Message"],
)
raise
def delete_portal(self, portal_id: str) -> None:
"""
Deletes an AWS IoT SiteWise Portal.
:param portal_id: The ID of the portal to delete.
"""
try:
self.iotsitewise_client.delete_portal(portalId=portal_id)
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error("Portal %s does not exist.", portal_id)
else:
logger.error(
"Error deleting portal %s. Here's why %s",
portal_id,
err.response["Error"]["Message"],
)
raise
def delete_asset(self, asset_id: str) -> None:
"""
Deletes an AWS IoT SiteWise Asset.
:param asset_id: The ID of the asset to delete.
"""
try:
self.iotsitewise_client.delete_asset(assetId=asset_id)
except ClientError as err:
logger.error(
"Error deleting asset %s. Here's why %s",
asset_id,
err.response["Error"]["Message"],
)
raise
def delete_asset_model(self, asset_model_id: str) -> None:
"""
Deletes an AWS IoT SiteWise Asset Model.
:param asset_model_id: The ID of the asset model to delete.
"""
try:
self.iotsitewise_client.delete_asset_model(assetModelId=asset_model_id)
except ClientError as err:
logger.error(
"Error deleting asset model %s. Here's why %s",
asset_model_id,
err.response["Error"]["Message"],
)
raise
def wait_asset_deleted(self, asset_id: str) -> None:
"""
Waits for an AWS IoT SiteWise Asset to be deleted.
:param asset_id: The ID of the asset to wait for.
"""
try:
waiter = self.iotsitewise_client.get_waiter("asset_not_exists")
waiter.wait(assetId=asset_id)
except ClientError as err:
logger.error(
"Error waiting for asset %s to be deleted. Here's why %s",
asset_id,
err.response["Error"]["Message"],
)
raise
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 다음 주제를 참조하세요.
+ [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/BatchPutAssetPropertyValue)
+ [CreateAsset](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/CreateAsset)
+ [CreateAssetModel](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/CreateAssetModel)
+ [CreateGateway](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/CreateGateway)
+ [DeleteAsset](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/DeleteAsset)
+ [DeleteAssetModel](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/DeleteAssetModel)
+ [DeleteGateway](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/DeleteGateway)
+ [DescribeAssetModel](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/DescribeAssetModel)
+ [DescribeGateway](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/DescribeGateway)
+ [GetAssetPropertyValue](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/GetAssetPropertyValue)
+ [ListAssetModelProperties](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/ListAssetModelProperties)
+ [ListAssetModels](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/ListAssetModels)
## 작업
### `BatchPutAssetPropertyValue`
다음 코드 예시는 `BatchPutAssetPropertyValue`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def batch_put_asset_property_value(
self, asset_id: str, values: List[Dict[str, str]]
) -> None:
"""
Sends data to an AWS IoT SiteWise Asset.
:param asset_id: The asset ID.
:param values: A list of dictionaries containing the values in the form
{propertyId : property_id,
valueType : [stringValue|integerValue|doubleValue|booleanValue],
value : the_value}.
"""
try:
entries = self.properties_to_values(asset_id, values)
self.iotsitewise_client.batch_put_asset_property_value(entries=entries)
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error("Asset %s does not exist.", asset_id)
else:
logger.error(
"Error sending data to asset. Here's why %s",
err.response["Error"]["Message"],
)
raise
```
값 목록에서 항목 파라미터를 생성하는 헬퍼 함수입니다.
```
def properties_to_values(
self, asset_id: str, values: list[dict[str, Any]]
) -> list[dict[str, Any]]:
"""
Utility function to convert a values list to the entries parameter for batch_put_asset_property_value.
:param asset_id : The asset ID.
:param values : A list of dictionaries containing the values in the form
{propertyId : property_id,
valueType : [stringValue|integerValue|doubleValue|booleanValue],
value : the_value}.
:return: An entries list to pass as the 'entries' parameter to batch_put_asset_property_value.
"""
entries = []
for value in values:
epoch_ns = time.time_ns()
self.entry_id += 1
if value["valueType"] == "stringValue":
property_value = {"stringValue": value["value"]}
elif value["valueType"] == "integerValue":
property_value = {"integerValue": value["value"]}
elif value["valueType"] == "booleanValue":
property_value = {"booleanValue": value["value"]}
elif value["valueType"] == "doubleValue":
property_value = {"doubleValue": value["value"]}
else:
raise ValueError("Invalid valueType: %s", value["valueType"])
entry = {
"entryId": f"{self.entry_id}",
"assetId": asset_id,
"propertyId": value["propertyId"],
"propertyValues": [
{
"value": property_value,
"timestamp": {
"timeInSeconds": int(epoch_ns / 1000000000),
"offsetInNanos": epoch_ns % 1000000000,
},
}
],
}
entries.append(entry)
return entries
```
다음은 헬퍼 함수에 전달할 값 목록의 예입니다.
```
values = [
{
"propertyId": humidity_property_id,
"valueType": "doubleValue",
"value": 65.0,
},
{
"propertyId": temperature_property_id,
"valueType": "doubleValue",
"value": 23.5,
},
]
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/BatchPutAssetPropertyValue)를 참조하세요.
### `CreateAsset`
다음 코드 예시는 `CreateAsset`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def create_asset(self, asset_name: str, asset_model_id: str) -> str:
"""
Creates an AWS IoT SiteWise Asset.
:param asset_name: The name of the asset to create.
:param asset_model_id: The ID of the asset model to associate with the asset.
:return: The ID of the created asset.
"""
try:
response = self.iotsitewise_client.create_asset(
assetName=asset_name, assetModelId=asset_model_id
)
asset_id = response["assetId"]
waiter = self.iotsitewise_client.get_waiter("asset_active")
waiter.wait(assetId=asset_id)
return asset_id
except ClientError as err:
if err.response["Error"] == "ResourceNotFoundException":
logger.error("Asset model %s does not exist.", asset_model_id)
else:
logger.error(
"Error creating asset %s. Here's why %s",
asset_name,
err.response["Error"]["Message"],
)
raise
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [CreateAsset](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/CreateAsset)을 참조하세요.
### `CreateAssetModel`
다음 코드 예시는 `CreateAssetModel`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def create_asset_model(
self, asset_model_name: str, properties: List[Dict[str, Any]]
) -> str:
"""
Creates an AWS IoT SiteWise Asset Model.
:param asset_model_name: The name of the asset model to create.
:param properties: The property definitions of the asset model.
:return: The ID of the created asset model.
"""
try:
response = self.iotsitewise_client.create_asset_model(
assetModelName=asset_model_name,
assetModelDescription="This is a sample asset model description.",
assetModelProperties=properties,
)
asset_model_id = response["assetModelId"]
waiter = self.iotsitewise_client.get_waiter("asset_model_active")
waiter.wait(assetModelId=asset_model_id)
return asset_model_id
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceAlreadyExistsException":
logger.error("Asset model %s already exists.", asset_model_name)
else:
logger.error(
"Error creating asset model %s. Here's why %s",
asset_model_name,
err.response["Error"]["Message"],
)
raise
```
다음은 함수에 전달할 속성 목록의 예입니다.
```
properties = [
{
"name": temperature_property_name,
"dataType": "DOUBLE",
"type": {
"measurement": {},
},
},
{
"name": humidity_property_name,
"dataType": "DOUBLE",
"type": {
"measurement": {},
},
},
]
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [CreateAssetModel](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/CreateAssetModel)을 참조하세요.
### `CreateGateway`
다음 코드 예시는 `CreateGateway`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def create_gateway(self, gateway_name: str, my_thing: str) -> str:
"""
Creates an AWS IoT SiteWise Gateway.
:param gateway_name: The name of the gateway to create.
:param my_thing: The core device thing name.
:return: The ID of the created gateway.
"""
try:
response = self.iotsitewise_client.create_gateway(
gatewayName=gateway_name,
gatewayPlatform={
"greengrassV2": {"coreDeviceThingName": my_thing},
},
tags={"Environment": "Production"},
)
gateway_id = response["gatewayId"]
return gateway_id
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceAlreadyExistsException":
logger.error("Gateway %s already exists.", gateway_name)
else:
logger.error(
"Error creating gateway %s. Here's why %s",
gateway_name,
err.response["Error"]["Message"],
)
raise
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [CreateGateway](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/CreateGateway)를 참조하세요.
### `DeleteAsset`
다음 코드 예시는 `DeleteAsset`의 사용 방법을 보여 줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def delete_asset(self, asset_id: str) -> None:
"""
Deletes an AWS IoT SiteWise Asset.
:param asset_id: The ID of the asset to delete.
"""
try:
self.iotsitewise_client.delete_asset(assetId=asset_id)
except ClientError as err:
logger.error(
"Error deleting asset %s. Here's why %s",
asset_id,
err.response["Error"]["Message"],
)
raise
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [DeleteAsset](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/DeleteAsset)을 참조하세요.
### `DeleteAssetModel`
다음 코드 예시는 `DeleteAssetModel`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def delete_asset_model(self, asset_model_id: str) -> None:
"""
Deletes an AWS IoT SiteWise Asset Model.
:param asset_model_id: The ID of the asset model to delete.
"""
try:
self.iotsitewise_client.delete_asset_model(assetModelId=asset_model_id)
except ClientError as err:
logger.error(
"Error deleting asset model %s. Here's why %s",
asset_model_id,
err.response["Error"]["Message"],
)
raise
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [DeleteAssetModel](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/DeleteAssetModel)을 참조하세요.
### `DeleteGateway`
다음 코드 예시는 `DeleteGateway`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def delete_gateway(self, gateway_id: str) -> None:
"""
Deletes an AWS IoT SiteWise Gateway.
:param gateway_id: The ID of the gateway to delete.
"""
try:
self.iotsitewise_client.delete_gateway(gatewayId=gateway_id)
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error("Gateway %s does not exist.", gateway_id)
else:
logger.error(
"Error deleting gateway %s. Here's why %s",
gateway_id,
err.response["Error"]["Message"],
)
raise
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [DeleteGateway](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/DeleteGateway)를 참조하세요.
### `DescribeGateway`
다음 코드 예시는 `DescribeGateway`의 사용 방법을 보여 줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def describe_gateway(self, gateway_id: str) -> Dict[str, Any]:
"""
Describes an AWS IoT SiteWise Gateway.
:param gateway_id: The ID of the gateway to describe.
:return: A dictionary containing information about the gateway.
"""
try:
response = self.iotsitewise_client.describe_gateway(gatewayId=gateway_id)
return response
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error("Gateway %s does not exist.", gateway_id)
else:
logger.error(
"Error describing gateway %s. Here's why %s",
gateway_id,
err.response["Error"]["Message"],
)
raise
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [DescribeGateway](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/DescribeGateway)를 참조하세요.
### `GetAssetPropertyValue`
다음 코드 예시는 `GetAssetPropertyValue`의 사용 방법을 보여 줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def get_asset_property_value(
self, asset_id: str, property_id: str
) -> Dict[str, Any]:
"""
Gets the value of an AWS IoT SiteWise Asset Property.
:param asset_id: The ID of the asset.
:param property_id: The ID of the property.
:return: A dictionary containing the value of the property.
"""
try:
response = self.iotsitewise_client.get_asset_property_value(
assetId=asset_id, propertyId=property_id
)
return response["propertyValue"]
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
logger.error(
"Asset %s or property %s does not exist.", asset_id, property_id
)
else:
logger.error(
"Error getting asset property value. Here's why %s",
err.response["Error"]["Message"],
)
raise
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [GetAssetPropertyValue](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/GetAssetPropertyValue)를 참조하세요.
### `ListAssetModels`
다음 코드 예시는 `ListAssetModels`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iotsitewise#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
class IoTSitewiseWrapper:
"""Encapsulates AWS IoT SiteWise actions using the client interface."""
def __init__(self, iotsitewise_client: client) -> None:
"""
Initializes the IoTSitewiseWrapper with an AWS IoT SiteWise client.
:param iotsitewise_client: A Boto3 AWS IoT SiteWise client. This client provides low-level
access to AWS IoT SiteWise services.
"""
self.iotsitewise_client = iotsitewise_client
self.entry_id = 0 # Incremented to generate unique entry IDs for batch_put_asset_property_value.
@classmethod
def from_client(cls) -> "IoTSitewiseWrapper":
"""
Creates an IoTSitewiseWrapper instance with a default AWS IoT SiteWise client.
:return: An instance of IoTSitewiseWrapper initialized with the default AWS IoT SiteWise client.
"""
iotsitewise_client = boto3.client("iotsitewise")
return cls(iotsitewise_client)
def list_asset_models(self) -> List[Dict[str, Any]]:
"""
Lists all AWS IoT SiteWise Asset Models.
:return: A list of dictionaries containing information about each asset model.
"""
try:
asset_models = []
paginator = self.iotsitewise_client.get_paginator("list_asset_models")
pages = paginator.paginate()
for page in pages:
asset_models.extend(page["assetModelSummaries"])
return asset_models
except ClientError as err:
logger.error(
"Error listing asset models. Here's why %s",
err.response["Error"]["Message"],
)
raise
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [ListAssetModels](https://docs.aws.amazon.com/goto/boto3/iotsitewise-2019-12-02/ListAssetModels)을 참조하세요.