Skip to content
Snippets Groups Projects
Commit 40d475b4 authored by Mayr, Hannes's avatar Mayr, Hannes
Browse files

Publish data now in a subdirectory of the given destination folder with the...

Publish data now in a subdirectory of the given destination folder with the corresponding ID as directory name.
parent 6708a678
Branches
Tags
7 merge requests!41Include latest changes in main branch,!37Merge dev upstream changes into improve/metadata,!34Include architecture diagram in docs,!32SAST implementation,!27Update documentation and version number,!26Merge !23, !24, !25 into main,!22Restructure and introduce additional object to transfer figures and IDs from...
Pipeline #802827 waiting for manual action
......@@ -51,8 +51,6 @@ IMGS_AS_LIST = [IMG1, IMG2]
FIGS_AND_IDS = tagplot(IMGS_AS_LIST, 'image', prefix=PROJECT_ID,
id_method='time', location='west')
print(FIGS_AND_IDS.figure_ids)
# %% Publish
# Arguments: Source directory or files as list, destination directory, figures,
# plots or images.
......
......@@ -142,35 +142,44 @@ class PublishOptions:
# Export plot figure to picture.
plot_paths = save_plot(self.figure, self.plot_names)
# If dst dir already exists ask user if it should be overwritten.
if os.path.isdir(self.dst_path):
warnings.warn(f'Folder "{self.dst_dirname}" already exists – '
'plot has already been published.')
overwrite_dir = input('Do you want to overwrite the existing'
' folder? (yes/no[default])\n')
if overwrite_dir in ('yes', 'y'):
shutil.rmtree(self.dst_path)
else:
raise RuntimeError('publish has finished without an export.\n'
'Rerun TagPlot if you need a new ID or '
'consider overwriting.')
# Create invisible folder
dst_path_invisible = os.path.join(self.dst_path_head,
'.' + self.dst_dirname)
match self.data_storage:
case 'centralized':
self.centralized_data_storage()
case 'individual':
for i, plot in enumerate(plot_paths):
try:
self.individual_data_storage(dst_path_invisible,
plot_paths)
except Exception as exc:
delete_dir = input('There was an error while publishing'
' the data. Should the partially copied'
f' data at {dst_path_invisible} be'
# Create folder with ID as name
dst_path = os.path.join(self.dst_path,
self.figure_ids[i])
dst_path_invisible = os.path.join(self.dst_path, '.'
+ self.figure_ids[i])
# If dir with the same ID already exists ask user
# if it should be overwritten.
if os.path.isdir(dst_path):
warnings.warn(f'Folder "{dst_path}" already exists'
' – plot has already been published.'
)
overwrite_dir = input('Do you want to overwrite '
'the existing folder? '
'(yes/no[default])\n')
if overwrite_dir in ('yes', 'y'):
shutil.rmtree(dst_path)
else:
raise RuntimeError('publish has finished '
'without an export.\nRerun '
'tagplot if you need a new'
' ID or consider '
'overwriting.')
self.individual_data_storage(dst_path_invisible, plot)
# If export was successful, make the directory visible
os.rename(dst_path_invisible, dst_path)
except FileExistsError as exc:
delete_dir = input('There was an error while '
'publishing the data. Should the '
'partially copied data at '
f'{dst_path_invisible} be'
' removed? (yes/no[default])\n')
if delete_dir in ('yes', 'y'):
shutil.rmtree(dst_path_invisible)
......@@ -180,8 +189,6 @@ class PublishOptions:
raise ValueError(f'The data storage method {self.data_storage}'
' is not available.')
# If export was successful, make the directory visible
os.rename(dst_path_invisible, self.dst_path)
print(f'Publish was successful.\nYour plot(s) {plot_paths},\nyour'
f' data {self.src_datapaths}\nand your script {sys.argv[0]}\n'
f'were copied to {self.dst_path}\nin {self.data_storage} mode.')
......@@ -199,7 +206,7 @@ class PublishOptions:
"""
# Does nothing, not implemented yet
def individual_data_storage(self, destination, pic_paths):
def individual_data_storage(self, destination, pic_path):
"""
Store all the data in an individual directory.
......@@ -229,10 +236,9 @@ class PublishOptions:
# Copy script that calls this function to folder
shutil.copy2(sys.argv[0], destination)
# Copy plot files to folder
for path in pic_paths:
if os.path.isfile(path):
shutil.copy2(path, destination)
os.remove(path)
if os.path.isfile(pic_path):
shutil.copy2(pic_path, destination)
os.remove(pic_path)
def publish(figs_and_ids, src_datapath, dst_path, plot_name, **kwargs):
......
......@@ -22,7 +22,6 @@ PIC_NAME = 'test_picture'
DST_DIR = 'test_dst_folder'
DST_PARENT_DIR = 'test_parent'
DST_PATH = os.path.join(DST_PARENT_DIR, DST_DIR)
INVISIBLE_PATH = os.path.join(DST_PARENT_DIR, '.' + DST_DIR)
FIG = plt.figure(figsize=[6.4, 4.8], dpi=100)
FIG2 = plt.figure(figsize=[6.4, 4.8], dpi=100)
FIGS_AS_LIST = [FIG, FIG2]
......@@ -51,7 +50,8 @@ class TestPublish(unittest.TestCase):
""" Test publish and check if an exported picture file exists. """
publish(PlotIDTransfer(FIG, 'testID'), SRC_DIR, DST_PATH + '/',
PIC_NAME, data_storage='individual')
assert os.path.isfile(os.path.join(DST_PATH, PIC_NAME + '.png'))
assert os.path.isfile(os.path.join(DST_PATH, 'testID',
PIC_NAME + '.png'))
# Skip test if tests are run from command line.
@unittest.skipIf(not os.path.isfile(sys.argv[0]), 'Publish is not called '
......@@ -63,8 +63,9 @@ class TestPublish(unittest.TestCase):
files exist.
"""
publish(FIGS_AND_IDS, SRC_DIR, DST_PATH, PIC_NAME_LIST)
for name in PIC_NAME_LIST:
assert os.path.isfile(os.path.join(DST_PATH, name + '.png'))
for i, name in enumerate(PIC_NAME_LIST):
assert os.path.isfile(os.path.join(DST_PATH, IDS_AS_LIST[i],
name + '.png'))
def test_wrong_ids(self):
""" Test if Error is raised if IDs are of wrong type. """
......@@ -82,9 +83,11 @@ class TestPublish(unittest.TestCase):
files_and_dir.append(SRC_DIR)
publish(FIGS_AND_IDS, files_and_dir, DST_PATH, PIC_NAME_LIST,
data_storage='individual')
assert os.path.isdir(DST_PATH)
for identifier in IDS_AS_LIST:
for file in SRC_FILES:
assert os.path.isfile(os.path.join(DST_PATH, file))
path = os.path.join(DST_PATH, identifier)
assert os.path.isdir(path)
assert os.path.isfile(os.path.join(path, file))
def test_src_directory(self):
""" Test if Error is raised when source directory does not exist."""
......@@ -149,6 +152,7 @@ class TestPublish(unittest.TestCase):
destination directory.
"""
os.mkdir(DST_PATH)
os.mkdir(os.path.join(DST_PATH, IDS_AS_LIST[0]))
# Mock user input as 'yes'
with patch('builtins.input', return_value='yes'):
publish(FIGS_AND_IDS, SRC_DIR, DST_PATH, PIC_NAME,
......@@ -164,6 +168,8 @@ class TestPublish(unittest.TestCase):
an existing destination directory by user input 'no'.
"""
os.mkdir(DST_PATH)
os.mkdir(os.path.join(DST_PATH, IDS_AS_LIST[0]))
os.mkdir(os.path.join(DST_PATH, IDS_AS_LIST[1]))
# Mock user input as 'no'
with patch('builtins.input', return_value='no'):
with self.assertRaises(RuntimeError):
......@@ -179,6 +185,8 @@ class TestPublish(unittest.TestCase):
an existing destination directory by missing user input.
"""
os.mkdir(DST_PATH)
os.mkdir(os.path.join(DST_PATH, IDS_AS_LIST[0]))
os.mkdir(os.path.join(DST_PATH, IDS_AS_LIST[1]))
# Mock user input as empty (no should be default).
with patch('builtins.input', return_value=''):
with self.assertRaises(RuntimeError):
......@@ -195,12 +203,14 @@ class TestPublish(unittest.TestCase):
removed.
To mock this, the invisible directory already exists.
"""
os.mkdir(INVISIBLE_PATH)
# Mock user input as 'yes'
with patch('builtins.input', return_value='yes'):
os.mkdir(DST_PATH)
INVISIBLE_PATH1 = os.path.join(DST_PATH, '.' + IDS_AS_LIST[0])
os.mkdir(INVISIBLE_PATH1)
# Mock user input as 'yes' for deleting the partial copied data
with patch('builtins.input', side_effect=['yes', 'yes']):
with self.assertRaises(RuntimeError):
publish(FIGS_AND_IDS, SRC_DIR, DST_PATH, PIC_NAME)
assert not os.path.isdir(INVISIBLE_PATH)
assert not os.path.isdir(INVISIBLE_PATH1)
os.remove('test_picture1.png')
os.remove('test_picture2.png')
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment