eive-tmtc/eive_tmtc/cfdp/user.py

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}")