Python Examples of contextlib.nullcontext (2023)

def __init__(self, fileobj, filename=None): super(tzfile, self).__init__() file_opened_here = False if isinstance(fileobj, string_types): self._filename = fileobj fileobj = open(fileobj, 'rb') file_opened_here = True elif filename is not None: self._filename = filename elif hasattr(fileobj, "name"): self._filename = fileobj.name else: self._filename = repr(fileobj) if fileobj is not None: if not file_opened_here: fileobj = _nullcontext(fileobj) with fileobj as file_stream: tzobj = self._read_tzfile(file_stream) self._set_tzdata(tzobj) 

def __init__(self, fileobj): global rrule from dateutil import rrule if isinstance(fileobj, string_types): self._s = fileobj # ical should be encoded in UTF-8 with CRLF fileobj = open(fileobj, 'r') else: self._s = getattr(fileobj, 'name', repr(fileobj)) fileobj = _nullcontext(fileobj) self._vtz = {} with fileobj as fobj: self._parse_rfc(fobj.read()) 

def check_model_compatibility(self, backend_config, save, load): C = self.n_channels // self.groups old_model = { 'avg_mean': numpy.random.uniform( -1, 1, (C,)).astype(self.dtype), 'avg_projection': numpy.random.uniform( 0.5, 1, (C, C)).astype(self.dtype), 'N': numpy.array(0) } save(self.temp_file_path, old_model) model = links.DecorrelatedBatchNormalization( self.n_channels, groups=self.groups, dtype=self.dtype) model.to_device(backend_config.device) with ( testing.assert_warns(UserWarning) if self.groups != 1 else nullcontext()): load(self.temp_file_path, model) x = numpy.random.rand(5, self.n_channels, 2).astype(self.dtype) x = backend_config.get_array(x) with chainer.using_config('train', False): model(x) model(x) 

def __init__( self, mode: int, ctx: moderngl.Context, prog: moderngl.Program, dtype: np.dtype, draw_context: ContextManager = nullcontext(), capacity: int = 256, index_capacity: int = 512): self.mode = mode self.ctx = ctx self.prog = prog self.dtype = dtype_to_moderngl(dtype) self.allocs: Dict[int, Tuple[slice, np.ndarray]] = {} self.verts = MemoryBackedBuffer(ctx, capacity, dtype) self.indexes = IndexBuffer(ctx) self.draw_context = draw_context self.dirty = False 

def iteration(self, data: Tuple[torch.Tensor, torch.Tensor]) -> Mapping[str, torch.Tensor]: input, target = data context = torch.cuda.amp.autocast if self._use_amp else contextlib.nullcontext with context(): output = self.model(input) loss = self.loss_f(output, target) if self.is_train: self.optimizer.zero_grad() if self._use_amp: self.scaler.scale(loss).backward() self.scaler.step(self.optimizer) self.scaler.update() else: loss.backward() self.optimizer.step() return TensorMap(loss=loss, output=output) 

def __init__(self, fileobj, filename=None): super(tzfile, self).__init__() file_opened_here = False if isinstance(fileobj, string_types): self._filename = fileobj fileobj = open(fileobj, 'rb') file_opened_here = True elif filename is not None: self._filename = filename elif hasattr(fileobj, "name"): self._filename = fileobj.name else: self._filename = repr(fileobj) if fileobj is not None: if not file_opened_here: fileobj = _nullcontext(fileobj) with fileobj as file_stream: tzobj = self._read_tzfile(file_stream) self._set_tzdata(tzobj) 

def __init__(self, fileobj): global rrule from dateutil import rrule if isinstance(fileobj, string_types): self._s = fileobj # ical should be encoded in UTF-8 with CRLF fileobj = open(fileobj, 'r') else: self._s = getattr(fileobj, 'name', repr(fileobj)) fileobj = _nullcontext(fileobj) self._vtz = {} with fileobj as fobj: self._parse_rfc(fobj.read()) 

def get_psl_context_manager(self, side_effect_parameter): if side_effect_parameter is None: return nullcontext() if callable(side_effect_parameter): side_effect = side_effect_parameter else: side_effect = partial( self._mock_get_public_suffix, public_suffixes=[side_effect_parameter] if not isinstance(side_effect_parameter, list) else list(side_effect_parameter) ) return mock.patch.object(psl, 'get_public_suffix', side_effect=side_effect) 

def get_distribution_scope(batch_size): if num_gpus() > 1: strategy = tf.distribute.MirroredStrategy() assert ( batch_size % strategy.num_replicas_in_sync == 0 ), f"Batch size {batch_size} cannot be divided onto {num_gpus()} GPUs" distribution_scope = strategy.scope else: if sys.version_info >= (3, 7): distribution_scope = contextlib.nullcontext else: distribution_scope = contextlib.suppress return distribution_scope() 

def _null_context() -> Iterator[None]: # TODO(b/154533346) # This should move to `contextlib.nullcontext` once TFF's minimum # Python version moves up to 3.7, yield None 

def forward(self, link, inputs, backend_config): x, = inputs with chainer.using_config('train', not self.test): y = link(x) return y,# TODO(kataoka) Use `contextlib.nullcontext` if Python 3.7 or higher is assumed 

def nullcontext(): yield 

def nullcontext(): yield 

def __init__(self, length, multithread_safe=False, do_pickle=False): self._multithread_safe = multithread_safe self.length = length assert self.length > 0 if self._multithread_safe: self.lock = threading.Lock() else: # Use contextlib.nullcontext() when Python 3.6 is dropped. self.lock = contextlib.suppress() self.data = [None for _ in range(self.length)] 

def nullcontext(obj): yield obj 

def get_distribution_scope(batch_size): if num_gpus() > 1: strategy = tf.distribute.MirroredStrategy() assert ( batch_size % strategy.num_replicas_in_sync == 0 ), f"Batch size {batch_size} cannot be divided onto {num_gpus()} GPUs" distribution_scope = strategy.scope else: if sys.version_info >= (3, 7): distribution_scope = contextlib.nullcontext else: distribution_scope = contextlib.suppress return distribution_scope() 

def nullcontext(): yield None 

def nullcontext(enter_result=None): """Backport of the Python >=3.7 trivial context manager. See `the Python documentation <https://docs.python.org/3/library/contextlib.html#contextlib.nullcontext>`_ for details. """ try: yield enter_result finally: pass 

def read_until_prompt( self, prompt: typing.Optional[ConvenientSearchString] = None, timeout: typing.Optional[float] = None, ) -> str: """ Read until prompt is detected. Read from the channel until the configured prompt string is detected. All data captured up until the prompt is returned, decoded as UTF-8. If ``prompt`` is ``None``, the prompt which was set using :py:meth:`tbot.machine.channel.Channel.with_prompt` is used. :param ConvenientSearchString prompt: The prompt to read up to. It must appear as the very last readable data in the channel's data stream. See :ref:`channel_search_string` for more info about which types can be passed for this parameter. :param float timeout: Optional timeout. If ``timeout`` is set and expires before the prompt was detected, ``read_until_prompt`` raises an execption. :rtype: str :returns: UTF-8 decoded string of all bytes read up to the prompt. """ ctx: typing.ContextManager[typing.Any] if prompt is not None: ctx = self.with_prompt(prompt) else: # contextlib.nullcontext() would be a better fit here but sadly it # is only available in 3.7+ ctx = contextlib.ExitStack() buf = bytearray() with ctx: for new in self.read_iter(timeout=timeout): buf += new if isinstance(self.prompt, bytes): if buf.endswith(self.prompt): return ( buf[: -len(self.prompt)] .decode("utf-8", errors="replace") .replace("\r\n", "\n") .replace("\n\r", "\n") ) elif isinstance(self.prompt, BoundedPattern): match = self.prompt.pattern.search(buf) if match is not None: return ( buf[: match.span()[0]] .decode("utf-8", errors="replace") .replace("\r\n", "\n") .replace("\n\r", "\n") ) raise RuntimeError("unreachable") # }}} # miscellaneous {{{ 

def test_single_case(test_name: str, test_input_path: pathlib.Path, test_output_path: Optional[pathlib.Path], *, lock: Optional[threading.Lock] = None, args: 'argparse.Namespace') -> Dict[str, Any]: # print the header earlier if not in parallel if lock is None: log.emit('') log.info('%s', test_name) # run the binary with test_input_path.open() as inf: info, proc = utils.exec_command(args.command, stdin=inf, timeout=args.tle, gnu_time=args.gnu_time) # TODO: the `answer` should be bytes, not str answer = (info['answer'] or b'').decode(errors='replace') # type: str elapsed = info['elapsed'] # type: float memory = info['memory'] # type: Optional[float] # lock is require to avoid mixing logs if in parallel nullcontext = contextlib.ExitStack() # TODO: use contextlib.nullcontext() after updating Python to 3.7 with lock or nullcontext: if lock is not None: log.emit('') log.info('%s', test_name) log.status('time: %f sec', elapsed) if memory: if memory < MEMORY_PRINT: if args.print_memory: log.status('memory: %f MB', memory) elif memory < MEMORY_WARNING: log.status('memory: %f MB', memory) else: log.warning('memory: %f MB', memory) status = compare_and_report(proc, answer, memory, test_input_path, test_output_path, mle=args.mle, mode=args.display_mode, error=args.error, does_print_input=args.print_input, silent=args.silent, rstrip=args.rstrip, judge=args.judge) # return the result testcase = { 'name': test_name, 'input': str(test_input_path.resolve()), } if test_output_path: testcase['output'] = str(test_output_path.resolve()) return { 'status': status, 'testcase': testcase, 'output': answer, 'exitcode': proc.returncode, 'elapsed': elapsed, 'memory': memory, } 

def exec_command(command_str: str, *, stdin: Optional[IO[Any]] = None, input: Optional[bytes] = None, timeout: Optional[float] = None, gnu_time: Optional[str] = None) -> Tuple[Dict[str, Any], subprocess.Popen]: if input is not None: assert stdin is None stdin = subprocess.PIPE # type: ignore if gnu_time is not None: context = tempfile.NamedTemporaryFile(delete=True) # type: Any else: context = contextlib.ExitStack() # TODO: we should use contextlib.nullcontext() if possible with context as fh: command = shlex.split(command_str) if gnu_time is not None: command = [gnu_time, '-f', '%M', '-o', fh.name, '--'] + command if os.name == 'nt': # HACK: without this encoding and decoding, something randomly fails with multithreading; see https://github.com/kmyk/online-judge-tools/issues/468 command = command_str.encode().decode() # type: ignore begin = time.perf_counter() # We need kill processes called from the "time" command using process groups. Without this, orphans spawn. see https://github.com/kmyk/online-judge-tools/issues/640 preexec_fn = None if gnu_time is not None and os.name == 'posix': preexec_fn = os.setsid try: proc = subprocess.Popen(command, stdin=stdin, stdout=subprocess.PIPE, stderr=sys.stderr, preexec_fn=preexec_fn) except FileNotFoundError: log.error('No such file or directory: %s', command) sys.exit(1) except PermissionError: log.error('Permission denied: %s', command) sys.exit(1) answer = None # type: Optional[bytes] try: answer, _ = proc.communicate(input=input, timeout=timeout) except subprocess.TimeoutExpired: pass finally: if preexec_fn is not None: try: os.killpg(os.getpgid(proc.pid), signal.SIGTERM) except ProcessLookupError: pass else: proc.terminate() end = time.perf_counter() memory = None # type: Optional[float] if gnu_time is not None: with open(fh.name) as fh1: reported = fh1.read() log.debug('GNU time says:\n%s', reported) if reported.strip() and reported.splitlines()[-1].isdigit(): memory = int(reported.splitlines()[-1]) / 1000 info = { 'answer': answer, # Optional[byte] 'elapsed': end - begin, # float, in second 'memory': memory, # Optional[float], in megabyte } return info, proc