使用自定义星历保留联系人 - AWS Ground Station
概述联系人预订工作流程工作流程 1：列出可用的联系人然后预订工作流程 2：直接联系预约监控联系人状态的变化最佳实践和注意事项

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

使用自定义星历保留联系人

概述

使用自定义星历（TLE、OEM 或方位角仰角）时，您可以使用 API 保留联系人。ReserveContact本节介绍两种用于预留联系人的常见工作流程以及确保成功安排联系人的重要注意事项。

AWS Ground Station 天线是多个客户共享的资源。这意味着，即使在您列出联系人时显示联系人窗口，其他客户也可能会在您之前预订。因此，至关重要的是要验证您的联系人在预订后是否到达该SCHEDULED州，并对接触状态的变化进行适当的监控。

重要

对于方位角高程星历，可以从ReserveContact请求中省略该satelliteArn参数，并且您必须提供星历表 ID。trackingOverrides对于 TLE 和 OEM 星历表，您仍然需要提供. satelliteArn

联系人预订工作流程

使用自定义星历保留联系人的主要工作流程有两个：

  1. List-then-reserve 工作流程：首先列出可用的联系人窗口 ListContacts，然后选择并保留特定的窗口。当您想在做出选择之前查看所有可用的机会时，这种方法非常有用。

  2. 直接预订工作流程：直接在特定时间段内保留联系人，无需先列出可用的联系人。当您已经知道所需的联系时间或正在使用预先确定的时间表时，这种方法非常有用。

两种工作流程均有效，具体选择取决于您的操作要求。以下各节提供了每种方法的示例。

工作流程 1：列出可用的联系人然后预订

此工作流程首先查询可用的联系人窗口，然后保留一个特定的窗口。当你想在做出选择之前查看所有可用的机会时，这很有用。

示例：使用方位角高程星历表列出并保留

import boto3
from datetime import datetime, timezone
import time

# Create AWS Ground Station client
ground_station_client = boto3.client("groundstation")

# Create azimuth elevation ephemeris
print("Creating azimuth elevation ephemeris...")
ephemeris_response = ground_station_client.create_ephemeris(
    name="AzEl Ephemeris for Contact",
    ephemeris={
        "azEl": {
            "groundStation": "Ohio 1",
            "data": {
                "azElData": {
                    "angleUnit": "DEGREE_ANGLE",
                    "azElSegmentList": [
                        {
                            "referenceEpoch": "2024-03-15T10:00:00Z",
                            "validTimeRange": {
                                "startTime": "2024-03-15T10:00:00Z",
                                "endTime": "2024-03-15T10:15:00Z",
                            },
                            "azElList": [
                                {"dt": 0.0, "az": 45.0, "el": 10.0},
                                {"dt": 180.0, "az": 50.0, "el": 15.0},
                                {"dt": 360.0, "az": 55.0, "el": 20.0},
                                {"dt": 540.0, "az": 60.0, "el": 25.0},
                                {"dt": 720.0, "az": 65.0, "el": 30.0},
                                {"dt": 900.0, "az": 70.0, "el": 35.0},
                            ],
                        }
                    ],
                }
            },
        }
    },
)

ephemeris_id = ephemeris_response["ephemerisId"]
print(f"Created ephemeris: {ephemeris_id}")

# Wait for ephemeris to become ENABLED
while True:
    status = ground_station_client.describe_ephemeris(ephemerisId=ephemeris_id)[
        "status"
    ]
    if status == "ENABLED":
        print("Ephemeris is ENABLED")
        break
    elif status in ["INVALID", "ERROR"]:
        raise RuntimeError(f"Ephemeris failed: {status}")
    time.sleep(5)

# List available contacts
print("Listing available contacts...")
contacts = ground_station_client.list_contacts(
    # Note: satelliteArn is omitted for azimuth elevation ephemeris
    groundStation="Ohio 1",
    missionProfileArn="arn:aws:groundstation:us-east-2:111122223333:mission-profile/example-profile",
    startTime=datetime(2024, 3, 15, 10, 0, 0, tzinfo=timezone.utc),
    endTime=datetime(2024, 3, 15, 10, 15, 0, tzinfo=timezone.utc),
    statusList=["AVAILABLE"],
    ephemeris={"azEl": {"id": ephemeris_id}},
)

if contacts["contactList"]:
    # Reserve the first available contact
    contact = contacts["contactList"][0]
    print(f"Reserving contact from {contact['startTime']} to {contact['endTime']}...")

    reservation = ground_station_client.reserve_contact(
        # Note: satelliteArn is omitted when using azimuth elevation ephemeris
        missionProfileArn="arn:aws:groundstation:us-east-2:111122223333:mission-profile/example-profile",
        groundStation="Ohio 1",
        startTime=contact["startTime"],
        endTime=contact["endTime"],
        trackingOverrides={
            "programTrackSettings": {"azEl": {"ephemerisId": ephemeris_id}}
        },
    )

    print(f"Reserved contact: {reservation['contactId']}")
else:
    print("No available contacts found")

示例：使用 TLE 星历表列出并预订

import boto3
from datetime import datetime, timedelta, timezone
import time

# Create AWS Ground Station client
ground_station_client = boto3.client("groundstation")

satellite_id = "12345678-1234-1234-1234-123456789012"
satellite_arn = f"arn:aws:groundstation::111122223333:satellite/{satellite_id}"

# Create TLE ephemeris
print("Creating TLE ephemeris...")
ephemeris_response = ground_station_client.create_ephemeris(
    name="TLE Ephemeris for Contact",
    satelliteId=satellite_id,
    enabled=True,
    expirationTime=datetime.now(timezone.utc) + timedelta(days=7),
    priority=1,  # Higher priority than default ephemeris
    ephemeris={
        "tle": {
            "tleData": [
                {
                    "tleLine1": "1 25994U 99068A   24075.54719794  .00000075  00000-0  26688-4 0  9997",
                    "tleLine2": "2 25994  98.2007  30.6589 0001234  89.2782  18.9934 14.57114995111906",
                    "validTimeRange": {
                        "startTime": datetime.now(timezone.utc),
                        "endTime": datetime.now(timezone.utc) + timedelta(days=7),
                    },
                }
            ]
        }
    },
)

ephemeris_id = ephemeris_response["ephemerisId"]
print(f"Created ephemeris: {ephemeris_id}")

# Wait for ephemeris to become ENABLED
while True:
    status = ground_station_client.describe_ephemeris(ephemerisId=ephemeris_id)[
        "status"
    ]
    if status == "ENABLED":
        print("Ephemeris is ENABLED")
        break
    elif status in ["INVALID", "ERROR"]:
        raise RuntimeError(f"Ephemeris failed: {status}")
    time.sleep(5)

# List available contacts
print("Listing available contacts...")
start_time = datetime.now(timezone.utc) + timedelta(hours=1)
end_time = start_time + timedelta(days=1)

contacts = ground_station_client.list_contacts(
    satelliteArn=satellite_arn,  # Required for TLE/OEM ephemeris
    groundStation="Hawaii 1",
    missionProfileArn="arn:aws:groundstation:us-west-2:111122223333:mission-profile/example-profile",
    startTime=start_time,
    endTime=end_time,
    statusList=["AVAILABLE"],
)

if contacts["contactList"]:
    # Reserve the first available contact
    contact = contacts["contactList"][0]
    print(f"Reserving contact from {contact['startTime']} to {contact['endTime']}...")

    reservation = ground_station_client.reserve_contact(
        satelliteArn=satellite_arn,  # Required for TLE/OEM ephemeris
        missionProfileArn="arn:aws:groundstation:us-west-2:111122223333:mission-profile/example-profile",
        groundStation="Hawaii 1",
        startTime=contact["startTime"],
        endTime=contact["endTime"],
        # Note: trackingOverrides is optional for TLE/OEM
        # The system will use the highest priority ephemeris automatically
    )

    print(f"Reserved contact: {reservation['contactId']}")
else:
    print("No available contacts found")

工作流程 2：直接联系预约

此工作流程直接保留联系人，无需先列出可用窗口。当您已经知道所需的联系时间或正在实施自动安排时，这种方法非常有用。

示例：使用方位角高程星历表直接预留

import boto3
from datetime import datetime, timezone
import time

# Create AWS Ground Station client
ground_station_client = boto3.client("groundstation")

# Define contact window
contact_start = datetime(2024, 3, 20, 14, 0, 0, tzinfo=timezone.utc)
contact_end = datetime(2024, 3, 20, 14, 15, 0, tzinfo=timezone.utc)

# Create azimuth elevation ephemeris for the specific contact time
print("Creating azimuth elevation ephemeris...")
ephemeris_response = ground_station_client.create_ephemeris(
    name="Direct Contact AzEl Ephemeris",
    ephemeris={
        "azEl": {
            "groundStation": "Ohio 1",
            "data": {
                "azElData": {
                    "angleUnit": "DEGREE_ANGLE",
                    "azElSegmentList": [
                        {
                            "referenceEpoch": contact_start.isoformat(),
                            "validTimeRange": {
                                "startTime": contact_start.isoformat(),
                                "endTime": contact_end.isoformat(),
                            },
                            "azElList": [
                                {"dt": 0.0, "az": 45.0, "el": 10.0},
                                {"dt": 180.0, "az": 50.0, "el": 15.0},
                                {"dt": 360.0, "az": 55.0, "el": 20.0},
                                {"dt": 540.0, "az": 60.0, "el": 25.0},
                                {"dt": 720.0, "az": 65.0, "el": 30.0},
                                {"dt": 900.0, "az": 70.0, "el": 35.0},
                            ],
                        }
                    ],
                }
            },
        }
    },
)

ephemeris_id = ephemeris_response["ephemerisId"]
print(f"Created ephemeris: {ephemeris_id}")

# Wait for ephemeris to become ENABLED
while True:
    status = ground_station_client.describe_ephemeris(ephemerisId=ephemeris_id)[
        "status"
    ]
    if status == "ENABLED":
        print("Ephemeris is ENABLED")
        break
    elif status in ["INVALID", "ERROR"]:
        raise RuntimeError(f"Ephemeris failed: {status}")
    time.sleep(5)

# Directly reserve the contact
print(f"Reserving contact from {contact_start} to {contact_end}...")

reservation = ground_station_client.reserve_contact(
    # Note: satelliteArn is omitted for azimuth elevation
    missionProfileArn="arn:aws:groundstation:us-east-2:111122223333:mission-profile/example-profile",
    groundStation="Ohio 1",
    startTime=contact_start,
    endTime=contact_end,
    trackingOverrides={"programTrackSettings": {"azEl": {"ephemerisId": ephemeris_id}}},
)

print(f"Reserved contact: {reservation['contactId']}")

示例：使用 TLE 星历表直接预订

import boto3
from datetime import datetime, timedelta, timezone
import time

# Create AWS Ground Station client
ground_station_client = boto3.client("groundstation")

satellite_id = "12345678-1234-1234-1234-123456789012"
satellite_arn = f"arn:aws:groundstation::111122223333:satellite/{satellite_id}"

# Define contact window (based on predicted pass)
contact_start = datetime(2024, 3, 21, 10, 30, 0, tzinfo=timezone.utc)
contact_end = datetime(2024, 3, 21, 10, 42, 0, tzinfo=timezone.utc)

# Create TLE ephemeris
print("Creating TLE ephemeris...")
ephemeris_response = ground_station_client.create_ephemeris(
    name="Direct Contact TLE Ephemeris",
    satelliteId=satellite_id,
    enabled=True,
    expirationTime=contact_end + timedelta(days=1),
    priority=1,
    ephemeris={
        "tle": {
            "tleData": [
                {
                    "tleLine1": "1 25994U 99068A   24080.50000000  .00000075  00000-0  26688-4 0  9999",
                    "tleLine2": "2 25994  98.2007  35.6589 0001234  89.2782  18.9934 14.57114995112000",
                    "validTimeRange": {
                        "startTime": (contact_start - timedelta(hours=1)).isoformat(),
                        "endTime": (contact_end + timedelta(hours=1)).isoformat(),
                    },
                }
            ]
        }
    },
)

ephemeris_id = ephemeris_response["ephemerisId"]
print(f"Created ephemeris: {ephemeris_id}")

# Wait for ephemeris to become ENABLED
while True:
    status = ground_station_client.describe_ephemeris(ephemerisId=ephemeris_id)[
        "status"
    ]
    if status == "ENABLED":
        print("Ephemeris is ENABLED")
        break
    elif status in ["INVALID", "ERROR"]:
        raise RuntimeError(f"Ephemeris failed: {status}")
    time.sleep(5)

# Directly reserve the contact
print(f"Reserving contact from {contact_start} to {contact_end}...")

reservation = ground_station_client.reserve_contact(
    satelliteArn=satellite_arn,  # Required for TLE ephemeris
    missionProfileArn="arn:aws:groundstation:us-west-2:111122223333:mission-profile/example-profile",
    groundStation="Hawaii 1",
    startTime=contact_start,
    endTime=contact_end,
    # Note: trackingOverrides is optional for TLE
    # The system will use the highest priority ephemeris automatically
)

print(f"Reserved contact: {reservation['contactId']}")

监控联系人状态的变化

预约联系人后，重要的是要监控其状态，以确保其成功过渡到任何问题SCHEDULED并收到通知。 AWS Ground Station EventBridge 针对所有联系状态变更向 Amazon 发送事件。

联系人状态遵循以下生命周期：

  • SCHEDULING-正在处理联系人以进行日程安排

  • SCHEDULED-已成功安排联系人并将执行

  • FAILED_TO_SCHEDULE-无法安排联系人（终端状态）

有关联系人状态和生命周期的更多信息，请参阅了解联系人生命周期。

使用实现接触状态监控 EventBridge

要实时监控联系人状态的变化，您可以设置一 EventBridge 条 Amazon 规则，该规则在 Ground Station 联系人更改状态时触发 Lambda 函数。这种方法比轮询联系人状态更有效且更具可扩展性。

实现步骤

  1. 创建 Lambda 函数来处理联系人状态变更事件

  2. 创建与 Ground Station 联系状态变更事件相匹配的 EventBridge 规则

  3. 添加 Lambda 函数作为规则的目标

Lambda 函数处理程序示例

有关处理联系人状态变更事件的 Lambda 函数的完整示例，请参阅模板中的GroundStationCloudWatchEventHandlerLambdaAquaSnppJpssTerraDigIF.yml CloudFormation 资源。此模板可在 AWS Ground Station 客户入门 Amazon S3 存储桶中找到。有关访问此模板的说明，请参阅 dataflow 端点示例把它放在一起部分。

EventBridge 规则配置

该 EventBridge 规则应使用以下事件模式来匹配所有 Ground Station 联系状态的变化：

{
  "source": ["aws.groundstation"],
  "detail-type": ["Ground Station Contact State Change"]
}

要仅筛选特定状态（例如故障），可以添加详细信息筛选器：

{
  "source": ["aws.groundstation"],
  "detail-type": ["Ground Station Contact State Change"],
  "detail": {
    "contactStatus": [
      "FAILED_TO_SCHEDULE",
      "FAILED",
      "AWS_FAILED",
      "AWS_CANCELLED"
    ]
  }
}

有关使用 Lambda 目标创建 EventBridge 规则的详细说明，请参阅 Amazon EventBridge 用户指南中的创建对事件做出反应的规则。

设置自动化 EventBridge 规则

您可以创建 EventBridge 规则以自动响应联系人状态的变化。例如：

  • 联系人无法安排时发送通知

  • 当联系人进入时触发 Lambda 函数以准备资源 PREPASS

  • 记录联系完成情况以供审计

有关设置 AWS Ground Station 事件 EventBridge 规则的详细信息，请参阅利用事件 AWS Ground Station 实现自动化。

最佳实践和注意事项

处理日程安排冲突

AWS Ground Station 由于天线是共享资源，因此在中显示的联系窗口ListContacts可能会被其他客户预留，然后您才能预订。要处理这个问题：

  1. 预订后请务必查看联系状态

  2. 使用备用时间窗口实现重试逻辑

  3. 如有可能，请考虑尽早预约联系人

  4. 使用 EventBridge 事件来监控FAILED_TO_SCHEDULE状态

星历验证时机

请记住，星历必须处于ENABLED状态，然后才能使用它来保留联系人。验证过程通常需要几秒钟到几分钟，具体取决于星历的类型和大小。在尝试保留联系人之前，请务必验证星历状态。

联系时机注意事项

使用自定义星历时：

按星历类型划分的 API 差异

根据星历类型，ReserveContactAPI 的行为会有所不同：

星历类型 需要卫星收入 需要追踪覆盖
TLE 是 否（可选）
OEM 是 否（可选）
方位角高程 否（可选） 是