Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def generate_tasks_input():
pyver = "py{}{}".format(sys.version_info.major, sys.version_info.minor)
if sys.version_info < (3, 0, 0):
DROP_TAG = "py3"
else:
DROP_TAG = "py2"
test_dir = ytcfg.get("yt", "test_data_dir")
answers_dir = os.path.join(test_dir, "answers")
with open('tests/tests.yaml', 'r') as obj:
lines = obj.read()
data = '\n'.join([line for line in lines.split('\n')
if DROP_TAG not in line])
tests = yaml.load(data)
base_argv = ['-s', '--nologcapture', '--with-xunit']
base_answer_argv = ['--local-dir=%s' % answers_dir, '--with-answer-testing',
'--answer-big-data', '--local']
args = []
for test in list(tests["other_tests"].keys()):
args.append(([test] + base_argv + tests["other_tests"][test], True))
def _get_girder_client():
try:
import girder_client
except ImportError:
raise YTCommandRequiresModule('girder_client')
if not ytcfg.get("yt", "hub_api_key"):
print("Before you can access the yt Hub you need an API key")
print("In order to obtain one, either register by typing:")
print(" yt hub register")
print("or follow the instruction on:")
print(" http://yt-project.org/docs/dev/sharing_data.html#obtaining-an-api-key")
sys.exit()
hub_url = urlparse(ytcfg.get("yt", "hub_url"))
gc = girder_client.GirderClient(apiUrl=hub_url.geturl())
gc.authenticate(apiKey=ytcfg.get("yt", "hub_api_key"))
return gc
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see .
"""
from yt.logger import fidoLogger as mylog
from yt.config import ytcfg
from yt.arraytypes import *
import os, os.path, shutil, time, sys, glob, types
OTHERFILES=["rates.out","cool_rates.out"]
NEW_OUTPUT_CREATED = ytcfg.get("fido","NewOutputCreated")
GlobPatterns = ytcfg.get("fido","GlobPatterns").split(",")
NewDirectoryPattern = ytcfg.get("fido","NewDirectoryPattern",raw=True)
# Let's define some useful functions. Maybe these should go elsewhere?
def get_parameter_line(filename, parameter):
f = open(filename)
lines = filter(lambda a: a.startswith(parameter), f)
if len(lines) > 1:
raise KeyError, "More than one line matches that parameter!"
else: return lines[0]
def get_parent_dir(filename):
return os.path.normpath( \
os.path.split(os.path.dirname(os.path.abspath(filename)))[0])
from OutputCollection import *
from FileHandling import *
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see .
"""
from yt.config import ytcfg
from yt.mods import *
import zipfile, tempfile, glob, os, stat
from dropio.client import DropIoClient
API_KEY = ytcfg.get("yt", "dropiokey")
class SharedDataProduct(object):
_upload_filename = None
_temp_file = None
def upload(self, drop_file_name, drop_name = None):
self._create_archive()
# This is where our drop.io API calls go
client = DropIoClient(API_KEY)
if drop_name is None:
drop = client.create_drop
else:
drop = client.get_drop(drop_name)
if drop is None: drop = client.create_drop(drop_name)
drop_name = drop.name
for asset in client.get_asset_list(drop_name):
if asset.name != drop_file_name: continue
def runFido():
pp = optparse.OptionParser(description="Archives data and creates plots.", \
version="Your Mom")
pp.add_option("-c","--config", action="store", dest="cfgFile", \
default=None, type="string")
pp.add_option("-p","--path", action="store", dest="path", \
default="", help="Where to move files when archiving")
pp.add_option("-d","--deliverator", action="store_true", dest="deliverator", \
default=False, \
help="Should we submit to the Deliverator?\n(This requires that httpPrefix be defined in [raven] in your yt config file!)")
opts, args = pp.parse_args()
imagePath = ytcfg.get("raven", "imagePath", raw=True)
imageSkel = ytcfg.get("raven", "imageSkel", raw=True)
import yt.fido as fido
import yt.raven as raven
def fidoPlots(h):
md = h["MetaDataString"]
if md.strip().rstrip() == "(null)":
fido.error("MetaDataString undefined in parameter file; either use it, or write your own durn function!")
raise KeyError
if opts.cfgFile != None:
RunID = -1
if opts.deliverator:
import yt.deliverator as deliverator
httpPrefix = ytcfg.get("raven", "httpPrefix", raw=True)
RunID, Response = deliverator.SubmitRun(md, ytcfg.get("Deliverator","user"))
imageDir = imagePath % h
if not os.path.isdir(imageDir):
def _get_db_name(self):
base_file_name = ytcfg.get("yt", "ParameterFileStore")
if not os.access(os.path.expanduser("~/"), os.W_OK):
return os.path.abspath(base_file_name)
return os.path.expanduser("~/.yt/%s" % base_file_name)
def get_data(fn):
warnings.warn("The yt.utilities.exodusII_reader module is deprecated "
"and will be removed in a future release. "
"Please use the normal yt.load() command to access "
"your data instead.")
try:
f = Dataset(fn)
except RuntimeError:
f = Dataset(os.path.join(ytcfg.get("yt", "test_data_dir"), fn))
fvars = f.variables
# Is this correct?
etypes = fvars["eb_status"][:]
nelem = etypes.shape[0]
varnames = [sanitize_string(v.tostring()) for v in
fvars["name_elem_var"][:]]
nodnames = [sanitize_string(v.tostring()) for v in
fvars["name_nod_var"][:]]
coord = np.array([fvars["coord%s" % ax][:]
for ax in 'xyz']).transpose().copy()
coords = []
connects = []
data = []
for i in range(nelem):
connects.append(fvars["connect%s" % (i+1)][:].astype("i8"))
ci = connects[-1]
def _get_db_name(self):
base_file_name = ytcfg.get("yt","ParameterFileStore")
if not os.access(os.path.expanduser("~/"), os.W_OK):
return os.path.abspath(base_file_name)
return os.path.expanduser("~/.yt/%s" % base_file_name)
YTSelectionContainer
from yt.funcs import iterable
from yt.geometry.selection_routines import convert_mask_to_indices
import yt.geometry.particle_deposit as particle_deposit
from yt.units.yt_array import YTArray
from yt.utilities.exceptions import \
YTFieldTypeNotFound, \
YTParticleDepositionNotImplemented
from yt.utilities.lib.interpolators import \
ghost_zone_interpolate
from yt.utilities.lib.mesh_utilities import \
clamp_edges
from yt.utilities.nodal_data_utils import \
get_nodal_slices
RECONSTRUCT_INDEX = bool(ytcfg.get('yt', 'reconstruct_index'))
class AMRGridPatch(YTSelectionContainer):
_spatial = True
_num_ghost_zones = 0
_grids = None
_id_offset = 1
_cache_mask = True
_type_name = 'grid'
_skip_add = True
_con_args = ('id', 'filename')
_container_fields = (("index", "dx"),
("index", "dy"),
("index", "dz"),
("index", "x"),
("index", "y"),