From 265ce737c8505b3e1ebb995c73040cefb22387ae Mon Sep 17 00:00:00 2001 From: bruno-f-cruz <7049351+bruno-f-cruz@users.noreply.github.com> Date: Wed, 24 Jul 2024 12:30:15 -0700 Subject: [PATCH] Linting --- setup.py | 2 +- src/aind_behavior_core_analysis/io/_core.py | 66 ++++++++++++------- src/aind_behavior_core_analysis/io/_utils.py | 28 +------- .../io/data_stream.py | 11 +--- 4 files changed, 49 insertions(+), 58 deletions(-) diff --git a/setup.py b/setup.py index 57c026b..7f1a176 100644 --- a/setup.py +++ b/setup.py @@ -1,4 +1,4 @@ from setuptools import setup if __name__ == "__main__": - setup() \ No newline at end of file + setup() diff --git a/src/aind_behavior_core_analysis/io/_core.py b/src/aind_behavior_core_analysis/io/_core.py index 3f9988e..d36331f 100644 --- a/src/aind_behavior_core_analysis/io/_core.py +++ b/src/aind_behavior_core_analysis/io/_core.py @@ -8,23 +8,22 @@ Any, Generic, List, + NamedTuple, Optional, Protocol, Sequence, - Tuple, Type, TypeVar, overload, runtime_checkable, ) -from aind_behavior_core_analysis.io._utils import StrPattern, validate_str_pattern +from aind_behavior_core_analysis.io._utils import StrPattern TData = TypeVar("TData", bound=Any) class DataStream(abc.ABC, Generic[TData]): - def __init__( self, /, @@ -35,7 +34,6 @@ def __init__( _data: Optional[TData] = None, **kwargs, ) -> None: - self._auto_load = auto_load self._data = _data @@ -90,7 +88,6 @@ def data(self) -> TData: return self._data def load(self, /, path: Optional[PathLike] = None, *, force_reload: bool = False, **kwargs) -> TData: - if force_reload is False and self._data: pass else: @@ -108,11 +105,13 @@ def __str__(self) -> str: @runtime_checkable class _DataStreamSourceBuilder(Protocol): + def build(self, /, source: Optional[DataStreamSource] = None, **kwargs) -> StreamCollection: + ... - def build(self, /, source: Optional[DataStreamSource] = None, **kwargs) -> StreamCollection: ... - -_SequenceDataStreamBuilderPattern = Sequence[Tuple[Type[DataStream], StrPattern]] +class DataStreamBuilderPattern(NamedTuple): + stream_type: Type[DataStream] + pattern: StrPattern class DataStreamSource: @@ -129,19 +128,34 @@ def __init__( name: Optional[str] = None, auto_load: bool = False, **kwargs, - ) -> None: ... + ) -> None: + ... + + @overload + def __init__( + self, + /, + path: PathLike, + builder: DataStreamBuilderPattern, + *, + name: Optional[str] = None, + auto_load: bool = False, + **kwargs, + ) -> None: + ... @overload def __init__( self, /, path: PathLike, - builder: _SequenceDataStreamBuilderPattern, + builder: Sequence[DataStreamBuilderPattern], *, name: Optional[str] = None, auto_load: bool = False, **kwargs, - ) -> None: ... + ) -> None: + ... @overload def __init__( @@ -153,7 +167,8 @@ def __init__( name: Optional[str] = None, auto_load: bool = False, **kwargs, - ) -> None: ... + ) -> None: + ... @overload def __init__( @@ -165,19 +180,23 @@ def __init__( name: Optional[str] = None, auto_load: bool = False, **kwargs, - ) -> None: ... + ) -> None: + ... def __init__( self, /, path: PathLike, - builder: None | Type[DataStream] | _SequenceDataStreamBuilderPattern | _DataStreamSourceBuilder = None, + builder: None + | Type[DataStream] + | DataStreamBuilderPattern + | Sequence[DataStreamBuilderPattern] + | _DataStreamSourceBuilder = None, *, name: Optional[str] = None, auto_load: bool = False, **kwargs, ) -> None: - self._streams: StreamCollection self._path = Path(path) @@ -206,15 +225,18 @@ def __init__( self.reload_streams() @staticmethod - def _normalize_builder_from_data_stream(builder: Type[DataStream] | Sequence) -> _SequenceDataStreamBuilderPattern: + def _normalize_builder_from_data_stream( + builder: Type[DataStream] | DataStreamBuilderPattern | Sequence + ) -> Sequence[DataStreamBuilderPattern]: _builder: Sequence if isinstance(builder, type(DataStream)): # If only a single data stream class is provided - _builder = ((builder, "*"),) + _builder = (DataStreamBuilderPattern(stream_type=builder, pattern="*"),) + if isinstance(builder, DataStreamBuilderPattern): # If only a single data stream class is provided + _builder = (builder,) for _tuple in _builder: - if not isinstance(_tuple[0], type(DataStream)): + if not isinstance(_tuple.stream_type, type(DataStream)): raise ValueError("builder must be a DataStream type") - validate_str_pattern(_tuple[1]) return _builder @property @@ -246,10 +268,10 @@ def _get_data_streams_helper( return streams @classmethod - def _build_from_data_stream(cls, path: PathLike, builder: _SequenceDataStreamBuilderPattern) -> StreamCollection: + def _build_from_data_stream(cls, path: PathLike, builder: Sequence[DataStreamBuilderPattern]) -> StreamCollection: streams = StreamCollection() - for stream_type, pattern in builder: - _this_type_stream = cls._get_data_streams_helper(path, stream_type, pattern) + for stream_builder in builder: + _this_type_stream = cls._get_data_streams_helper(path, stream_builder.stream_type, stream_builder.pattern) for stream in _this_type_stream: if stream.name is None: raise ValueError(f"Stream {stream} does not have a name") diff --git a/src/aind_behavior_core_analysis/io/_utils.py b/src/aind_behavior_core_analysis/io/_utils.py index f4216c4..6709cf0 100644 --- a/src/aind_behavior_core_analysis/io/_utils.py +++ b/src/aind_behavior_core_analysis/io/_utils.py @@ -1,27 +1,3 @@ -import re -from typing import Sequence, Union +from typing import List, Union -StrPattern = Union[str, Sequence[str]] - - -def validate_str_pattern(pattern: StrPattern) -> None: - """ - Validates a string pattern or a sequence of string patterns. - - Args: - pattern (StrPattern): The string pattern or sequence of string patterns to validate. - - Raises: - re.error: If any of the patterns is not a valid regex pattern. - - Returns: - None - """ - if isinstance(pattern, Sequence): - for pat in pattern: - validate_str_pattern(pat) - else: - try: - re.compile(pattern) - except re.error as err: - raise re.error(f"Pattern {pattern} is not a valid regex pattern") from err +StrPattern = Union[str, List[str]] diff --git a/src/aind_behavior_core_analysis/io/data_stream.py b/src/aind_behavior_core_analysis/io/data_stream.py index 23b9e64..3bbaf2a 100644 --- a/src/aind_behavior_core_analysis/io/data_stream.py +++ b/src/aind_behavior_core_analysis/io/data_stream.py @@ -121,7 +121,6 @@ def _reader( col_names: Optional[List[str]] = None, **kwargs, ) -> DataFrameOrSeries: - has_header = csv.Sniffer().has_header(value) _header = 0 if has_header is True else None df = pd.read_csv(io.StringIO(value), header=_header, index_col=infer_index_col, names=col_names) @@ -129,8 +128,6 @@ def _reader( class SingletonStream(DataStream[str | BaseModel]): - """Represents a generic Software event.""" - def __init__( self, /, @@ -174,7 +171,6 @@ def _apply_inner_parser(self, value: Optional[str | BaseModel]) -> str | BaseMod class HarpDataStream(DataStream[DataFrameOrSeries]): - def __init__( self, /, @@ -221,7 +217,6 @@ def load(self, /, path: Optional[PathLike] = None, *, force_reload: bool = False def _bin_file_inference_helper( root_path: PathLike, register_reader: harp.reader.RegisterReader, name_hint: Optional[str] = None ) -> Path: - root_path = Path(root_path) candidate_files = list(root_path.glob(f"*_{register_reader.register.address}.bin")) @@ -244,7 +239,6 @@ def _bin_file_inference_helper( class HarpDataStreamSourceBuilder(_DataStreamSourceBuilder): - _reader_default_params = { "include_common_registers": True, "keep_type": True, @@ -258,17 +252,16 @@ def __init__( device_hint: Optional[DeviceReader | WhoAmI | PathLike] = None, default_inference_mode: _available_inference_modes = "yml", ) -> None: - self.device_hint = device_hint self.default_inference_mode = default_inference_mode @overload - def build(self, /, source: Optional[DataStreamSource] = None, **kwargs) -> StreamCollection: ... + def build(self, /, source: Optional[DataStreamSource] = None, **kwargs) -> StreamCollection: + ... def build( self, /, source: Optional[DataStreamSource] = None, *, path: Optional[PathLike] = None, **kwargs ) -> StreamCollection: - # Leaving this undocumented here for now... device_hint = kwargs.get("device_hint", self.device_hint) default_inference_mode = kwargs.get("default_inference_mode", self.default_inference_mode)