/  Python Yamcs Client  /  Examples  /  archive_retrieval.py

archive_retrieval.pyΒΆ

from datetime import datetime, timedelta, timezone
from itertools import islice

from yamcs.client import YamcsClient


def print_last_packets():
    """Print the last 10 packets."""
    for packet in islice(archive.list_packets(descending=True), 0, 10):
        print(packet)


def print_packet_range():
    """Print the 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)


def iterate_specific_packet_range():
    """Count the number of packets in a specific range."""
    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")


def export_raw_packets():
    """Export raw packet binary."""
    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)


def iterate_specific_event_range():
    """Count the number of events in a specific range."""
    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")


def print_last_values():
    """Print the last 10 values."""
    iterable = archive.list_parameter_values(
        "/YSS/SIMULATOR/BatteryVoltage1", descending=True
    )
    for pval in islice(iterable, 0, 10):
        print(pval)


def iterate_specific_parameter_range():
    """Count the number of parameter values in a specific range."""
    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")


def stream_specific_parameter_range():
    """Stream one or more parameters in a specific range."""
    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")


def print_last_commands():
    """Print the last 10 commands."""
    iterable = archive.list_command_history(descending=True)
    for entry in islice(iterable, 0, 10):
        print(entry)


if __name__ == "__main__":
    client = YamcsClient("localhost:8090")
    archive = client.get_archive(instance="simulator")

    print("Last 10 packets:")
    print_last_packets()

    print("\nPacket range:")
    print_packet_range()

    print("\nIterate specific packet range:")
    iterate_specific_packet_range()

    print("\nIterate specific event range:")
    iterate_specific_event_range()

    print("\nLast 10 parameter values:")
    print_last_values()

    print("\nIterate specific parameter range:")
    iterate_specific_parameter_range()

    # print("\nStream specific parameter range:")
    # stream_specific_parameter_range()

    print("\nLast 10 commands:")
    print_last_commands()