def generate_rscfl_subsystems_header(json_fname, header_file):
"""Using the JSON list of subsystems, generate a header file that creates
a enum of subsystems.
Save this header file to $header_file
Args:
json_file: File object with a JSON list of subsystems.
header_file: File to write a C header file containing an enum of
possible subsystems.
"""
json_file = open(json_fname, 'r')
subsys_json = json.load(json_file)
subsystems = sorted(subsys_json.items(), key=lambda x: x[1]['id'])
template = jinja2.Template(RSCFL_SUBSYS_HEADER_TEMPLATE)
args = {}
args['subsystems'] = subsystems
args['ADDR_INVALID'] = ADDR_INVALID
args['ADDR_CALLQ'] = ADDR_CALLQ
args['ADDR_USER_SYSCALL'] = ADDR_USER_SYSCALL
args['ADDR_KERNEL_SYSCALL'] = ADDR_KERNEL_SYSCALL
header_file.write(template.render(args))
json_file.close()
python类Template()的实例源码
def write(self, file_path: Path, template: str, context: dict={}, preserve: bool = False):
"""Using a template file name it renders a template
into a file given a context
"""
if not context:
context = self.context
error = False
try:
self._write(file_path, template, context, preserve)
except TemplateSyntaxError as exc:
message = '{0}:{1} error: {2}'.format(exc.filename, exc.lineno, exc.message)
click.secho(message, fg='red')
error = True
except TemplateNotFound as exc:
message = '{0} error: Template not found'.format(exc.name)
click.secho(message, fg='red')
error = True
except TemplateError as exc:
message = 'error: {0}'.format(exc.message)
click.secho(message, fg='red')
error = True
if error and Generator.strict:
sys.exit(-1)
def create_cd(col_name, type, display_name):
url = 'https://api.kentik.com/api/v5/customdimension'
json_template = '''
{
"name": "{{ column }}",
"type": "{{ data_type }}",
"display_name": "{{ pretty_name }}"
}
'''
t = Template(json_template)
data = json.loads(t.render(column = col_name, data_type = type, pretty_name = display_name))
response = requests.post(url, headers=headers, data=data)
if response.status_code != 201:
print("Unable to create custom dimension column. Exiting.")
print("Status code: {}").format(response.status_code)
print("Error message: {}").format(response.json()['error'])
exit()
else:
print("Custom dimension \"{}\" created as id: {}").format(display_name, \
response.json()['customDimension']['id'])
return(response.json()['customDimension']['id'])
def load_template(template):
""" Try to read a file from a given path, if file
does not exist, load default one. """
file = None
try:
if template:
with open(template, "r") as f:
file = f.read()
except Exception as err:
print("Error: Your Template wasn't loaded", err,
"Loading Default Template", sep="\n")
finally:
if not file:
with open(DEFAULT_TEMPLATE, "r") as f:
file = f.read()
return file
def get_html(data):
template_raw = open('feed.tmpl', 'r').read()
for post in data:
if 'message' in post:
if (type(post['message']) is str):
post['message'] = fixnewlines(post['message'])
if 'flag' not in post :
post['message'] = enable_links(post['message'])
post['flag'] = 1
post['message'] = post['message'].replace("\"","'")
post['short_message'] = truncate(post['message'],150)
post['read_more'] = truncate_length(post['message'],150)
json.dump(data, open('docs/feed.json', 'w'))
template = Template(template_raw)
html = template.render(data=data)
# smart_str helps in unicode rendering
return smart_str(html)
def parse_settings(config: Any) -> Optional[Dict[str, Any]]:
ret = {
'webhook-url': config.get('slack-webhook-url'),
'tmpl-msg': config.get('slack-tmpl-msg'),
'tmpl-duration': config.get('slack-tmpl-duration', fallback=''),
'tmpl-url': config.get('slack-tmpl-url', fallback='')
} # type: Any
if not ret['webhook-url'] or not ret['tmpl-msg']:
log.debug('Slack settings missing, no slack notifications will be sent', 'NOTIFICATIONS')
ret = None
else:
log.debug('Valid slack notification settings found', 'NOTIFICATIONS')
ret['tmpl-msg'] = jinja2.Template(ret['tmpl-msg'])
if ret['tmpl-duration']:
ret['tmpl-duration'] = jinja2.Template(ret['tmpl-duration'])
if ret['tmpl-url']:
ret['tmpl-url'] = jinja2.Template(ret['tmpl-url'])
return ret
def parse_settings(config: Any) -> Optional[Dict[str, Any]]:
"""Parse clicksend sms settings.
Should only be called from sms.parse_settings.
"""
ret = {
'provider': 'clicksend',
'username': config.get('sms-clicksend-username'),
'api-key': config.get('sms-clicksend-api-key'),
'sender': config.get('sms-clicksend-sender'),
'tmpl': config.get('sms-tmpl'),
} # type: Any
if not ret['username'] or not ret['api-key'] or not ['sender'] or not ['tmpl']:
log.msg('SMS settings missing, no sms notifications will be sent', 'NOTIFICATIONS')
ret = None
else:
log.debug('Valid SMS notification settings found', 'NOTIFICATIONS')
ret['tmpl'] = jinja2.Template(ret['tmpl'])
return ret
def update(self, update_params: Dict[str, Any]) -> None:
async def _run(cur: sql.Cursor) -> None:
for param in ['name', 'description', 'active', 'cmdline_filename', 'cmdline_args_tmpl', 'description_tmpl']:
if param in update_params:
q = """update active_monitor_defs set %s=%%s where id=%%s""" % param
q_args = (update_params[param], self.id)
await cur.execute(q, q_args)
self.log_msg('updating monitor def')
if 'name' in update_params:
self.name = update_params['name']
if 'active' in update_params:
self.active = update_params['active']
if 'cmdline_filename' in update_params:
self.cmdline_filename = update_params['cmdline_filename']
if 'cmdline_args_tmpl' in update_params:
self.cmdline_args_tmpl = update_params['cmdline_args_tmpl']
self.jinja_cmdline_args = jinja2.Template(self.cmdline_args_tmpl)
if 'description_tmpl' in update_params:
self.description_tmpl = update_params['description_tmpl']
self.jinja_description_tmpl = jinja2.Template(self.description_tmpl)
self.tmpl_cache.flush_all()
await self.manager.dbcon.transact(_run)
def reply(self, response, **kwargs):
if self.is_CLI_agent(kwargs.get("agent", None)): # are we called from CLI
formatting = CLIFormatter.getAsDict()
# first pass: assemble
template = env.get_template("base.cli")
rendered = template.render(content=response)
# second pass, format!
formatTemp = Template(rendered)
return formatTemp.render(**formatting)
else:
formatting = HtmlFormatter.getAsDict()
template = env.get_template("base.html")
rendered = template.render(content=response, google_analytics=os.getenv("GOOGLE_ANALYTICS", Config.google_analytics))
formatTemp = Template(rendered)
return formatTemp.render(**formatting)
def discover_and_make_plots(destination_prefix, github_urlbase):
with open(os.path.join(os.path.dirname(__file__), 'index.tmpl.html')) as f:
index_template = Template(f.read())
benchmark_pages = defaultdict(list)
index_page = os.path.join(destination_prefix, 'index.html')
for root, dirs, files in os.walk(destination_prefix):
output_subdir = os.path.relpath(root, destination_prefix)
results_filename = os.path.join(root, 'results.json')
plot_filename = os.path.join(root, 'results.html')
github_url = github_urlbase + '/' + output_subdir
if os.path.exists(results_filename):
print(' Found: %s' % results_filename)
results = generate_plots(results_filename, plot_filename, github_url=github_url)
benchmark_pages[os.path.dirname(output_subdir)].append(dict(name=results['name'], path=os.path.join(output_subdir, 'results.html')))
# Generate index page
with open(index_page, 'w') as f:
f.write(index_template.render(sections=benchmark_pages))
def interpolate_value(value: str, context: dict):
"""Expand Jinja templating in the definitions."""
if type(value) == str and "{{" in value:
t = jinja2.Template(value, undefined=jinja2.StrictUndefined)
try:
v = t.render(**context)
except jinja2.exceptions.TemplateError as e:
raise RuntimeError("Could not expand template value: {}".format(value)) from e
# Jinja template rendering does not have explicit int support,
# so we have this hack in place
try:
v = int(v)
except ValueError:
pass
return v
else:
return value
def generate_contracts(token):
context = {
'TOKEN_CLASS_NAME': token.class_name,
'TOKEN_PUBLIC_NAME': token.public_name,
'CROWDSALE_CLASS_NAME': token.crowdsale_class_name,
'TOKEN_SYMBOL_NAME': token.symbol,
'TOKEN_DECIMALS': token.decimals
}
in_fname = os.path.join(settings.SOLIDITY_TEMPLATES_DIR, 'Token.sol.in')
out_fname = os.path.join(settings.SOLIDITY_CONTRACTS_DIR, token.class_name + '.sol')
with open(in_fname, 'r') as in_f, open(out_fname, 'w') as out_f:
template = Template(in_f.read())
out_f.write(template.render(**context))
in_fname = os.path.join(
settings.SOLIDITY_TEMPLATES_DIR, token.token_type + 'Crowdsale.sol.in'
)
out_fname = os.path.join(settings.SOLIDITY_CONTRACTS_DIR,
token.class_name + token.token_type + 'Crowdsale.sol')
with open(in_fname, 'r') as in_f, open(out_fname, 'w') as out_f:
template = Template(in_f.read())
out_f.write(template.render(**context))
def main():
parser = argparse.ArgumentParser()
parser.add_argument('-t', '--template', help='Template file', required=True)
parser.add_argument('-o', '--output', help='Output file', required=True)
parser.add_argument('-n', '--image-name', help='Image name', required=True)
parser.add_argument('-d', '--image-descriptors', help='Image descriptors', required=True)
args = parser.parse_args()
# Get the image descriptors json file
with open(args.image_descriptors, 'r') as f:
images = json.loads(f.read())['images']
if args.image_name not in images.keys():
print('%s does not exist in %s' % (args.image_name, args.image_descriptors))
sys.exit(-1)
with open(args.template) as fp:
template = jinja2.Template(fp.read())
context = images[args.image_name]
output = template.render(**context)
with open(args.output, 'w') as f:
f.write(output)
def gen():
TEMPLATE_FILE = os.path.join(os.path.dirname(__file__), "index.jinja2")
GENERATE_FILE = os.path.join(os.path.dirname(__file__), "gen.yaml")
CONFIG_FILE = os.path.join(os.path.dirname(__file__), "config.yaml")
INDEX_FILE = os.path.join(os.path.dirname(__file__), "web/index.html")
with open(TEMPLATE_FILE) as f:
template = jinja2.Template(f.read())
with open(GENERATE_FILE) as f:
gen = yaml.load(f.read())
with open(CONFIG_FILE) as f:
conf = yaml.load(f.read())
with open(INDEX_FILE, "w") as f:
f.write(template.render(title=conf["title"], update_time=gen["update_time"], entries=gen["entries"], victims=gen["fail_list"], subscriptions=sorted(conf["subscriptions"], key=lambda x: x["title"])))
return len(gen["fail_list"])
def tick(game, template, is_json, no_color):
if not game:
raise click.BadParameter('Missing required parameter "game"')
matches = download_history(game)
if is_json:
click.echo(json.dumps(list(matches), indent=2, sort_keys=True))
return
template = template if template else DEFAULT_TEMPLATE_RECAP
template = Template(template)
for m in matches:
if no_color or not COLOR_ENABLED: # if color is disabled just stdout
print_match(m, template)
continue
if m['t1_score'] > m['t2_score']:
m['t1'] = Fore.GREEN + m['t1'] + Fore.RESET
m['t2'] = Fore.RED + m['t2'] + Fore.RESET
else:
m['t2'] = Fore.GREEN + m['t2'] + Fore.RESET
m['t1'] = Fore.RED + m['t1'] + Fore.RESET
print_match(m, template)
def write_data(self, file_descr):
"""
Format data from self.data_df, write into file_descr (opened with opener).
"""
columns = self.data_df.columns
col0 = columns[0] # First column title (usually Database)
col1 = columns[1] # Second column title (usually Collection or Table)
tmpl_variables = OrderedDict()
for db in self.data_df[col0].unique():
tmpl_variables[db] = OrderedDict()
df_db = self.data_df.query('{} == @db'.format(col0)).iloc[:, 1:]
for col in df_db[col1].unique():
df_col = df_db.query('{} == @col'.format(col1)).iloc[:, 1:]
tmpl_variables[db][col] = df_col.values.tolist()
tmpl_filename = os.path.join(os.path.dirname(os.path.dirname(__file__)),
'resources', 'data_dict.tmpl')
with open(tmpl_filename) as tmpl_fd:
tmpl = jinja2.Template(tmpl_fd.read())
file_descr.write(tmpl.render(col_titles=list(self.data_df)[2:],
data=tmpl_variables))
def process_to_tempfile(filepath: str, context: Dict[str, Any]) -> str:
"""
renders the template in ``filepath`` using ``context`` through Jinja2. The result
is saved into a temporary file, which will be garbage collected automatically when the
program exits.
:return: the full path of the temporary file containing the result
"""
outfd, ofname = tempfile.mkstemp()
with open(outfd, mode="w") as outf:
with open(filepath) as inf:
tplstr = inf.read()
tpl = jinja2.Template(tplstr)
outf.write(tpl.render(context))
import gopythongo.main
gopythongo.main.tempfiles.append(ofname)
return ofname
def create_template():
conf_list = []
with open("huge_file.json") as json_data:
info = json.load(json_data)
t = jinja2.Template(template)
asn_list = info['asns'].keys()
for i in asn_list:
abcd = info['asns'][i]
out = t.render(data=abcd, seq_num=1)
conf_list += ([y for y in (x.strip() for x in out.splitlines()) if y])
return conf_list
def run(self, **kwargs):
"""Run madx as a subprocess."""
self._input += self._syntax['stop']
template_input = jinja2.Template(self._input).render(kwargs.get("context", {}))
if kwargs.get("debug", False) >= 2:
print(template_input)
if self._get_exec() is None:
raise MadxException("Can't run MADX if no valid path and executable are defined.")
p = sub.Popen([self._get_exec()],
stdin=sub.PIPE,
stdout=sub.PIPE,
stderr=sub.STDOUT,
cwd=".",
shell=True
)
self._output = p.communicate(input=template_input.encode())[0].decode()
self._warnings = [line for line in self._output.split('\n') if re.search('warning|fatal', line)]
self._fatals = [line for line in self._output.split('\n') if re.search('fatal', line)]
self._last_context = kwargs.get("context", {})
if kwargs.get('debug', False):
print(self._output)
return self
def __init__(self, base_log_folder, filename_template):
"""
:param base_log_folder: Base log folder to place logs.
:param filename_template: template filename string
"""
super(FileProcessorHandler, self).__init__()
self.handler = None
self.base_log_folder = base_log_folder
self.dag_dir = os.path.expanduser(conf.get('core', 'DAGS_FOLDER'))
self.filename_template = filename_template
self.filename_jinja_template = None
if "{{" in self.filename_template: #jinja mode
self.filename_jinja_template = Template(self.filename_template)
self._cur_date = datetime.today()
if not os.path.exists(self._get_log_directory()):
os.makedirs(self._get_log_directory())
self._symlink_latest_log_directory()
def _render_url_params(endpoint: APITestEndPoint, template: Template, payload_content: dict):
"""
Generates tests for parameters with information in URL, like:
wwww.mysite.com/index.php?id=1&page=abc
This function will generate tests for params:
- id
- page
"""
content_type = APITestContentType(endpoint.request.body.content_type).value
url = endpoint.request.url
scheme, netloc, url, params, query, fragment = urlparse(url)
base_url = urljoin("%s://%s" % (scheme, netloc), url)
url_params = form_content2dict(query) if len(query) != 0 else {}
template.render(url=base_url,
method=endpoint.request.method,
content_type=content_type,
body_content=endpoint.request.body.value,
url_params=url_params,
payloads=payload_content)
def render_confs():
"""
Renders server configurations.
"""
require('settings', provided_by=['production', 'staging'])
with settings(warn_only=True):
local('mkdir confs/rendered')
# Copy the app_config so that when we load the secrets they don't
# get exposed to other management commands
context = copy.copy(app_config.__dict__)
context.update(app_config.get_secrets())
for service, remote_path, extension in app_config.SERVER_SERVICES:
template_path = _get_template_conf_path(service, extension)
rendered_path = _get_rendered_conf_path(service, extension)
with open(template_path, 'r') as read_template:
with open(rendered_path, 'w') as write_template:
payload = Template(read_template.read())
write_template.write(payload.render(**context))
def read(self):
data = self.load()
try:
data = unicode(data)
template = Template(data)
literal = template.render(CONTEXT)
# TODO: might be useful to write the literal result to a file for debugging
location = self.location
if isinstance(location, basestring) and location.endswith('.jinja'):
# Use reader based on the location with the ".jinja" prefix stripped off
location = location[:-6]
next_reader = self.context.reading.reader_source.get_reader(
self.context, LiteralLocation(literal, name=location), LiteralLoader(literal))
else:
# Use reader for literal loader
next_reader = self.context.reading.reader_source.get_reader(
self.context, LiteralLocation(literal), LiteralLoader(literal))
return next_reader.read()
except Exception as e:
raise ReaderSyntaxError(u'Jinja: {0}'.format(e), cause=e)
def setup_load_process(self, lib, cell, hist_name, outputs, precision):
# type: (str, str, str, Dict[str, str], int) -> ProcInfo
init_file = self.sim_config['init_file']
view = self.sim_config['view']
# create temporary save directory and log/script names
save_dir = bag.io.make_temp_dir(prefix='%s_data' % hist_name, parent_dir=self.tmp_dir)
log_fname = os.path.join(save_dir, 'ocn_output.log')
script_fname = os.path.join(save_dir, 'run.ocn')
# setup ocean load script
script = Template(load_script).render(lib=lib,
cell=cell,
view=view,
init_file=init_file,
save_dir='{{ save_dir }}',
precision=precision,
hist_name=hist_name,
outputs=outputs,
)
bag.io.write_file(script_fname, script + '\n')
# launch ocean
return self._get_ocean_info(save_dir, script_fname, log_fname)
def setup_export_schematic(self, lib_name, cell_name, out_file, view_name='schematic', params=None):
# type: (str, str, str, str, Optional[Dict[str, Any]]) -> ProcInfo
out_file = os.path.abspath(out_file)
run_dir = os.path.dirname(out_file)
out_name = os.path.basename(out_file)
log_file = os.path.join(run_dir, 'schematic_export.log')
# fill in stream out configuration file.
content = Template(sch_template).render(lib_name=lib_name,
cell_name=cell_name,
view_name=view_name,
output_name=out_name,
source_added_file=self._source_added_file,
run_dir=run_dir,
)
# create configuration file.
config_fname = os.path.join(run_dir, 'si.env')
write_file(config_fname, content)
# run command
cmd = ['si', run_dir, '-batch', '-command', 'netlist']
return cmd, log_file, None, os.environ['BAG_WORK_DIR']
def save_parameter_from_order_in_memory(cls, order_parameters):
"""
Save key from the temp dict (where parameters loaded from the voice order where placed temporary)
into the memory dict
:param order_parameters: dict of key to save. {'key_name_in_memory': 'key_name_in_temp_dict'}
:return True if a value has been saved in the kalliope memory
"""
order_saved = False
if order_parameters is not None:
logger.debug("[Cortex] save_parameter_from_order_in_memory - User want to save: %s" % order_parameters)
logger.debug("[Cortex] save_parameter_from_order_in_memory - Available parameters in orders: %s"
% cls.temp)
for key, value in order_parameters.items():
# ask the cortex to save in memory the target "key" if it was in the order
if Utils.is_containing_bracket(value):
# if the key exist in the temp dict we can load it with jinja
value = jinja2.Template(value).render(Cortex.temp)
if value:
Cortex.save(key, value)
order_saved = True
return order_saved
def __init__(self, *args, **kwargs):
""" Creates a new Jinja2 templating engine.
"""
# Call the parent
super().__init__(*args, **kwargs)
# Create a Jinja2 environment. We could use jinja2.Template().render()
# directly, but having an environment gives us more control over, e.g.,
# custom filters.
self.env = jinja2.Environment()
# Registering custom filters is described here:
# http://jinja.pocoo.org/docs/dev/api/#custom-filters
self.register_custom_filters(self.env)
# Built-in Jinja2 filters are listed here:
# http://jinja.pocoo.org/docs/dev/templates/#builtin-filters
###########################################################################
def hall_of_fame(msg):
user = models.User(msg.author.name)
if user.is_registered():
message = "Donation Tip to " + config.bot_name + " : "
donator_list = {}
hist = models.HistoryStorage.get_user_history(config.bot_name)
message += "\n\nUser|Donation Ammount\n"
message += "---|---\n"
for tip in hist:
if tip["sender"] in donator_list.keys():
donator_list[tip["sender"]] = float(donator_list[tip["sender"]]) + tip['amount']
else:
donator_list[tip["sender"]] = tip['amount']
for donor in sorted(donator_list.items(), key=lambda user: user[1], reverse=True):
message += "%s|%s\n" % (donor[0], str(donor[1]))
user.send_private_message("Hall Of Fame", message)
else:
bot_logger.logger.info('user %s not registered (command : hall_of_fame) ' % user.username)
msg.reply(Template(lang.message_need_register + lang.message_footer).render(username=user.username))
def donate(msg, tx_queue, failover_time):
user = models.User(msg.author.name)
if user.is_registered():
split_message = msg.body.lower().strip().split()
donate_index = split_message.index('+donate')
amount = split_message[donate_index + 1]
if utils.check_amount_valid(amount) and split_message[donate_index + 2] == 'doge':
crypto.tip_user(user.username.address, models.User(config.bot_name).address, amount, tx_queue,
failover_time)
models.HistoryStorage.add_to_history(msg.author.name, msg.author.name, config.bot_name, amount, "donate")
else:
bot_logger.logger.info(lang.message_invalid_amount)
user.send_private_message('invalid amount', lang.message_invalid_amount)
else:
bot_logger.logger.info('user %s not registered (command : donate) ' % user.username)
msg.reply(Template(lang.message_need_register + lang.message_footer).render(username=user.username))
def get_template(not_type, name=None):
if name is None:
temp_name = 'default.html'
else:
temp_name = name + '.html'
temp_file = os.path.join(sc_dir(),
'templates',
not_type,
temp_name)
try:
temp_str = open(temp_file, 'r')
except Exception:
temp_file = os.path.join(sc_dir(),
'templates',
not_type,
'default.html')
temp_str = open(temp_file, 'r')
template = Template(temp_str.read())
temp_str.close()
return template
