/  Python API  /  Archive


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



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 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.utcnow()
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.utcnow()
start = now - timedelta(hours=1)

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

Parameter Retrieval

Retrieve the last 10 values of a parameter:

iterable = archive.list_parameter_values('/YSS/SIMULATOR/BatteryVoltage1',
for pval in islice(iterable, 0, 10):

Iterate a specific range of values:

now = datetime.utcnow()
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')

Event Retrieval

Iterate a specific range of events:

now = datetime.utcnow()
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):

Histogram Retrieval

Print the number of packets grouped by packet name:

for name in archive.list_packet_names():
    if not name == '/YSS/SIMULATOR/DHS':
    packet_count = 0
    for group in archive.list_packet_histogram(name):
        for rec in group.records:
            packet_count += rec.count
    print('  {: <40} {: >20}'.format(name, packet_count))

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('  {: <40} {: >20}'.format(source, event_count))

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('  {: <40} {: >20}'.format(group, frame_count))

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('  {: <40} {: >20}'.format(command.qualified_name, total))