KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use iter_streams method in stestr

Best Python code snippet using stestr_python

stream_classes.py

Source:stream_classes.py

1from typing import Optional, NoReturn2from datetime import datetime3from random import randint4TMP_FILES_TEMPLATE = 'stream_{}.tmp'5TMP_FILES_ENCODING = 'utf8'6try:  # Assume we're a submodule in a package.7    from utils import arguments as arg8    from utils.decorators import deprecated_with_alternative9    from interfaces import (10        StreamInterface, IterableStreamInterface, LocalStreamInterface, RegularStreamInterface, PairStreamInterface,11        StreamBuilderInterface, ContextInterface, ConnectorInterface,12        TemporaryLocationInterface, TemporaryFilesMaskInterface,13        StreamType, ItemType, JoinType, How, Auto, AUTO,14    )15    from streams.abstract.abstract_stream import AbstractStream16    from streams.abstract.iterable_stream import IterableStream, MAX_ITEMS_IN_MEMORY17    from streams.abstract.local_stream import LocalStream18    from streams.abstract.wrapper_stream import WrapperStream19    from streams.mixin.columnar_mixin import ColumnarMixin20    from streams.mixin.convert_mixin import ConvertMixin21    from streams.regular.any_stream import AnyStream22    from streams.regular.line_stream import LineStream23    from streams.regular.row_stream import RowStream24    from streams.pairs.key_value_stream import KeyValueStream25    from streams.regular.struct_stream import StructStream26    from streams.regular.record_stream import RecordStream27    from streams.wrappers.pandas_stream import PandasStream28    from streams.wrappers.sql_stream import SqlStream29    from streams.stream_builder import StreamBuilder30    from connectors.filesystem.temporary_files import TemporaryLocation31    from content.struct import struct_row as sr32except ImportError:  # Apparently no higher-level package has been imported, fall back to a local import.33    from ..utils import arguments as arg34    from ..utils.decorators import deprecated_with_alternative35    from ..interfaces import (36        StreamInterface, IterableStreamInterface, LocalStreamInterface, RegularStreamInterface, PairStreamInterface,37        StreamBuilderInterface, ContextInterface, ConnectorInterface,38        TemporaryLocationInterface, TemporaryFilesMaskInterface,39        StreamType, ItemType, JoinType, How, Auto, AUTO,40    )41    from .abstract.abstract_stream import AbstractStream42    from .abstract.iterable_stream import IterableStream, MAX_ITEMS_IN_MEMORY43    from .abstract.local_stream import LocalStream44    from .abstract.wrapper_stream import WrapperStream45    from .mixin.columnar_mixin import ColumnarMixin46    from .mixin.convert_mixin import ConvertMixin47    from .regular.any_stream import AnyStream48    from .regular.line_stream import LineStream49    from .regular.row_stream import RowStream50    from .pairs.key_value_stream import KeyValueStream51    from .regular.struct_stream import StructStream52    from .regular.record_stream import RecordStream53    from .wrappers.pandas_stream import PandasStream54    from .wrappers.sql_stream import SqlStream55    from .stream_builder import StreamBuilder56    from ..connectors.filesystem.temporary_files import TemporaryLocation57    from ..content.struct import struct_row as sr58STREAM_CLASSES = (59    AbstractStream, IterableStream,60    AnyStream,61    LineStream, RowStream, RecordStream,62    StructStream,63    KeyValueStream,64    PandasStream, SqlStream,65)66DICT_STREAM_CLASSES = dict(67    AnyStream=AnyStream,68    LineStream=LineStream,69    RowStream=RowStream,70    KeyValueStream=KeyValueStream,71    StructStream=StructStream,72    RecordStream=RecordStream,73    PandasStream=PandasStream,74    SqlStream=SqlStream,75)76_context = None  # global77StreamType.set_default(AnyStream.__name__)78StreamType.set_dict_classes(DICT_STREAM_CLASSES)79@deprecated_with_alternative('StreamType.get_class()')80def get_class(stream_type):81    return StreamType(stream_type).get_class()82DICT_ITEM_TO_STREAM_TYPE = {83    ItemType.Any: StreamType.AnyStream,84    ItemType.Line: StreamType.LineStream,85    ItemType.Record: StreamType.RecordStream,86    ItemType.Row: StreamType.RowStream,87    ItemType.StructRow: StreamType.StructStream,88}89StreamBuilder._dict_classes = DICT_ITEM_TO_STREAM_TYPE90def get_context() -> Optional[ContextInterface]:91    global _context92    return _context93def set_context(cx: ContextInterface) -> NoReturn:94    global _context95    _context = cx96    storage = cx.get_local_storage()97    assert isinstance(storage, ConnectorInterface)98    TemporaryLocation.set_default_storage(storage)99def stream(stream_type, *args, **kwargs) -> StreamInterface:100    if is_stream_class(STREAM_CLASSES):101        stream_class = stream_type102    else:103        stream_class = StreamType(stream_type).get_class()104    if 'context' not in kwargs:105        kwargs['context'] = get_context()106    return stream_class(*args, **kwargs)107def is_stream_class(obj) -> bool:108    return obj in STREAM_CLASSES109def is_stream(obj) -> bool:110    return isinstance(obj, STREAM_CLASSES)111def is_row(item) -> bool:112    return RowStream.is_valid_item_type(item)113def is_record(item) -> bool:114    return RecordStream.is_valid_item_type(item)115def is_struct_row(item) -> bool:116    return isinstance(item, sr.StructRow)117@deprecated_with_alternative('AbstractStream.generate_name()')118def generate_name() -> str:119    cur_time = datetime.now().strftime('%y%m%d_%H%M%S')120    random = randint(0, 1000)121    cur_name = '{}_{:03}'.format(cur_time, random)122    return cur_name123def get_tmp_mask(name: str) -> TemporaryFilesMaskInterface:124    context = get_context()125    if context:126        location = context.get_tmp_folder()127    else:128        location = TemporaryLocation()129    assert isinstance(location, TemporaryLocation), 'got {}'.format(type(location))130    tmp_mask = location.mask(name)131    assert isinstance(tmp_mask, TemporaryFilesMaskInterface)132    return tmp_mask133def concat(*iter_streams, context=AUTO) -> StreamInterface:134    global _context135    context = arg.acquire(context, _context)136    return StreamBuilder.concat(*iter_streams, context=context)137def join(*iter_streams, key, how: How = JoinType.Left, step=AUTO, name=AUTO, context=None) -> StreamInterface:138    global _context139    context = arg.acquire(context, _context)...

stream_builder.py

Source:stream_builder.py

1from typing import Union, Iterable2import gc3try:  # Assume we're a sub-module in a package.4    from utils import arguments as arg5    from interfaces import (6        StreamInterface, Stream, StreamBuilderInterface,7        StreamType, ItemType, JoinType, How, OptionalFields, Auto, AUTO,8    )9except ImportError:  # Apparently no higher-level package has been imported, fall back to a local import.10    from ..utils import arguments as arg11    from ..interfaces import (12        StreamInterface, Stream, StreamBuilderInterface,13        StreamType, ItemType, JoinType, How, OptionalFields, Auto, AUTO,14    )15class StreamBuilder(StreamBuilderInterface):16    _dict_classes = dict()17    _stream_types = StreamType18    @classmethod19    def stream(20            cls,21            data: Iterable,22            stream_type: Union[StreamType, StreamInterface, Auto] = AUTO,23            **kwargs24    ) -> Stream:25        if not arg.is_defined(stream_type):26            example_item = cls._get_one_item(data)27            item_type = cls._detect_item_type(example_item)28            stream_type = cls._get_dict_classes('stream(stream_type=AUTO)').get(item_type)29        return cls._get_stream_types().of(stream_type).stream(data, **kwargs)30    @staticmethod31    def is_same_stream_type(*iter_streams) -> bool:32        iter_streams = arg.update(iter_streams)33        stream_types = [i.get_stream_type() for i in iter_streams]34        return len(set(stream_types)) == 135    @classmethod36    def stack(cls, *iter_streams, how: How = 'vertical', name=AUTO, context=None, **kwargs):37        iter_streams = arg.update(iter_streams)38        assert cls.is_same_stream_type(iter_streams), 'concat(): streams must have same type: {}'.format(iter_streams)39        result = None40        for cur_stream in iter_streams:41            assert isinstance(cur_stream, StreamInterface)42            if result is None:43                if hasattr(cur_stream, 'copy'):44                    result = cur_stream.copy()45                else:46                    result = cur_stream47                if arg.is_defined(name):48                    result.set_name(name)49                if arg.is_defined(context):50                    result.set_context(context)51            elif how == 'vertical':52                result = result.add_stream(cur_stream)53            else:54                result = result.join(cur_stream, how=how, **kwargs)55            gc.collect()56        return result57    @classmethod58    def concat(cls, *iter_streams, name=AUTO, context=None):59        return cls.stack(*iter_streams, name=name, context=context)60    @classmethod61    def join(cls, *iter_streams, key, how: How = JoinType.Left, step=AUTO, name=AUTO, context=None):62        return cls.stack(*iter_streams, key=key, how=how, step=step, name=name, context=context)63    @classmethod64    def _get_dict_classes(cls, operation_name='this'):65        dict_classes = cls._dict_classes66        assert dict_classes, 'For {} operation dict_classes must be defined'.format(operation_name)67        return dict_classes68    @classmethod69    def _get_stream_types(cls, operation_name='this'):70        stream_types = cls._stream_types71        assert stream_types, 'For {} operation stream_types must be defined'.format(operation_name)72        return stream_types73    @staticmethod74    def _get_one_item(data: Iterable):75        for i in data:76            return i77    @staticmethod78    def _detect_item_type(item) -> ItemType:79        if isinstance(item, str):80            return ItemType.Line81        elif isinstance(item, dict):82            return ItemType.Record83        elif isinstance(item, (list, tuple)):84            return ItemType.Row85        elif hasattr(item, 'get_struct'):86            return ItemType.StructRow87        else:...

__init__.py

Source:__init__.py

...38    def is_online(self):39        if self._is_online == self.STATE_OFFLINE:40            return self.STATE_OFFLINE41        else:42            for s in self.iter_streams():43                if s.on:44                    return self.STATE_BROADCASTING45        return self.STATE_ONLINE46    def iter_streams(self):47        for s_list in self.streams.itervalues():48            for s in s_list.itervalues():49                yield s50    def rtmp_publish(self, quality, rtmp_url, stream_id):51        stream = self._match_stream(quality, preferred_type=['rtsp'])52        stream.rtmp_publish(rtmp_url, stream_id)53    def rtmp_stop_publish(self, stream_id):54        for s in self.iter_streams():55            if s.is_source_of(stream_id):56                s.rtmp_stop_publish(stream_id)57    def _match_stream(self, quality, preferred_type=None):58        if not preferred_type:59            preferred_type = ('rtsp', 'rtmp')60        stream_type = None61        for _type in preferred_type:62            if _type in self.streams:63                stream_type = _type64        if stream_type is None:65            for stream_type in self.streams:66                break67        if not stream_type:68            raise IVRError('No possible stream for "{0}"'.format(self))...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.