def __init__(self):
self._classification_rules = {}
self._exclusion_rules = {}
self._enrichment_rules = {}
self._routing_rules = {}
self._config = None
try:
# TODO: Absolute path? Where should this live?
with open('config/klaxer.yml', 'r') as ymlfile:
self._config = yaml.load(ymlfile)
except yaml.YAMLError as ye:
raise ConfigurationError('failed to parse config') from ye
for section in self._config:
# Subsequent definitions of the same service will overwrite the
# previous ones.
self._build_rules(section)
python类load()的实例源码
def process_abot_test_result(file_path):
""" Process ABoT Result """
with open(file_path) as test_result:
data = json.load(test_result)
res = []
for tests in data:
tests = update_data(tests)
try:
flatten_steps = tests['elements'][0].pop('flatten_steps')
for steps in flatten_steps:
steps['result'] = steps['step_status']
res.append(steps)
except:
logging.error("Could not post data to ElasticSearch host")
raise
return res
def post(self):
if self.current_user['level'] != 0:
self.custom_error()
settings = {
'init_money': int(self.get_body_argument('init_money')),
'reg_type': self.get_body_argument('reg_type'),
'cookie_secret': self.get_body_argument('cookie_secret') or self.settings['cookie_secret'],
'site': {
'name': self.get_body_argument('sitename'),
'keyword': self.get_body_argument('keyword'),
'description': self.get_body_argument('description')
}
}
self.settings.update(settings)
custom_settings = {}
with open(self.settings['config_file'], 'r') as f:
custom_settings = yaml.load(f)
custom_settings['global'].update(settings)
with open(self.settings['config_file'], 'w') as f:
yaml.dump(custom_settings, f,
default_flow_style=False, default_style='"')
self.redirect('/ushio/setting')
def parse_config(conf_file):
if not os.path.isfile(conf_file):
logging.error('Invalid config file: %s.' % conf_file)
return False
return yaml.load(open(conf_file).read())
def __init__(self, *args):
self.required_options = args
self['config'] = hookenv.config()
with open(os.path.join(hookenv.charm_dir(), 'config.yaml')) as fp:
self.config = yaml.load(fp).get('options', {})
def read_context(self, file_name):
if not os.path.isabs(file_name):
file_name = os.path.join(hookenv.charm_dir(), file_name)
with open(file_name, 'r') as file_stream:
data = yaml.load(file_stream)
if not data:
raise OSError("%s is empty" % file_name)
return data
def get_matchmaker_map(mm_file='/etc/oslo/matchmaker_ring.json'):
mm_map = {}
if os.path.isfile(mm_file):
with open(mm_file, 'r') as f:
mm_map = json.load(f)
return mm_map
def _git_yaml_load(projects_yaml):
"""
Load the specified yaml into a dictionary.
"""
if not projects_yaml:
return None
return yaml.load(projects_yaml)
def __init__(self, iface):
self.iface = iface
self.path = os.path.dirname(os.path.realpath(__file__))
locale = QSettings().value("locale/userLocale")[0:2]
localePath = os.path.join(self.path, 'i18n', '{}.qm'.format(locale))
if os.path.exists(localePath):
self.translator = QTranslator()
self.translator.load(localePath)
QCoreApplication.installTranslator(self.translator)
def createRotationLayer(self):
pies = QgsMapLayerRegistry.instance().mapLayersByName(PIE_LAYER)
if not pies:
self.iface.messageBar().pushCritical(
self.tr(u'No layer'), self.tr(u'Please add "{}" layer.').format(PIE_LAYER))
return
pie = pies[0]
if not pie.featureCount():
self.iface.messageBar().pushInfo(
self.tr(u'No data'), self.tr(u'No features in the "{}" layer.').format(PIE_LAYER))
return
if pie.isEditable():
self.iface.vectorLayerTools().saveEdits(pie)
boxes = runalg('qgis:orientedminimumboundingbox', pie, True, None)
boxesLayer = QgsVectorLayer(boxes['OUTPUT'], ROTATION_LAYER, 'ogr')
if not boxesLayer.isValid():
self.iface.messageBar().pushCritical(
self.tr(u'Access error'), self.tr(u'Failed to load a temporary processing layer.'))
return
self.addFieldToLayer(boxesLayer, NAME_FIELD, QVariant.String)
rotIndex = boxesLayer.dataProvider().fieldNameIndex('ANGLE')
nameIndex = boxesLayer.dataProvider().fieldNameIndex(NAME_FIELD)
iterpie = pie.getFeatures()
for box in boxesLayer.getFeatures():
name = next(iterpie)['name']
angle = round(box['ANGLE'])
if box['WIDTH'] > box['HEIGHT']:
angle += 90 if angle < 0 else -90
geom = QgsGeometry(box.geometry())
geom.rotate(angle, box.geometry().boundingBox().center())
boxesLayer.dataProvider().changeAttributeValues(
{box.id(): {rotIndex: angle, nameIndex: name}})
boxesLayer.dataProvider().changeGeometryValues({box.id(): geom})
QgsMapLayerRegistry.instance().addMapLayer(boxesLayer)
self.iface.legendInterface().setLayerVisible(boxesLayer, False)
self.iface.legendInterface().setLayerVisible(pie, False)
return boxesLayer
def show_io(input_dir, output_dir):
swrite('\n=== DIRECTORIES ===\n\n')
# Show this directory
swrite("-- Current directory " + pwd() + ":\n")
write_list(ls('.'))
write_list(ls('./*'))
write_list(ls('./*/*'))
swrite("\n")
# List input and output directories
swrite("-- Input directory " + input_dir + ":\n")
write_list(ls(input_dir))
write_list(ls(input_dir + '/*'))
write_list(ls(input_dir + '/*/*'))
write_list(ls(input_dir + '/*/*/*'))
swrite("\n")
swrite("-- Output directory " + output_dir + ":\n")
write_list(ls(output_dir))
write_list(ls(output_dir + '/*'))
swrite("\n")
# write meta data to sdterr
swrite('\n=== METADATA ===\n\n')
swrite("-- Current directory " + pwd() + ":\n")
try:
metadata = yaml.load(open('metadata', 'r'))
for key,value in metadata.items():
swrite(key + ': ')
swrite(str(value) + '\n')
except:
swrite("none\n");
swrite("-- Input directory " + input_dir + ":\n")
try:
metadata = yaml.load(open(os.path.join(input_dir, 'metadata'), 'r'))
for key,value in metadata.items():
swrite(key + ': ')
swrite(str(value) + '\n')
swrite("\n")
except:
swrite("none\n");
def handle(self, *args, **options):
reference = 0
# open file with data
directory = os.path.dirname(os.path.dirname(__file__))
path = os.path.join(directory, 'data', 'albums.yml')
with open(path, 'r') as file:
data = yaml.load(file)
albums = data['albums']
for album in albums:
# Create artists
artists = []
for artist in album['artists']:
try:
stored_artist = Artist.objects.get(name=artist)
lg.info('Artist found: %s'%stored_artist)
except ObjectDoesNotExist:
stored_artist = Artist.objects.create(name=artist)
lg.info('Artist created: %s'%stored_artist)
artists.append(stored_artist)
# Find or create album
try:
stored_album = Album.objects.get(title=album['title'])
lg.info('Album found: %s'%stored_album.title)
except ObjectDoesNotExist:
reference += 1
album = Album.objects.create(
title=album['title'],
reference=reference,
picture=album['picture']
)
album.artists = artists
lg.info('New album: %s'%stored_artist)
def __init__(self, conf_file):
f = open(conf_file, "r")
conf = yaml.load(f)
f.close()
cert_bin = self.__hex2bin(conf["cert_hex"])
priv = conf["priv_key"]
P_bin = self.__hex2bin(priv["P_hex"])
Q_bin = self.__hex2bin(priv["Q_hex"])
d_mod_p_1_bin = self.__hex2bin(priv["d_mod_p-1_hex"])
d_mod_q_1_bin = self.__hex2bin(priv["d_mod_q-1_hex"])
inv_q_mod_p_bin = self.__hex2bin(priv["inv_q_mod_p_hex"])
self.__cert_bin = cert_bin
self.__priv_key_P = int.from_bytes(P_bin, "big")
self.__priv_key_Q = int.from_bytes(Q_bin, "big")
self.__priv_key_d_mod_p_1 = int.from_bytes(d_mod_p_1_bin, "big")
self.__priv_key_d_mod_q_1 = int.from_bytes(d_mod_q_1_bin, "big")
self.__priv_key_inv_q_mod_p = int.from_bytes(inv_q_mod_p_bin, "big")
f = open("yaml/global.yaml", "r")
conf = yaml.load(f)
f.close()
global_constant_hex = re.sub(r'\s', "", conf["global_constant"])
self.__global_constant_bin = bytes.fromhex(global_constant_hex)
def __init__(self, conf_yaml):
f = open(conf_yaml, "r")
conf = yaml.load(f)
f.close()
def load(self, filename):
self.filename = filename
with open(filename, 'r') as stream:
self.config = yaml.load(stream)
def reload(self):
self.load(self.filename)
def load(self, name, force_reload_if_unmanaged=False):
if name in sys.modules and name not in self.plugins:
# we're getting an already loaded module, which we has not been
# loaded through PluginManager, return it from sys.modules and
# add it to our list
module = sys.modules[name]
if force_reload_if_unmanaged:
importlib.reload(module)
else:
module = importlib.import_module(name)
self.plugins.add(name)
return module
def load_yaml(self, job_yaml):
obj = yaml.load(job_yaml)
job_steps = []
for step in obj['steps']:
handler_name = list(step.keys())[0]
plugin = engine.plugins.get_job_handler(handler_name)
job_steps.append(plugin(step[handler_name]))
job = Job()
job.steps = job_steps
return job
def docs():
"""
Provides the manifest.json file for the 21 endpoint crawler.
"""
with open('./manifest.yaml', 'r') as f:
manifest = yaml.load(f)
return json.dumps(manifest)
def load_yaml_config(filepath):
with open(filepath) as f:
return yaml.load(f)
