18 from typing
import TypeVar, Generic, Dict, Any, List
19 from typing_extensions
import override
20 from datetime
import datetime
21 from pathlib
import Path
27 from .sample
import Sample
28 from ..project
import ProjectType
29 from ..._folder_manager
import folder_manager
30 from ...codable
import KeyDescriptor
31 from ...networking
import NetworkObject, networkManager, NetworkRequestError, \
32 fileChunkUpload, MAX_CHUNK_SIZE, FileData
33 from ...utils
import TIME_ZONE
34 from ...cryptography
import getProjectKey, aes
37 SampleDataType = TypeVar(
"SampleDataType")
40 def _relinkSample(samplePath: Path) ->
None:
41 for datasetPath
in folder_manager.datasetsFolder.iterdir():
42 linkPath = datasetPath / samplePath.name
43 if not linkPath.exists():
47 os.link(samplePath, linkPath)
50 class NetworkSample(Generic[SampleDataType], Sample[SampleDataType], NetworkObject):
53 Represents a base class for all Sample classes which are
54 comunicating with Coretex.ai
59 projectType: ProjectType
60 lastModified: datetime
68 Path -> path for network sample
71 return folder_manager.samplesFolder / str(self.id)
78 Path -> zip path for network sample
81 return self.
pathpathpath.with_suffix(
".zip")
88 Path -> path to which the network sample is downloaded to
94 def _keyDescriptors(cls) -> Dict[str, KeyDescriptor]:
95 descriptors = super()._keyDescriptors()
97 descriptors[
"projectType"] =
KeyDescriptor(
"project_task", ProjectType)
98 descriptors[
"lastModified"] =
KeyDescriptor(
"storage_last_modified", datetime)
103 def _endpoint(cls) -> str:
108 Checking if sample has been modified since last download, if the sample is already
113 bool -> False if sample has not changed since last download, True otherwise
117 FileNotFoundError -> sample file cannot be found
121 raise FileNotFoundError(
122 f
">> [Coretex] Sample file could not be found at {self.downloadPath}. "
123 "Cannot check if file has been modified since last download"
126 lastModified = datetime.fromtimestamp(self.
downloadPathdownloadPath.stat().st_mtime).astimezone(TIME_ZONE)
127 return self.lastModified > lastModified
129 def decrypt(self, ignoreCache: bool =
False) ->
None:
131 Decrypts the content of this Sample and caches
132 the results. Is ignored if the "isEncrypted" value is False.
137 defines if content should be decrypted if a cache for decryption
141 if not self.isEncrypted:
156 def _download(self, ignoreCache: bool =
False) ->
None:
172 response = networkManager.download(f
"{self._endpoint()}/export", self.
downloadPathdownloadPath, params)
173 if response.hasFailed():
174 raise NetworkRequestError(response, f
"Failed to download Sample \"{self.name}\"")
176 if not response.isHead():
187 def download(self, decrypt: bool =
True, ignoreCache: bool =
False) ->
None:
189 Downloads and optionally decrypts sample from Coretex.ai
193 NetworkRequestError -> if some kind of error happened during
197 if decrypt
and not self.isEncrypted:
206 self.
decryptdecrypt(ignoreCache)
215 def unzip(self, ignoreCache: bool =
False) ->
None:
217 raise RuntimeError(
"You must first download the Sample before you can unzip it")
220 raise RuntimeError(
"You must first decrypt the Sample before you can unzip it")
222 super().
unzip(ignoreCache)
225 def load(self) -> SampleDataType:
226 return super().load()
229 def _updateArchive(self) -> None:
230 super()._updateArchive()
235 def _overwriteSample(self, samplePath: Path) ->
None:
236 if not self.isEncrypted:
237 raise RuntimeError(
"Only encrypted samples can be overwriten.")
239 with folder_manager.tempFile()
as encryptedPath:
240 aes.encryptFile(getProjectKey(self.projectId), samplePath, encryptedPath)
242 params: Dict[str, Any] = {
246 files: List[FileData] = []
250 size = encryptedPath.stat().st_size
252 if size > MAX_CHUNK_SIZE:
253 params[
"file_id"] = fileChunkUpload(encryptedPath)
255 files.append(FileData.createFromPath(
"file", encryptedPath))
257 response = networkManager.formData(f
"{self._endpoint()}/upload", params, files)
258 if response.hasFailed():
259 raise NetworkRequestError(response, f
"Failed to overwrite Sample \"{self.name}\"")
None _download(self, bool ignoreCache=False)
None download(self, bool decrypt=True, bool ignoreCache=False)
None unzip(self, bool ignoreCache=False)
bool modifiedSinceLastDownload(self)
None decrypt(self, bool ignoreCache=False)