Source code for openlifu.cloud.components.protocols

from __future__ import annotations

from datetime import datetime
from pathlib import Path
from typing import Any, List

from openlifu.cloud.api.api import Api
from openlifu.cloud.api.dto import CreateObjectRequestDto, DatabaseSyncRequestDto
from openlifu.cloud.components.abstract_component import AbstractComponent
from openlifu.cloud.sync_thread import SyncThread




class Protocols(AbstractComponent):

    def __init__(self, api: Api, parent_path: Path, database_id: int, sync_thread: SyncThread):
        super().__init__(api, parent_path, database_id, sync_thread, download_only=False)

    def get_config_ids_key(self) -> str:
        return "protocol_ids"

    def get_component_type_plural(self) -> str:
        return "protocols"

    def get_sync_date_from_cloud(self) -> datetime | None:
        return self.api.databases().get_database(self.db_id).protocols_sync_date

    def send_sync_date_to_cloud(self, sync_date: datetime):
        self.api.databases().update_database_sync_date(self.db_id, DatabaseSyncRequestDto(protocols_sync_date=sync_date))

    def upload_config(self, data: bytes, modification_date: datetime, local_id: str, remote_id: int | None) -> int:
        if not remote_id:
            remote_id = self.api.protocols().create(
                CreateObjectRequestDto(database_id=self.db_id, local_id=local_id)
            ).id

        self.api.protocols().upload_config(remote_id, data, modification_date)
        return remote_id

    def download_config(self, local_id: str, remote_id: int) -> bytes:
        return self.api.protocols().get_config(remote_id)

    def delete_on_cloud(self, local_id: str, remote_id: int):
        self.api.protocols().delete(remote_id)

    def get_cloud_items(self) -> List[Any]:
        return self.api.protocols().get_all(self.db_id)