Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def make_project(alias='project', root=None, **kwargs):
"""Initialize a project for testing.
The initialized project has a few operations and a few jobs that are in
various points in the workflow defined by the project.
"""
init(alias=alias, root=root, template='testing')
project = signac.init_project(name=alias, root=root)
signac.testing.init_jobs(project, **kwargs)
return project
def test_sync_file(self):
self.call('python -m signac init ProjectA'.split())
project_a = signac.Project()
project_b = signac.init_project('ProjectB', os.path.join(self.tmpdir.name, 'b'))
job_src = project_a.open_job({'a': 0}).init()
job_dst = project_b.open_job({'a': 0}).init()
for i, job in enumerate([job_src, job_dst]):
with open(job.fn('test'), 'w') as file:
file.write('x'*(i+1))
# FileSyncConflict
with pytest.raises(ExitCodeError):
self.call('python -m signac sync {} {}'
.format(os.path.join(self.tmpdir.name, 'b'), self.tmpdir.name).split())
self.call('python -m signac sync {} {} --strategy never'
.format(os.path.join(self.tmpdir.name, 'b'), self.tmpdir.name).split())
with open(job_dst.fn('test'), 'r') as file:
assert file.read() == 'xx'
with pytest.raises(ExitCodeError):
self.call('python -m signac sync {} {}'
def test_get_job_nested_project(self):
# Test case: The job workspace dir is also a project root dir.
root = self._tmp_dir.name
project = signac.init_project(name='testproject', root=root)
job = project.open_job({'a': 1})
job.init()
with job:
nestedproject = signac.init_project('nestedproject')
nestedproject.open_job({'b': 2}).init()
assert project.get_job() == job
assert signac.get_job() == job
def main(args):
if not os.path.exists(PROJECT_DIR):
os.makedirs(PROJECT_DIR)
elif args.force:
import shutil
shutil.rmtree(PROJECT_DIR)
os.makedirs(PROJECT_DIR)
else:
return
p = signac.init_project(name=gen.PROJECT_NAME, root=PROJECT_DIR)
p.import_from(origin=gen.ARCHIVE_DIR)
def test_get_project_non_local(self):
root = self._tmp_dir.name
subdir = os.path.join(root, 'subdir')
os.mkdir(subdir)
project = signac.init_project(root=root, name='testproject')
assert project == project.get_project(root=root)
assert project == signac.get_project(root=root)
assert project == project.get_project(root=root, search=False)
assert project == signac.get_project(root=root, search=False)
assert project == project.get_project(root=os.path.relpath(root), search=False)
assert project == signac.get_project(root=os.path.relpath(root), search=False)
with pytest.raises(LookupError):
assert project == project.get_project(root=subdir, search=False)
with pytest.raises(LookupError):
assert project == signac.get_project(root=subdir, search=False)
assert project == project.get_project(root=subdir, search=True)
assert project == signac.get_project(root=subdir, search=True)
assert project.workspace() == os.path.join(root, 'workspace')
assert project.root_directory() == root
project = signac.Project.get_project(root=root)
assert project.id == 'testproject'
assert project.workspace() == os.path.join(root, 'workspace')
assert project.root_directory() == root
# Deviating initialization parameters should result in errors.
with pytest.raises(RuntimeError):
signac.init_project(name='testproject2', root=root)
with pytest.raises(RuntimeError):
signac.init_project(
name='testproject',
root=root,
workspace='workspace2')
with pytest.raises(RuntimeError):
signac.init_project(
name='testproject2',
root=root,
workspace='workspace2')
def main_init(args):
"Initialize a FlowProject from one of the templates defined in the template module."
if not _is_identifier(args.alias):
raise ValueError(
"The alias '{}' is not a valid Python identifier and can therefore "
"not be used as a FlowProject alias.".format(args.alias))
try:
get_project()
except LookupError:
init_project(name=args.alias)
print("Initialized signac project with name '{}' in "
"current directory.".format(args.alias), file=sys.stderr)
try:
return template.init(alias=args.alias, template=args.template)
except OSError as e:
raise RuntimeError(
"Error occured while trying to initialize a flow project: {}".format(e))
#!/usr/bin/env python3
# Copyright (c) 2019 The Regents of the University of Michigan
# All rights reserved.
# This software is licensed under the BSD 3-Clause License.
import signac
if __name__ == '__main__':
project = signac.init_project('document_editor')
job = project.open_job({'play': 'Hamlet'})
job.doc['plot'] = 'Tragedy starring Prince Hamlet'
job.doc['characters'] = ['Hamlet', 'Claudius', 'Gertrude', 'The Ghost',
'Polonius', 'Laertes', 'Ophelia', 'Horatio',
'Rosencrantz', 'Guildenstern', 'Fortinbras']
job.doc['line1'] = {
'speaker': 'Hamlet',
'quote': 'To be, or not to be: that is the question'
}
job.doc['line2'] = {
'speaker': 'Polonius',
'quote': 'This above all: to thine own self be true'
}
job = project.open_job({'play': 'Julius Caesar'})
#!/usr/bin/env python3
# Copyright (c) 2018 The Regents of the University of Michigan
# All rights reserved.
# This software is licensed under the BSD 3-Clause License.
import signac
import numpy as np
import matplotlib
# Force matplotlib to not use any Xwindows backend.
matplotlib.use('Agg')
import matplotlib.pyplot as plt # noqa: E402
try:
project = signac.get_project()
except LookupError:
project = signac.init_project('plots')
def plot_coherence(job):
# Plot script adapted from:
# https://matplotlib.org/gallery/lines_bars_and_markers/cohere.html
print('Making plots for coherence time {}, job {}'.format(
job.sp.coherence_time, job))
# Fixing random state for reproducibility
np.random.seed(job.sp.seed)
dt = 0.01
t = np.arange(0, 30, dt)
nse1 = np.random.randn(len(t)) # white noise 1
nse2 = np.random.randn(len(t)) # white noise 2
#!/usr/bin/env python3
# Copyright (c) 2019 The Regents of the University of Michigan
# All rights reserved.
# This software is licensed under the BSD 3-Clause License.
import signac
project = signac.init_project('flow_status')
for i in range(10):
job = project.open_job({'step': i})
job.doc['step'] = i