eive-tmtc/eive_tmtc/cfdp/user.py
Robin Mueller db70b33bd6
All checks were successful
EIVE/-/pipeline/pr-v6.0.0-dev This commit looks good
bump tmtccmd again
2024-01-24 17:59:16 +01:00

78 lines
2.5 KiB
Python

from datetime import timedelta
import logging
from spacepackets.cfdp import ConditionCode
from spacepackets.util import UnsignedByteField
from cfdppy import CfdpUserBase, TransactionId
from cfdppy.mib import CheckTimerProvider, Countdown, EntityType
from cfdppy.user import (
TransactionFinishedParams,
MetadataRecvParams,
FileSegmentRecvdParams,
TransactionParams,
)
_LOGGER = logging.getLogger(__name__)
class EiveCheckTimerProvider(CheckTimerProvider):
def provide_check_timer(
self,
local_entity_id: UnsignedByteField,
remote_entity_id: UnsignedByteField,
entity_type: EntityType,
) -> Countdown:
return Countdown(timedelta(seconds=5.0))
class EiveCfdpUser(CfdpUserBase):
def transaction_indication(
self,
transaction_indication_params: TransactionParams,
):
_LOGGER.info(
f"CFDP User: Start of File {transaction_indication_params.transaction_id}"
)
def eof_sent_indication(self, transaction_id: TransactionId):
_LOGGER.info(f"CFDP User: EOF sent for {transaction_id}")
def transaction_finished_indication(self, params: TransactionFinishedParams):
_LOGGER.info(f"CFDP User: {params.transaction_id} finished")
_LOGGER.info(f"Delivery Code: {params.finished_params.delivery_code!r}")
_LOGGER.info(f"Condition code: {params.finished_params.condition_code!r}")
_LOGGER.info(f"File delivery status: {params.finished_params.delivery_code!r}")
def metadata_recv_indication(self, params: MetadataRecvParams):
pass
def file_segment_recv_indication(self, params: FileSegmentRecvdParams):
_LOGGER.info(
f"CFDP User: Received File Data PDU for {params.transaction_id} | Offset:"
f" {params.offset} | Segment Length: {params.length}"
)
def report_indication(self, transaction_id: TransactionId, status_report: any):
pass
def suspended_indication(
self, transaction_id: TransactionId, cond_code: ConditionCode
):
pass
def resumed_indication(self, transaction_id: TransactionId, progress: int):
pass
def fault_indication(
self, transaction_id: TransactionId, cond_code: ConditionCode, progress: int
):
pass
def abandoned_indication(
self, transaction_id: TransactionId, cond_code: ConditionCode, progress: int
):
pass
def eof_recv_indication(self, transaction_id: TransactionId):
_LOGGER.info(f"CFDP User: EOF received for {transaction_id}")