def _predict(predictors: Dict[str, str]):
def predict_inner(args: argparse.Namespace) -> None:
predictor = _get_predictor(args, predictors)
output_file = None
if args.silent and not args.output_file:
print("--silent specified without --output-file.")
print("Exiting early because no output will be created.")
sys.exit(0)
# ExitStack allows us to conditionally context-manage `output_file`, which may or may not exist
with ExitStack() as stack:
input_file = stack.enter_context(args.input_file) # type: ignore
if args.output_file:
output_file = stack.enter_context(args.output_file) # type: ignore
_run(predictor, input_file, output_file, args.batch_size, not args.silent, args.cuda_device)
return predict_inner
python类ExitStack()的实例源码
def adversarial_discriminator(net, layers, scope='adversary', leaky=False):
if leaky:
activation_fn = tflearn.activations.leaky_relu
else:
activation_fn = tf.nn.relu
with ExitStack() as stack:
stack.enter_context(tf.variable_scope(scope))
stack.enter_context(
slim.arg_scope(
[slim.fully_connected],
activation_fn=activation_fn,
weights_regularizer=slim.l2_regularizer(2.5e-5)))
for dim in layers:
net = slim.fully_connected(net, dim)
net = slim.fully_connected(net, 2, activation_fn=None)
return net
def search(timeout):
'''
Search for devices implementing WANCommonInterfaceConfig on the network.
Search ends the specified number of seconds after the last result (if any) was received.
Returns an iterator of root device URLs.
'''
with contextlib.ExitStack() as stack:
sockets = []
sockets.append(stack.enter_context(socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP)))
sockets.append(stack.enter_context(socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)))
for s in sockets:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
if s.family == socket.AF_INET6:
s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
with concurrent.futures.ThreadPoolExecutor(len(sockets)) as ex:
return itertools.chain.from_iterable(ex.map(lambda s: search_socket(s, timeout, ns['i']), sockets))
def write_split_emote(cls):
retweet_re = re.compile(r'RT\s*"?[@?][a-zA-Z0-9_]+:?')
with open('emote/all.txt', encoding='utf-8') as in_sr, ExitStack() as stack:
out_srs = {
name: stack.enter_context(open('emote/class_{}.txt'.format(name), 'w', encoding='utf-8'))
for name in ['pos', 'neg']
}
for i, line in enumerate(in_sr):
if retweet_re.search(line):
continue
counts = [0, 0, 0]
for match in cls.emoticon_re.finditer(line):
counts[Emoticons.assess_match(match)] += 1
label = None
if counts[0] > 0 and counts[1] == 0 and counts[2] == 0:
label = 0
elif counts[0] == 0 and counts[1] == 0 and counts[2] > 0:
label = 2
if label is not None:
out_srs[label].write(cls.emoticon_re.sub(' ', line).strip() + '\n')
def post(url: str, data: object, file_paths: List[str] = None, username: str="", password: str=""):
request = {
"url": url,
"data": json.dumps(data, cls=JSONFloatEncoder)
}
if username:
if not password:
password = getpass.getpass("Enter password for '{}': ".format(username))
request["auth"] = (username, password)
with ExitStack() as es:
if file_paths:
request["data"] = {"data": request["data"]}
request["files"] = [__create_file_tuple(path, es) for path in file_paths]
response = requests.post(**request)
response.raise_for_status()
def test_exit_stack_exception_propagate():
h1 = mock.MagicMock()
h2 = mock.MagicMock()
v1 = mock.MagicMock()
v2 = mock.MagicMock()
error = ValueError('FUUU')
with pytest.raises(ValueError) as exc:
with ExitStack() as stack:
v = stack.enter_context(AutoClose(h1, v=v1))
assert v is v1
v = stack.enter_context(AutoClose(h2, v=v2))
assert v is v2
raise error
assert exc.value is error
h2.close.assert_called_once_with(ValueError, error, mock.ANY)
h1.close.assert_called_once_with(ValueError, error, mock.ANY)
def __call__(self):
status = 1
with log_exception(status=1):
args = self.parser.parse_args()
log_args(args)
config.log_cached()
logger = getLogger('django')
with ExitStack() as stack:
if self._pid_file:
stack.enter_context(pid_file(dirname=config.PID_DIR, max_age=self._pid_file_max_age))
if self._stoppable:
self._stoppable_instance = stoppable()
stack.enter_context(self._stoppable_instance)
status = self.run(args, logger) or 0
sys.exit(status)
def setUp(self):
super().setUp()
self._resources = ExitStack()
self.addCleanup(self._resources.close)
# Capture builtin print() output.
self._stdout = StringIO()
self._stderr = StringIO()
self._resources.enter_context(
patch('argparse._sys.stdout', self._stdout))
# Capture stderr since this is where argparse will spew to.
self._resources.enter_context(
patch('argparse._sys.stderr', self._stderr))
# Set up a few other useful things for these tests.
self._resources.enter_context(
patch('ubuntu_image.__main__.logging.basicConfig'))
self.model_assertion = resource_filename(
'ubuntu_image.tests.data', 'model.assertion')
self.classic_gadget_tree = resource_filename(
'ubuntu_image.tests.data', 'gadget_tree')
def test_duplicate_volume_name(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first:
bootloader: u-boot
structure:
- name: one
type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
second:
structure:
- name: two
type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
first:
structure:
- name: three
type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
""")
self.assertEqual(str(cm.exception), 'Duplicate key: first')
def test_bad_hybrid_volume_type_2(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
bootloader: u-boot
structure:
- type: 00000000-0000-0000-0000-0000deadbeef,\
00000000-0000-0000-0000-0000deadbeef
size: 400M
""")
self.assertEqual(
str(cm.exception),
'Invalid gadget.yaml @ volumes:first-image:structure:0:type')
def test_volume_filesystem_bad(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
bootloader: u-boot
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
filesystem: zfs
""")
self.assertEqual(
str(cm.exception),
("Invalid gadget.yaml value 'zfs' @ "
'volumes:<volume name>:structure:<N>:filesystem'))
def test_volume_structure_role_system_data_bad_label(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
bootloader: u-boot
structure:
- type: 00000000-0000-0000-0000-0000feedface
size: 200
role: system-data
filesystem-label: foobar
""")
self.assertEqual(
str(cm.exception),
('`role: system-data` structure must have an implicit label, '
"or 'writable': foobar"))
def test_volume_structure_type_role_conflict_1(self):
# type:none means there's no partition, so you can't have a role of
# system-{boot,data}.
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""
volumes:
first-image:
schema: mbr
bootloader: u-boot
structure:
- type: ef
size: 100
role: mbr
second-image:
structure:
- type: bare
size: 200
role: system-boot
""")
self.assertEqual(
str(cm.exception),
'Invalid gadget.yaml: structure role/type conflict')
def test_volume_structure_type_role_conflict_2(self):
# type:none means there's no partition, so you can't have a role of
# system-{boot,data}.
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""
volumes:
first-image:
schema: mbr
bootloader: u-boot
structure:
- type: ef
size: 100
role: mbr
second-image:
structure:
- type: bare
size: 200
role: system-data
""")
self.assertEqual(
str(cm.exception),
'Invalid gadget.yaml: structure role/type conflict')
def test_volume_structure_mbr_conflicting_id(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: mbr
bootloader: u-boot
structure:
- type: ef
role: mbr
size: 100
id: 00000000-0000-0000-0000-0000deadbeef
""")
self.assertEqual(
str(cm.exception),
'mbr structures must not specify partition id')
def test_volume_structure_mbr_conflicting_filesystem(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: mbr
bootloader: u-boot
structure:
- type: ef
role: mbr
size: 100
filesystem: ext4
""")
self.assertEqual(
str(cm.exception),
'mbr structures must not specify a file system')
def test_volume_special_type_mbr_and_role(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: mbr
bootloader: u-boot
structure:
- type: mbr
role: mbr
size: 100
""")
self.assertEqual(
str(cm.exception),
'Type mbr and role fields assigned at the same time, please use '
'the mbr role instead')
def test_volume_special_type_mbr_and_filesystem(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: mbr
bootloader: u-boot
structure:
- type: mbr
size: 100
filesystem: ext4
""")
self.assertEqual(
str(cm.exception),
'mbr structures must not specify a file system')
def test_bad_schema(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: bad
bootloader: u-boot
structure:
- type: ef
size: 400M
""")
self.assertEqual(
str(cm.exception),
"Invalid gadget.yaml value 'bad' @ volumes:<volume name>:schema")
def test_mbr_with_bogus_type(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: mbr
bootloader: u-boot
structure:
- type: 801
size: 400M
""")
self.assertEqual(
str(cm.exception),
'Invalid gadget.yaml @ volumes:first-image:structure:0:type')
def test_bad_bootloader(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: gpt
bootloader: u-boat
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
""")
self.assertEqual(
str(cm.exception),
("Invalid gadget.yaml value 'u-boat' @ "
'volumes:<volume name>:bootloader'))
def test_missing_bootloader_multiple_volumes(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: gpt
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
second-image:
schema: gpt
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
third-image:
schema: gpt
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
""")
self.assertEqual(str(cm.exception), 'No bootloader structure named')
def test_bad_volume_id(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: gpt
bootloader: u-boot
id: 3g
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
""")
self.assertEqual(str(cm.exception),
'Invalid gadget.yaml @ volumes:first-image:id')
def test_bad_integer_volume_id(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: gpt
bootloader: u-boot
id: 801
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
""")
self.assertEqual(str(cm.exception),
'Invalid gadget.yaml @ volumes:first-image:id')
def test_disallow_hybrid_volume_id(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
bootloader: u-boot
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
id: 80,00000000-0000-0000-0000-0000deadbeef
""")
self.assertEqual(
str(cm.exception),
'Invalid gadget.yaml @ volumes:first-image:structure:0:id')
def test_volume_offset_write_relative_syntax_error(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: gpt
bootloader: u-boot
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
offset-write: some_label%2112
""")
self.assertEqual(
str(cm.exception),
('Invalid gadget.yaml @ '
'volumes:first-image:structure:0:offset-write'))
def test_volume_offset_write_larger_than_32bit(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: gpt
bootloader: u-boot
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
offset-write: 8G
""")
self.assertEqual(
str(cm.exception),
('Invalid gadget.yaml @ '
'volumes:first-image:structure:0:offset-write'))
def test_volume_offset_write_is_4G(self):
# 4GiB is just outside 32 bits.
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
schema: gpt
bootloader: u-boot
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
offset-write: 4G
""")
self.assertEqual(
str(cm.exception),
('Invalid gadget.yaml @ '
'volumes:first-image:structure:0:offset-write'))
def test_bad_hybrid_volume_type_2(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
bootloader: u-boot
structure:
- type: 00000000-0000-0000-0000-0000deadbeef,\
00000000-0000-0000-0000-0000deadbeef
size: 400M
""")
self.assertEqual(
str(cm.exception),
'Invalid gadget.yaml @ volumes:first-image:structure:0:type')
def test_volume_filesystem_bad(self):
with ExitStack() as resources:
cm = resources.enter_context(
self.assertRaises(GadgetSpecificationError))
parse("""\
volumes:
first-image:
bootloader: u-boot
structure:
- type: 00000000-0000-0000-0000-0000deadbeef
size: 400M
filesystem: zfs
""")
self.assertEqual(
str(cm.exception),
("Invalid gadget.yaml value 'zfs' @ "
'volumes:<volume name>:structure:<N>:filesystem'))
