Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def upload_test_resources(args: Args):
# Try running the download pipeline
try:
# Get test resources dir
resources_dir = (
Path(__file__).parent.parent / "aicsimageio" / "tests" / "resources"
).resolve(strict=True)
# Report with directory will be used for upload
log.info(f"Using contents of directory: {resources_dir}")
# Create quilt package
package = Package()
package.set_dir("resources", resources_dir)
# Report package contents
log.info(f"Package contents: {package}")
# Construct package name
package_name = "aicsimageio/test_resources"
# Check for dry run
if args.dry_run:
# Attempt to build the package
built = package.build(package_name)
# Get resolved save path
manifest_save_path = Path("upload_manifest.jsonl").resolve()
with open(manifest_save_path, "w") as manifest_write:
# Wait for a minute for the cluster to fully shutdown
time.sleep(60)
#######################################################################
log.info(f"Completed all tests")
log.info(f"=" * 80)
# Ensure save dir exists and save results
args.save_path.parent.mkdir(parents=True, exist_ok=True)
with open(args.save_path, "w") as write_out:
json.dump(all_results, write_out)
# Construct and push package
if args.upload:
p = Package()
p.set("results.json", args.save_path)
p.push(
"aicsimageio/benchmarks",
"s3://aics-modeling-packages-test-resources",
message=f"aicsimageio version: {aicsimageio.__version__}",
)
# Catch any exception
except Exception as e:
log.error("=============================================")
if args.debug:
log.error("\n\n" + traceback.format_exc())
log.error("=============================================")
log.error("\n\n" + str(e) + "\n")
log.error("=============================================")
sys.exit(1)
import quilt3
from appdirs import user_data_dir
from requests.exceptions import Timeout
from shapely import wkb, wkt
appname = "geosnap"
appauthor = "geosnap"
data_dir = user_data_dir(appname, appauthor)
if not os.path.exists(data_dir):
pathlib.Path(data_dir).mkdir(parents=True, exist_ok=True)
# look for local storage and create if missing
try:
from quilt3.data.geosnap_data import storage
except ImportError:
storage = quilt3.Package()
class _Map(dict):
"""tabbable dict."""
def __init__(self, *args, **kwargs):
super(_Map, self).__init__(*args, **kwargs)
for arg in args:
if isinstance(arg, dict):
for k, v in arg.iteritems():
self[k] = v
if kwargs:
for k, v in kwargs.iteritems():
self[k] = v
You can push a dataset with the same name multiple times to the same bucket multiple times as instead of
overriding a prior dataset, Quilt simply creates a new dataset version. Please refer to Quilt documentation for
more details: https://docs.quiltdata.com
:param push_uri: The S3 bucket uri to push to. Example: "s3://quilt-jacksonb"
:param message: An optional message to attach to that version of the dataset.
:param attach_associates: Boolean option to attach associates as metadata to each file. Associates are used
to retain quick navigation between related files.
:return: The built and optionally pushed quilt3.Package.
"""
# Confirm name matches approved pattern
# We previously checked during init, but the name could have been changed
name = self.return_or_raise_approved_name(self.name)
# Create empty package
pkg = quilt3.Package()
# Write any extra files to tempdir to send to the build
with tempfile.TemporaryDirectory() as tmpdir:
# Set all referenced files
text = self.readme.text
for rf in self.readme.referenced_files:
replaced = f"referenced_files/{rf.resolved.name}"
text = text.replace(rf.target, replaced)
pkg.set(replaced, str(rf.resolved))
# Write the updated readme to temp
readme_pk = Path(tmpdir, "README.md")
with open(readme_pk, "w") as readme_write:
readme_write.write(text)
# Set the readme
_fipstable = pd.read_csv(
os.path.join(os.path.dirname(os.path.abspath(__file__)), "stfipstable.csv"),
converters={"FIPS Code": str},
)
appname = "geosnap"
appauthor = "geosnap"
data_dir = user_data_dir(appname, appauthor)
if not os.path.exists(data_dir):
pathlib.Path(data_dir).mkdir(parents=True, exist_ok=True)
# look for local storage and create if missing
try:
from quilt3.data.geosnap_data import storage
except ImportError:
storage = quilt3.Package()
def store_census():
"""Save census data to the local quilt package storage.
Returns
-------
None
Data will be available in the geosnap.data.datasets and will be used
in place of streaming data for all census queries. The raster package
is 3.05 GB.
"""
quilt3.Package.install("census/tracts_cartographic", "s3://quilt-cgs")
quilt3.Package.install("census/administrative", "s3://quilt-cgs")
_fipstable = pd.read_csv(
os.path.join(os.path.dirname(os.path.abspath(__file__)), "stfipstable.csv"),
converters={"FIPS Code": str},
)
appname = "geosnap"
appauthor = "geosnap"
data_dir = user_data_dir(appname, appauthor)
if not os.path.exists(data_dir):
pathlib.Path(data_dir).mkdir(parents=True, exist_ok=True)
# look for local storage and create if missing
try:
from quilt3.data.geosnap_data import storage
except ImportError:
storage = quilt3.Package()
class DataStore(object):
"""Storage for geosnap data. Currently supports US Census data."""
def __init__(self):
"""Instantiate a new DataStore object."""
try: # if any of these aren't found, stream them insteead
from quilt3.data.census import tracts_cartographic, administrative
except ImportError:
warn(
"Unable to locate local census data. Streaming instead.\n"
"If you plan to use census data repeatedly you can store it locally "
"with the data.store_census function for better performance"
)
try:
def create_pkg(size_str):
pkg = quilt3.Package()
pkg.set_dir(f"{size_str}/", f"s3://quilt-ml-data/data/download_benchmark/{size_str}/")
pkg.push(f"download-benchmark/{size_str}", registry="s3://quilt-ml-data/", selector_fn=move_nothing)