File Transfer¶

The File Transfer API provides methods that you can use to programmatically work with file transfers such as CFDP.

Reference¶

Client
Model

Snippets¶

Create a FileTransferClient for a specific instance:

from yamcs.client import YamcsClient

client = YamcsClient('localhost:8090')
cfdp = client.get_file_transfer_client(instance='cfdp')

# Operations are grouped by service.
# Here: take the first available
service = next(cfdp.list_services())

Upload a file to the specified location on the remote entity:

# Transfer myfile from bucket to spacecraft
upload = service.upload(out_bucket.name, "myfile", "/CF:/mytarget")
upload.await_complete(timeout=10)

if not upload.is_success():
    print("Upload failure:", upload.error)
else:
    print(f"Successfully uploaded {upload.remote_path} ({upload.size} bytes)")

Start file download:

# Download todownload from the remote
download = service.download(in_bucket.name, "todownload")

Initiate transfer with non-default parameters:

# Transfer myfile, but use an alternative destination entity
upload = service.upload(
    out_bucket.name, "myfile", "/CF:/mytarget", destination_entity="target2"
)

Initiate transfer with extra transfer options:

upload = service.upload(
    out_bucket.name, "myfile", "/CF:/mytarget", options={"reliable": True}
)

Subscribe to file list changes:

subscription = service.create_filelist_subscription(on_data=filelist_callback)

Fetch file list from remote directory:

filelist_request = service.fetch_filelist("/")

Get the latest saved remote file list for the given directory, and display it:

filelist = service.get_filelist("/")

# Display file list
if filelist:
    print("File list received:")
    if not filelist.files:
        print("\tEmpty file list")
    for file in filelist.files:
        print(f"\t{file.name + ('/' if file.is_directory else ''):<12}\
        t{str(file.size) + ' bytes':>12}\tLast Modified: {file.modified}")
else:
    print("No filelist found")