def load(stream, filename):
if filename.endswith('.cdg'):
import ubjson
cg = ubjson.load(stream)
elif filename.endswith('.json'):
import json
cg = json.load(stream)
elif filename.endswith('.yaml'):
import yaml
try:
from yaml import CLoader as Loader
except ImportError:
from yaml import Loader
cg = yaml.load(stream, Loader=Loader)
else:
raise ValueError('Unhandled file type: %s' % filename)
graph = create(filename)
for (name, props) in cg['functions'].items():
graph.add_node(name)
if 'attributes' in props:
graph.node[name].update(props['attributes'])
if 'calls' in props:
calls = props['calls']
if calls:
for target in calls:
graph.add_edge(name, target, kind = EdgeKind.Call)
if 'flows' in props:
flows = props['flows']
if flows:
for source in flows:
graph.add_edge(source, name, kind = EdgeKind.Flow)
return graph
python类CLoader()的实例源码
def __init__(self, args, out_postfix="_out", with_video_output=True):
self.global_video_offset = 0
self.flip_video = False
self.datapath = "./"
self.__dict__.update(vars(args))
self.writer = None
if os.path.exists("settings.yaml"):
stream = open("settings.yaml", mode='r')
self.settings = load(stream, Loader=Loader)
stream.close()
self.datapath = self.settings['datapath'].replace("<current_user>", getuser())
print("Processing path: ", self.datapath)
if 'raw_options' in self.settings:
raw_options = self.settings['raw_options']
if self.in_video in raw_options:
self.global_video_offset = raw_options[args.in_video]['global_offset']
self.flip_video = raw_options[args.in_video]['flip']
self.cap = None
self.reload_video()
print("Processing video file {:s}.".format(self.in_video))
last_frame = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT) - 1)
if self.end_with == -1:
self.end_with = last_frame
else:
if self.end_with > last_frame:
print(("Warning: specified end frame ({:d}) is beyond the last video frame" +
" ({:d}). Stopping after last frame.")
.format(self.end_with, last_frame))
self.end_with = last_frame
print("Frame range: {:d}--{:d}".format(self.start_from, self.end_with))
if with_video_output:
if self.out_video == "":
self.out_video = args.in_video[:-4] + "_" + out_postfix + ".mp4"
self.writer = cv2.VideoWriter(os.path.join(self.datapath, self.out_video),
cv2.VideoWriter_fourcc('X', '2', '6', '4'),
self.cap.get(cv2.CAP_PROP_FPS),
(int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))),
True)
self.writer.set(cv2.VIDEOWRITER_PROP_NSTRIPES, cpu_count())
else:
self.writer = None
self.frame = None
self.cur_frame_number = None
def process_arguments(program_arguments_enum, program_help_description):
argproc = ArgumentProcessor(program_arguments_enum)
defaults = argproc.generate_defaults_dict()
conf_parser = \
argproc.generate_parser(defaults, console_only=True, description=
"Test stereo algorithms on two image files.")
# ============== STORAGE/RETRIEVAL OF CONSOLE SETTINGS ===========================================#
args, remaining_argv = conf_parser.parse_known_args()
defaults[ArgumentProcessor.save_settings_name] = args.save_settings
if args.settings_file:
defaults[ArgumentProcessor.settings_file_name] = args.settings_file
if os.path.isfile(args.settings_file):
file_stream = open(args.settings_file, "r", encoding="utf-8")
config_defaults = load(file_stream, Loader=Loader)
file_stream.close()
if config_defaults:
for key, value in config_defaults.items():
defaults[key] = value
else:
if not args.save_settings:
raise ValueError("Settings file not found at: {0:s}".format(args.settings_file))
parser = argproc.generate_parser(defaults, parents=[conf_parser])
args = parser.parse_args(remaining_argv)
# process "special" setting values
if args.settings_file and os.path.isfile(args.settings_file):
for key in args.__dict__.keys():
if key in argproc.setting_file_location_args and args.__dict__[key] == \
Argument.setting_file_location_wildcard:
args.__dict__[key] = os.path.dirname(args.settings_file)
# save settings if prompted to do so
if args.save_settings and args.settings_file:
setting_dict = vars(args)
file_stream = open(args.settings_file, "w", encoding="utf-8")
file_name = setting_dict[ArgumentProcessor.save_settings_name]
del setting_dict[ArgumentProcessor.save_settings_name]
del setting_dict[ArgumentProcessor.settings_file_name]
dump(setting_dict, file_stream, Dumper=Dumper, indent=3, default_flow_style=False)
file_stream.close()
setting_dict[ArgumentProcessor.save_settings_name] = file_name
setting_dict[ArgumentProcessor.settings_file_name] = True
return args
def load_aten_declarations(path):
with open(path, 'r') as f:
declarations = yaml.load(f, Loader=Loader)
# enrich declarations with additional information
for declaration in declarations:
for arg in declaration['arguments']:
simple_type = arg['type']
simple_type = simple_type.replace(' &', '').replace('const ', '')
simple_type = simple_type.replace('Generator *', 'Generator')
arg['simple_type'] = simple_type
declaration['formals'] = [arg['type'] + ' ' + arg['name']
for arg in declaration['arguments']]
declaration['args'] = [arg['name'] for arg in declaration['arguments']]
declaration['api_name'] = declaration['name']
declaration['return_type'] = format_return_type(declaration['returns'])
declaration['base_name'] = declaration['name']
# Compute the Python function prototype for argument parsing
typed_args = []
positional = True
for arg in declaration['arguments']:
if arg.get('kwarg_only', False) and positional:
typed_args.append('*')
positional = False
typename = arg['simple_type']
if arg.get('is_nullable'):
typename = '{}?'.format(typename)
if arg.get('size') is not None:
typename = '{}[{}]'.format(typename, arg['size'])
param = typename + ' ' + arg['name']
default = None
if arg.get('default') is not None:
default = arg['default']
if default == 'nullptr' or default == '{}':
default = 'None'
if arg.get('python_default_init') is not None:
default = 'None'
if default is not None:
param += '=' + str(default)
typed_args.append(param)
# Python function prototype.
# This is the string that we give to FunctionParameter, which is
# then parsed into the actual structure which we do parsing
# with.
declaration['typed_args'] = typed_args
declaration['prototype'] = FUNCTION_PROTOTYPE.substitute(declaration)
return declarations
