Archive¶

The Archive API provides methods that you can use to programmatically retrieve the content of a Yamcs Archive.

Reference¶

Snippets¶

Create an ArchiveClient for a specific instance:

from yamcs.client import YamcsClient

client = YamcsClient('localhost:8090')
archive = client.get_archive(instance='simulator')

Packet Retrieval¶

Print the last 10 packets:

for packet in islice(archive.list_packets(descending=True), 0, 10):
    print(packet)

Print available range of archived packets:

first_packet = next(iter(archive.list_packets()))
last_packet = next(iter(archive.list_packets(descending=True)))
print("First packet:", first_packet)
print("Last packet:", last_packet)

td = last_packet.generation_time - first_packet.generation_time
print("Timespan:", td)

Iterate a specific range of packets:

now = datetime.now(tz=timezone.utc)
start = now - timedelta(hours=1)

total = 0
for packet in archive.list_packets(start=start, stop=now):
    total += 1
    # print(packet)
print("Found", total, "packets in range")

Download raw packet binary to a file:

now = datetime.now(tz=timezone.utc)
start = now - timedelta(hours=1)

with open("/tmp/dump.raw", "wb") as f:
    for chunk in archive.export_packets(start=start, stop=now):
        f.write(chunk)

Parameter Retrieval¶

Retrieve the last 10 values of a parameter:

iterable = archive.list_parameter_values(
    "/YSS/SIMULATOR/BatteryVoltage1", descending=True
)
for pval in islice(iterable, 0, 10):
    print(pval)

Iterate a specific range of values:

now = datetime.now(tz=timezone.utc)
start = now - timedelta(hours=1)

total = 0
for pval in archive.list_parameter_values(
    "/YSS/SIMULATOR/BatteryVoltage1", start=start, stop=now
):
    total += 1
    # print(pval)
print("Found", total, "parameter values in range")

Iterate values of multiple parameters by unique generation time:

now = datetime.now(tz=timezone.utc)
start = now - timedelta(hours=1)

total = 0
for pdata in archive.stream_parameter_values(
    ["/YSS/SIMULATOR/BatteryVoltage1", "/YSS/SIMULATOR/BatteryVoltage2"],
    start=start,
    stop=now,
):
    total += 1
    # print(pdata)
print("Found", total, "updates in range")

Event Retrieval¶

Iterate a specific range of events:

now = datetime.now(tz=timezone.utc)
start = now - timedelta(hours=1)

total = 0
for event in archive.list_events(start=start, stop=now):
    total += 1
    # print(event)
print("Found", total, "events in range")

Command Retrieval¶

Retrieve the last 10 issued commands:

iterable = archive.list_command_history(descending=True)
for entry in islice(iterable, 0, 10):
    print(entry)

Histogram Retrieval¶

Print the number of packets grouped by packet name:

for name in archive.list_packet_names():
    packet_count = 0
    for group in archive.list_packet_histogram(name):
        for rec in group.records:
            packet_count += rec.count
    print(f"  {name: <40} {packet_count: >20}")

Print the number of events grouped by source:

for source in archive.list_event_sources():
    event_count = 0
    for group in archive.list_event_histogram(source):
        for rec in group.records:
            event_count += rec.count
    print(f"  {source: <40} {event_count: >20}")

Print the number of processed parameter frames grouped by group name:

for group in archive.list_processed_parameter_groups():
    frame_count = 0
    for pp_group in archive.list_processed_parameter_group_histogram(group):
        for rec in pp_group.records:
            frame_count += rec.count
    print(f"  {group: <40} {frame_count: >20}")

Print the number of commands grouped by name:

mdb = client.get_mdb(instance="simulator")
for command in mdb.list_commands():
    total = 0
    for group in archive.list_command_histogram(command.qualified_name):
        for rec in group.records:
            total += rec.count
    print(f"  {command.qualified_name: <40} {total: >20}")