18 from typing
import Type, Any, Optional, List, Generator
19 from pathlib
import Path
20 from zipfile
import ZipFile, ZIP_DEFLATED
26 from ..dataset
import Dataset, LocalDataset, NetworkDataset, LocalCustomDataset, \
27 CustomDataset, LocalImageDataset, ImageDataset
28 from ..project
import ProjectType
29 from ..._folder_manager
import folder_manager
32 def getDatasetType(type_: ProjectType, isLocal: bool) -> Type[Dataset]:
33 if type_ == ProjectType.other:
35 return LocalCustomDataset
39 if type_
in [ProjectType.computerVision, ProjectType.imageSegmentation]:
41 return LocalImageDataset
45 logging.getLogger(
"coretexpylib").debug(f
">> [Coretex] ProjectType ({type_}) does not have a dataset type using CustomDataset")
49 return LocalCustomDataset
54 def fetchDataset(datasetType: Type[Dataset], value: Any) -> Optional[Dataset]:
55 if issubclass(datasetType, LocalDataset):
56 return datasetType(value)
58 if issubclass(datasetType, NetworkDataset):
59 return datasetType.fetchById(value)
64 def getSnapshotFiles(dirPath: Path, ignoredFiles: List[str]) -> List[Path]:
65 snapshotFiles: List[Path] = []
67 if dirPath.joinpath(
".coretexignore").exists():
70 for path
in dirPath.iterdir():
72 snapshotFiles.extend(getSnapshotFiles(path, ignoredFiles))
73 elif str(path)
not in ignoredFiles:
74 snapshotFiles.append(path)
79 def getDefaultEntryPoint() -> Optional[str]:
80 for defaultEntryPoint
in [Path(
".",
"main.py"), Path(
".",
"main.r"), Path(
".",
"main.R")]:
81 if defaultEntryPoint.exists():
82 return defaultEntryPoint.name
87 def chunks(lst: List, n: int) -> Generator[List,
None,
None]:
88 for i
in range(0, len(lst), n):
92 def createSnapshot() -> Path:
93 entryPoint = getDefaultEntryPoint()
94 if entryPoint
is None or not Path(
".", entryPoint).exists():
95 raise FileNotFoundError(
">> [Coretex] Entry point file not found")
97 ignoredFiles: List[str] = []
99 snapshotPath = folder_manager.temp /
"snapshot.zip"
100 with ZipFile(snapshotPath,
"w", ZIP_DEFLATED)
as snapshotArchive:
101 repo = git.Repo(Path.cwd(), search_parent_directories =
True)
102 for paths
in chunks(list(Path.cwd().rglob(
"*")), 256):
103 ignoredFiles.extend(repo.ignored(*paths))
105 if not Path(entryPoint).exists()
or not Path(
"requirements.txt").exists():
106 raise FileNotFoundError(f
">> [Coretex] Required files \"{entryPoint}\" and \"requirements.txt\"")
108 for path
in getSnapshotFiles(Path.cwd(), ignoredFiles):
109 snapshotArchive.write(path.relative_to(Path.cwd()))