Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_cannot_instantiate(self):
with self.assertRaises(TypeError):
Union()
with self.assertRaises(TypeError):
type(Union)()
u = Union[int, float]
with self.assertRaises(TypeError):
u()
with self.assertRaises(TypeError):
type(u)()
import numpy as np
import pandas as pd
import ibis.common.exceptions as com
import ibis.expr.operations as ops
import ibis.expr.types as ir
import ibis.util as util
def _sequence_to_tuple(x):
return tuple(x) if util.is_iterable(x) else x
RowsWithMaxLookback = NamedTuple(
'RowsWithMaxLookback',
[('rows', Union[int, np.integer]), ('max_lookback', ir.IntervalValue)],
)
def _choose_non_empty_val(first, second):
if isinstance(first, (int, np.integer)) and first:
non_empty_value = first
elif not isinstance(first, (int, np.integer)) and first is not None:
non_empty_value = first
else:
non_empty_value = second
return non_empty_value
def _determine_how(preceding):
offset_type = type(_get_preceding_value(preceding))
if issubclass(offset_type, (int, np.integer)):
async def generator_to_async_generator(
get_iterable: Callable[[], Iterable[_T]]
) -> AsyncGenerator[_T, None]:
"""
Turn a generator or iterable into an async generator.
This works by running the generator in a background thread.
:param get_iterable: Function that returns a generator or iterable when
called.
"""
quitting = False
_done = _Done()
q: Queue[Union[_T, _Done]] = Queue()
loop = get_event_loop()
def runner() -> None:
"""
Consume the generator in background thread.
When items are received, they'll be pushed to the queue.
"""
try:
for item in get_iterable():
loop.call_soon_threadsafe(q.put_nowait, item)
# When this async generator was cancelled (closed), stop this
# thread.
if quitting:
break
from typing import Union, List, Tuple, Optional, Dict, Any, Sequence, overload
from typing_extensions import Literal
from .. import core, config, utils
__all__ = ['generate_colors', 'prepare_connector_cmap', 'prepare_colormap',
'eval_color', 'hex_to_rgb', 'vary_colors']
logger = config.logger
# Some definitions for mypy
RGB_color = Tuple[float, float, float]
RGBA_color = Tuple[float, float, float, float]
Str_color = str
ColorList = Sequence[Union[RGB_color, RGBA_color, Str_color]]
AnyColor = Union[RGB_color, RGBA_color, Str_color, ColorList]
def generate_colors(N: int,
color_space: Union[Literal['RGB'],
Literal['Grayscale']] = 'RGB',
color_range: Union[Literal[1],
Literal[255]] = 1
) -> List[Tuple[float, float, float]]:
"""Divide colorspace into N evenly distributed colors.
Returns
-------
colormap : list
[(r, g, b), (r, g, b), ...]
"""
if N == 1:
else:
fd = open(filename, mode)
return fd
def wrap_read_function(read, filename, index=None, **kwargs):
"""Convert read-function to generator."""
if index is None:
yield read(filename, **kwargs)
else:
for atoms in read(filename, index, **kwargs):
yield atoms
NameOrFile = Union[str, PurePath, IO]
def write(
filename: NameOrFile,
images: Union[Atoms, Sequence[Atoms]],
format: str = None,
parallel: bool = True,
append: bool = False,
**kwargs: dict
) -> None:
"""Write Atoms object(s) to file.
filename: str or file
Name of the file to write to or a file descriptor. The name '-'
means standard output.
images: Atoms object or list of Atoms objects
else:
raise TaborException('The algorithm is not smart enough to make this sequence table longer')
elif _check_partial_unroll(program, i, min_seq_len=min_seq_len):
i += 1
else:
raise TaborException('The algorithm is not smart enough to make this sequence table longer')
else:
i += 1
ParsedProgram = NamedTuple('ParsedProgram', [('advanced_sequencer_table', Sequence[TableEntry]),
('sequencer_tables', Sequence[Sequence[
Tuple[TableDescription, Optional[VolatileProperty]]]]),
('waveforms', Tuple[Waveform, ...]),
('volatile_parameter_positions', Dict[Union[int, Tuple[int, int]],
VolatileRepetitionCount])])
def parse_aseq_program(program: Loop, used_channels: FrozenSet[ChannelID]) -> ParsedProgram:
volatile_parameter_positions = {}
advanced_sequencer_table = []
# we use an ordered dict here to avoid O(n**2) behaviour while looking for duplicates
sequencer_tables = OrderedDict()
waveforms = OrderedDict()
for adv_position, sequencer_table_loop in enumerate(program):
current_sequencer_table = []
for position, (waveform, repetition_definition, volatile_repetition) in enumerate(
(waveform_loop.waveform.get_subset_for_channels(used_channels),
waveform_loop.repetition_definition, waveform_loop.volatile_repetition)
for waveform_loop in cast(Sequence[Loop], sequencer_table_loop)):
from typing_extensions import (
Literal,
Protocol,
)
from eth_utils import (
is_list_like,
)
# The RLP-decoded node is either blank, or a list, full of bytes or recursive nodes
# Recursive definitions don't seem supported at the moment, follow:
# https://github.com/python/mypy/issues/731
# Another option is to manually declare a few levels of the type. It should be possible
# to determine the maximum number of embeds with single-nibble keys and single byte values.
RawHexaryNode = Union[
# Blank node
Literal[b''],
# Leaf or extension node are length 2
# Branch node is length 17
List[Union[
# keys, hashes to next nodes, and values
bytes,
# embedded subnodes
"RawHexaryNode",
]],
]
class Nibble(enum.IntEnum):
return IEMLSyntaxType.__ge__(self, other) and self != other
def __le__(self, other):
return IEMLSyntaxType.__lt__(self, other) and self == other
def __ge__(self, other):
return not IEMLSyntaxType.__lt__(self, other)
def __lt__(self, other):
return self.__rank < other.__rank
def syntax_rank(self):
return self.__rank
Literal = Union[List[str], str]
class Usl(metaclass=IEMLSyntaxType):
def __init__(self, literals: Literal=None):
super().__init__()
self._paths = None
_literals = []
if literals is not None:
if isinstance(literals, str):
_literals = [literals]
else:
try:
_literals = tuple(literals)
except TypeError:
raise InvalidIEMLObjectArgument(self.__class__, "The literals argument %s must be an iterable of "
"""
return []
@abstractproperty
def bindings(self) -> List[Binding]:
"""
List of `Binding` objects.
(These need to be exposed, so that `KeyBindings` objects can be merged
together.)
"""
return []
# `add` and `remove` don't have to be part of this interface.
T = TypeVar("T", bound=Union[KeyHandlerCallable, Binding])
class KeyBindings(KeyBindingsBase):
"""
A container for a set of key bindings.
Example usage::
kb = KeyBindings()
@kb.add('c-t')
def _(event):
print('Control-T pressed')
@kb.add('c-a', 'c-b')
def _(event):
from typing import List, Union, Tuple
from exactly_lib.util.textformat.structure import document as doc
from exactly_lib.util.textformat.structure import lists, table, structures as docs
from exactly_lib.util.textformat.structure.core import ParagraphItem
from exactly_lib.util.textformat.structure.paragraph import Paragraph
from exactly_lib.util.textformat.structure.structures import cell
InitialParagraphsOrSectionContents = Union[doc.SectionContents, List[ParagraphItem]]
def append_section_if_contents_is_non_empty(
output_list_of_sections: List[doc.SectionItem],
section_title_str_or_text,
initial_paragraphs_or_section_contents: InitialParagraphsOrSectionContents):
section_contents = initial_paragraphs_or_section_contents
if isinstance(initial_paragraphs_or_section_contents, list):
section_contents = doc.SectionContents(initial_paragraphs_or_section_contents)
if not section_contents.is_empty:
output_list_of_sections.append(doc.Section(docs.text_from_unknown(section_title_str_or_text),
section_contents))
def append_sections_if_contents_is_non_empty(
output_list_of_sections: List[doc.SectionItem],