Coverage for src/evutils/io/stream.py: 100%

1from typing import Iterator, Any 

2 

3from evutils.io._source import make_source 

4from evutils.io.decoders import resolve_decoder_cls 

5from evutils.types import TriggerArray 

6 

7class EventStreamer: 

8 """A low-level streamer that yields raw event blocks exactly as the parser outputs them. 

9 It performs ZERO slicing or logic; it purely drives the decoder. 

10 """ 

11 def __init__(self, file: Any, ext_trigger: bool = False, decoder_cls: Any = None, **kwargs: Any): 

12 self.source = make_source(file) 

13 if decoder_cls is None: 

14 decoder_cls = resolve_decoder_cls(self.source) 

15 

16 self.decoder = decoder_cls(self.source, read_external_triggers=ext_trigger, **kwargs) 

17 self.ext_trigger = ext_trigger 

18 

19 def __iter__(self) -> Iterator[Any]: 

20 self.decoder.init() 

21 while True: 

22 chunk = self.decoder.read_chunk() 

23 

24 if self.ext_trigger: 

25 if isinstance(chunk, tuple): 

26 ev_chunk, tr_chunk = chunk 

27 else: 

28 ev_chunk = chunk 

29 tr_chunk = TriggerArray.empty() 

30 else: 

31 ev_chunk = chunk 

32 tr_chunk = None 

33 

34 if len(ev_chunk) == 0 and (tr_chunk is None or len(tr_chunk) == 0): 

35 break 

36 

37 if self.ext_trigger: 

38 yield ev_chunk, tr_chunk 

39 else: 

40 yield ev_chunk 

41 

42