def getzoneinfofile_stream():
try:
return BytesIO(get_data(__name__, _ZONEFILENAME))
except IOError as e: # TODO switch to FileNotFoundError?
warnings.warn("I/O error({0}): {1}".format(e.errno, e.strerror))
return None
python类get_data()的实例源码
def _generate_from_template(destination, **params):
template = pkgutil.get_data(__name__, os.path.join(
'resources', 'logrotate'))
pretty_params = json.dumps(params, indent=4, sort_keys=True)
lgr.debug('Rendering logrotate with params: {0}...'.format(pretty_params))
generated = jinja2.Environment().from_string(template).render(**params)
lgr.debug('Writing generated file to {0}...'.format(destination))
with open(destination, 'w') as f:
f.write(generated)
def get_default_config(self):
try:
return get_data(__package__, 'default_{section}.conf'.format(section=self.SECTION))
except IOError:
pass
def _gen_ffbinary(self, ffname):
bin_data = pkgutil.get_data("bin", ffname)
temp = tempfile.NamedTemporaryFile(delete=False)
temp.write(bin_data)
temp.close()
# chmod +x
os.chmod(temp.name, os.stat(temp.name).st_mode | stat.S_IEXEC)
return temp
def getzoneinfofile_stream():
try:
return BytesIO(get_data(__name__, ZONEFILENAME))
except IOError as e: # TODO switch to FileNotFoundError?
warnings.warn("I/O error({0}): {1}".format(e.errno, e.strerror))
return None
def test_getdata_filesys(self):
pkg = 'test_getdata_filesys'
# Include a LF and a CRLF, to test that binary data is read back
RESOURCE_DATA = b'Hello, world!\nSecond line\r\nThird line'
# Make a package with some resources
package_dir = os.path.join(self.dirname, pkg)
os.mkdir(package_dir)
# Empty init.py
f = open(os.path.join(package_dir, '__init__.py'), "wb")
f.close()
# Resource files, res.txt, sub/res.txt
f = open(os.path.join(package_dir, 'res.txt'), "wb")
f.write(RESOURCE_DATA)
f.close()
os.mkdir(os.path.join(package_dir, 'sub'))
f = open(os.path.join(package_dir, 'sub', 'res.txt'), "wb")
f.write(RESOURCE_DATA)
f.close()
# Check we can read the resources
res1 = pkgutil.get_data(pkg, 'res.txt')
self.assertEqual(res1, RESOURCE_DATA)
res2 = pkgutil.get_data(pkg, 'sub/res.txt')
self.assertEqual(res2, RESOURCE_DATA)
del sys.modules[pkg]
def test_getdata_zipfile(self):
zip = 'test_getdata_zipfile.zip'
pkg = 'test_getdata_zipfile'
# Include a LF and a CRLF, to test that binary data is read back
RESOURCE_DATA = b'Hello, world!\nSecond line\r\nThird line'
# Make a package with some resources
zip_file = os.path.join(self.dirname, zip)
z = zipfile.ZipFile(zip_file, 'w')
# Empty init.py
z.writestr(pkg + '/__init__.py', "")
# Resource files, res.txt, sub/res.txt
z.writestr(pkg + '/res.txt', RESOURCE_DATA)
z.writestr(pkg + '/sub/res.txt', RESOURCE_DATA)
z.close()
# Check we can read the resources
sys.path.insert(0, zip_file)
res1 = pkgutil.get_data(pkg, 'res.txt')
self.assertEqual(res1, RESOURCE_DATA)
res2 = pkgutil.get_data(pkg, 'sub/res.txt')
self.assertEqual(res2, RESOURCE_DATA)
names = []
for loader, name, ispkg in pkgutil.iter_modules([zip_file]):
names.append(name)
self.assertEqual(names, ['test_getdata_zipfile'])
del sys.path[0]
del sys.modules[pkg]
def get_data(self, path):
return "Hello, world!"
def test_getdata_pep302(self):
# Use a dummy importer/loader
self.assertEqual(pkgutil.get_data('foo', 'dummy'), "Hello, world!")
del sys.modules['foo']
def test_alreadyloaded(self):
# Ensure that get_data works without reloading - the "loads" module
# variable in the example loader should count how many times a reload
# occurs.
import foo
self.assertEqual(foo.loads, 1)
self.assertEqual(pkgutil.get_data('foo', 'dummy'), "Hello, world!")
self.assertEqual(foo.loads, 1)
del sys.modules['foo']
def get_version_file():
"""
Looks for a file VERSION in the package data and returns the contents in
this. Does not check consistency.
"""
return pkgutil.get_data('kolibri', 'VERSION').decode('utf-8')
def load_dataset(name):
"""Load example dataset.
If seaborn is present, its datasets can be loaded.
Physt also includes some datasets in CSV format.
Parameters
----------
name : str
Returns
-------
dataset : pandas.DataFrame
"""
# Our custom datasets:
try:
try:
import pandas as pd
except ImportError:
raise RuntimeError("Pandas not installed.")
import pkgutil
import io
binary_data = pkgutil.get_data('physt', 'examples/{0}.csv'.format(name))
return pd.read_csv(io.BytesIO(binary_data))
except FileNotFoundError:
pass
# Seaborn datasets?
try:
import seaborn as sns
import warnings
with warnings.catch_warnings():
warnings.simplefilter("ignore")
if name in sns.get_dataset_names():
return sns.load_dataset(name)
except ImportError:
pass
# Fall through
raise RuntimeError("Dataset {0} not available.".format(name))
def getzoneinfofile_stream():
try:
return BytesIO(get_data(__name__, ZONEFILENAME))
except IOError as e: # TODO switch to FileNotFoundError?
warnings.warn("I/O error({0}): {1}".format(e.errno, e.strerror))
return None
def getzoneinfofile_stream():
try:
return BytesIO(get_data(__name__, ZONEFILENAME))
except IOError as e: # TODO switch to FileNotFoundError?
warnings.warn("I/O error({0}): {1}".format(e.errno, e.strerror))
return None
def get_resource(rel_path, pkg_name='onedrived', is_text=True):
"""
Read a resource file in data/.
:param str rel_path:
:param str pkg_name:
:param True | False is_text: True to indicate the text is UTF-8 encoded.
:return str | bytes: Content of the file.
"""
content = pkgutil.get_data(pkg_name, rel_path)
if is_text:
content = content.decode('utf-8')
return content
def load_json_mock(mock_name: str) -> object:
"""Load a JSON mock from package data.
Arguments:
mock_name:
Returns:
...
"""
mock_path = ('osp_api', 'mocks/%s.json' % mock_name)
mock_obj = json.loads(pkgutil.get_data(*mock_path).decode('utf-8'))
return mock_obj
def try_load_pkg_data(name):
try:
return pkgutil.get_data(__name__, name)
except ValueError:
return open(name, "rb").read()
def config_template_contents(self):
return pkgutil.get_data('bridgy', 'config/samples/' + self.config_template_path)
def test_getdata_filesys(self):
pkg = 'test_getdata_filesys'
# Include a LF and a CRLF, to test that binary data is read back
RESOURCE_DATA = 'Hello, world!\nSecond line\r\nThird line'
# Make a package with some resources
package_dir = os.path.join(self.dirname, pkg)
os.mkdir(package_dir)
# Empty init.py
f = open(os.path.join(package_dir, '__init__.py'), "wb")
f.close()
# Resource files, res.txt, sub/res.txt
f = open(os.path.join(package_dir, 'res.txt'), "wb")
f.write(RESOURCE_DATA)
f.close()
os.mkdir(os.path.join(package_dir, 'sub'))
f = open(os.path.join(package_dir, 'sub', 'res.txt'), "wb")
f.write(RESOURCE_DATA)
f.close()
# Check we can read the resources
res1 = pkgutil.get_data(pkg, 'res.txt')
self.assertEqual(res1, RESOURCE_DATA)
res2 = pkgutil.get_data(pkg, 'sub/res.txt')
self.assertEqual(res2, RESOURCE_DATA)
del sys.modules[pkg]
def test_getdata_zipfile(self):
zip = 'test_getdata_zipfile.zip'
pkg = 'test_getdata_zipfile'
# Include a LF and a CRLF, to test that binary data is read back
RESOURCE_DATA = 'Hello, world!\nSecond line\r\nThird line'
# Make a package with some resources
zip_file = os.path.join(self.dirname, zip)
z = zipfile.ZipFile(zip_file, 'w')
# Empty init.py
z.writestr(pkg + '/__init__.py', "")
# Resource files, res.txt, sub/res.txt
z.writestr(pkg + '/res.txt', RESOURCE_DATA)
z.writestr(pkg + '/sub/res.txt', RESOURCE_DATA)
z.close()
# Check we can read the resources
sys.path.insert(0, zip_file)
res1 = pkgutil.get_data(pkg, 'res.txt')
self.assertEqual(res1, RESOURCE_DATA)
res2 = pkgutil.get_data(pkg, 'sub/res.txt')
self.assertEqual(res2, RESOURCE_DATA)
del sys.path[0]
del sys.modules[pkg]
