Enable Ruff PLC (Pylint Convention) (#13306)

This commit is contained in:
Avasam
2025-03-03 15:39:40 +01:00
committed by GitHub
parent 738cc5046a
commit 6d6e858e63
17 changed files with 173 additions and 153 deletions
+43 -43
View File
@@ -14,21 +14,21 @@ from tensorflow.dtypes import DType
from tensorflow.io import _CompressionTypes
from tensorflow.python.trackable.base import Trackable
_T1 = TypeVar("_T1", covariant=True)
_T1_co = TypeVar("_T1_co", covariant=True)
_T2 = TypeVar("_T2")
_T3 = TypeVar("_T3")
class Iterator(_Iterator[_T1], Trackable, ABC):
class Iterator(_Iterator[_T1_co], Trackable, ABC):
@property
@abstractmethod
def element_spec(self) -> ContainerGeneric[TypeSpec[Any]]: ...
@abstractmethod
def get_next(self) -> _T1: ...
def get_next(self) -> _T1_co: ...
@abstractmethod
def get_next_as_optional(self) -> tf.experimental.Optional[_T1]: ...
def get_next_as_optional(self) -> tf.experimental.Optional[_T1_co]: ...
class Dataset(ABC, Generic[_T1]):
def apply(self, transformation_func: Callable[[Dataset[_T1]], Dataset[_T2]]) -> Dataset[_T2]: ...
class Dataset(ABC, Generic[_T1_co]):
def apply(self, transformation_func: Callable[[Dataset[_T1_co]], Dataset[_T2]]) -> Dataset[_T2]: ...
def as_numpy_iterator(self) -> Iterator[np.ndarray[Any, Any]]: ...
def batch(
self,
@@ -37,10 +37,10 @@ class Dataset(ABC, Generic[_T1]):
num_parallel_calls: int | None = None,
deterministic: bool | None = None,
name: str | None = None,
) -> Dataset[_T1]: ...
) -> Dataset[_T1_co]: ...
def bucket_by_sequence_length(
self,
element_length_func: Callable[[_T1], ScalarTensorCompatible],
element_length_func: Callable[[_T1_co], ScalarTensorCompatible],
bucket_boundaries: Sequence[int],
bucket_batch_sizes: Sequence[int],
padded_shapes: ContainerGeneric[tf.TensorShape | TensorCompatible] | None = None,
@@ -49,14 +49,14 @@ class Dataset(ABC, Generic[_T1]):
no_padding: bool = False,
drop_remainder: bool = False,
name: str | None = None,
) -> Dataset[_T1]: ...
def cache(self, filename: str = "", name: str | None = None) -> Dataset[_T1]: ...
) -> Dataset[_T1_co]: ...
def cache(self, filename: str = "", name: str | None = None) -> Dataset[_T1_co]: ...
def cardinality(self) -> int: ...
@staticmethod
def choose_from_datasets(
datasets: Sequence[Dataset[_T2]], choice_dataset: Dataset[tf.Tensor], stop_on_empty_dataset: bool = True
) -> Dataset[_T2]: ...
def concatenate(self, dataset: Dataset[_T1], name: str | None = None) -> Dataset[_T1]: ...
def concatenate(self, dataset: Dataset[_T1_co], name: str | None = None) -> Dataset[_T1_co]: ...
@staticmethod
def counter(
start: ScalarTensorCompatible = 0, step: ScalarTensorCompatible = 1, dtype: DType = ..., name: str | None = None
@@ -64,9 +64,9 @@ class Dataset(ABC, Generic[_T1]):
@property
@abstractmethod
def element_spec(self) -> ContainerGeneric[TypeSpec[Any]]: ...
def enumerate(self, start: ScalarTensorCompatible = 0, name: str | None = None) -> Dataset[tuple[int, _T1]]: ...
def filter(self, predicate: Callable[[_T1], bool | tf.Tensor], name: str | None = None) -> Dataset[_T1]: ...
def flat_map(self, map_func: Callable[[_T1], Dataset[_T2]], name: str | None = None) -> Dataset[_T2]: ...
def enumerate(self, start: ScalarTensorCompatible = 0, name: str | None = None) -> Dataset[tuple[int, _T1_co]]: ...
def filter(self, predicate: Callable[[_T1_co], bool | tf.Tensor], name: str | None = None) -> Dataset[_T1_co]: ...
def flat_map(self, map_func: Callable[[_T1_co], Dataset[_T2]], name: str | None = None) -> Dataset[_T2]: ...
# PEP 646 can be used here for a more precise type when better supported.
@staticmethod
def from_generator(
@@ -81,26 +81,26 @@ class Dataset(ABC, Generic[_T1]):
def from_tensors(tensors: Any, name: str | None = None) -> Dataset[Any]: ...
@staticmethod
def from_tensor_slices(tensors: TensorCompatible, name: str | None = None) -> Dataset[Any]: ...
def get_single_element(self, name: str | None = None) -> _T1: ...
def get_single_element(self, name: str | None = None) -> _T1_co: ...
def group_by_window(
self,
key_func: Callable[[_T1], tf.Tensor],
reduce_func: Callable[[tf.Tensor, Dataset[_T1]], Dataset[_T2]],
key_func: Callable[[_T1_co], tf.Tensor],
reduce_func: Callable[[tf.Tensor, Dataset[_T1_co]], Dataset[_T2]],
window_size: ScalarTensorCompatible | None = None,
window_size_func: Callable[[tf.Tensor], tf.Tensor] | None = None,
name: str | None = None,
) -> Dataset[_T2]: ...
def ignore_errors(self, log_warning: bool = False, name: str | None = None) -> Dataset[_T1]: ...
def ignore_errors(self, log_warning: bool = False, name: str | None = None) -> Dataset[_T1_co]: ...
def interleave(
self,
map_func: Callable[[_T1], Dataset[_T2]],
map_func: Callable[[_T1_co], Dataset[_T2]],
cycle_length: int | None = None,
block_length: int | None = None,
num_parallel_calls: int | None = None,
deterministic: bool | None = None,
name: str | None = None,
) -> Dataset[_T2]: ...
def __iter__(self) -> Iterator[_T1]: ...
def __iter__(self) -> Iterator[_T1_co]: ...
@staticmethod
def list_files(
file_pattern: str | Sequence[str] | TensorCompatible,
@@ -134,8 +134,8 @@ class Dataset(ABC, Generic[_T1]):
padding_values: ContainerGeneric[ScalarTensorCompatible] | None = None,
drop_remainder: bool = False,
name: str | None = None,
) -> Dataset[_T1]: ...
def prefetch(self, buffer_size: ScalarTensorCompatible, name: str | None = None) -> Dataset[_T1]: ...
) -> Dataset[_T1_co]: ...
def prefetch(self, buffer_size: ScalarTensorCompatible, name: str | None = None) -> Dataset[_T1_co]: ...
def ragged_batch(
self,
batch_size: ScalarTensorCompatible,
@@ -162,62 +162,62 @@ class Dataset(ABC, Generic[_T1]):
) -> Dataset[tf.Tensor]: ...
def rebatch(
self, batch_size: ScalarTensorCompatible, drop_remainder: bool = False, name: str | None = None
) -> Dataset[_T1]: ...
def reduce(self, initial_state: _T2, reduce_func: Callable[[_T2, _T1], _T2], name: str | None = None) -> _T2: ...
) -> Dataset[_T1_co]: ...
def reduce(self, initial_state: _T2, reduce_func: Callable[[_T2, _T1_co], _T2], name: str | None = None) -> _T2: ...
def rejection_resample(
self,
class_func: Callable[[_T1], ScalarTensorCompatible],
class_func: Callable[[_T1_co], ScalarTensorCompatible],
target_dist: TensorCompatible,
initial_dist: TensorCompatible | None = None,
seed: int | None = None,
name: str | None = None,
) -> Dataset[_T1]: ...
def repeat(self, count: ScalarTensorCompatible | None = None, name: str | None = None) -> Dataset[_T1]: ...
) -> Dataset[_T1_co]: ...
def repeat(self, count: ScalarTensorCompatible | None = None, name: str | None = None) -> Dataset[_T1_co]: ...
@staticmethod
def sample_from_datasets(
datasets: Sequence[Dataset[_T1]],
datasets: Sequence[Dataset[_T1_co]],
weights: TensorCompatible | None = None,
seed: int | None = None,
stop_on_empty_dataset: bool = False,
rerandomize_each_iteration: bool | None = None,
) -> Dataset[_T1]: ...
) -> Dataset[_T1_co]: ...
# Incomplete as tf.train.CheckpointOptions not yet covered.
def save(
self,
path: str,
compression: _CompressionTypes = None,
shard_func: Callable[[_T1], int] | None = None,
shard_func: Callable[[_T1_co], int] | None = None,
checkpoint_args: Incomplete | None = None,
) -> None: ...
def scan(
self, initial_state: _T2, scan_func: Callable[[_T2, _T1], tuple[_T2, _T3]], name: str | None = None
self, initial_state: _T2, scan_func: Callable[[_T2, _T1_co], tuple[_T2, _T3]], name: str | None = None
) -> Dataset[_T3]: ...
def shard(
self, num_shards: ScalarTensorCompatible, index: ScalarTensorCompatible, name: str | None = None
) -> Dataset[_T1]: ...
) -> Dataset[_T1_co]: ...
def shuffle(
self,
buffer_size: ScalarTensorCompatible,
seed: int | None = None,
reshuffle_each_iteration: bool = True,
name: str | None = None,
) -> Dataset[_T1]: ...
def skip(self, count: ScalarTensorCompatible, name: str | None = None) -> Dataset[_T1]: ...
) -> Dataset[_T1_co]: ...
def skip(self, count: ScalarTensorCompatible, name: str | None = None) -> Dataset[_T1_co]: ...
def snapshot(
self,
path: str,
compression: _CompressionTypes = "AUTO",
reader_func: Callable[[Dataset[Dataset[_T1]]], Dataset[_T1]] | None = None,
shard_func: Callable[[_T1], ScalarTensorCompatible] | None = None,
reader_func: Callable[[Dataset[Dataset[_T1_co]]], Dataset[_T1_co]] | None = None,
shard_func: Callable[[_T1_co], ScalarTensorCompatible] | None = None,
name: str | None = None,
) -> Dataset[_T1]: ...
) -> Dataset[_T1_co]: ...
def sparse_batch(
self, batch_size: ScalarTensorCompatible, row_shape: tf.TensorShape | TensorCompatible, name: str | None = None
) -> Dataset[tf.SparseTensor]: ...
def take(self, count: ScalarTensorCompatible, name: str | None = None) -> Dataset[_T1]: ...
def take_while(self, predicate: Callable[[_T1], ScalarTensorCompatible], name: str | None = None) -> Dataset[_T1]: ...
def unbatch(self, name: str | None = None) -> Dataset[_T1]: ...
def unique(self, name: str | None = None) -> Dataset[_T1]: ...
def take(self, count: ScalarTensorCompatible, name: str | None = None) -> Dataset[_T1_co]: ...
def take_while(self, predicate: Callable[[_T1_co], ScalarTensorCompatible], name: str | None = None) -> Dataset[_T1_co]: ...
def unbatch(self, name: str | None = None) -> Dataset[_T1_co]: ...
def unique(self, name: str | None = None) -> Dataset[_T1_co]: ...
def window(
self,
size: ScalarTensorCompatible,
@@ -225,8 +225,8 @@ class Dataset(ABC, Generic[_T1]):
stride: ScalarTensorCompatible = 1,
drop_remainder: bool = False,
name: str | None = None,
) -> Dataset[Dataset[_T1]]: ...
def with_options(self, options: Options, name: str | None = None) -> Dataset[_T1]: ...
) -> Dataset[Dataset[_T1_co]]: ...
def with_options(self, options: Options, name: str | None = None) -> Dataset[_T1_co]: ...
@overload
@staticmethod
def zip(
@@ -3,10 +3,10 @@ from typing import Generic, TypeVar
from tensorflow._aliases import AnyArray
_Value = TypeVar("_Value", covariant=True)
_Value_co = TypeVar("_Value_co", covariant=True)
class RemoteValue(Generic[_Value]):
class RemoteValue(Generic[_Value_co]):
def fetch(self) -> AnyArray: ...
def get(self) -> _Value: ...
def get(self) -> _Value_co: ...
def __getattr__(name: str) -> Incomplete: ...
@@ -11,8 +11,8 @@ from tensorflow.keras.constraints import Constraint
from tensorflow.keras.initializers import _Initializer
from tensorflow.keras.regularizers import Regularizer, _Regularizer
_InputT = TypeVar("_InputT", contravariant=True)
_OutputT = TypeVar("_OutputT", covariant=True)
_InputT_contra = TypeVar("_InputT_contra", contravariant=True)
_OutputT_co = TypeVar("_OutputT_co", covariant=True)
class InputSpec:
dtype: str | None
@@ -39,9 +39,9 @@ class InputSpec:
# Most layers have input and output type of just Tensor and when we support default type variables,
# maybe worth trying.
class Layer(tf.Module, Generic[_InputT, _OutputT]):
class Layer(tf.Module, Generic[_InputT_contra, _OutputT_co]):
# The most general type is ContainerGeneric[InputSpec] as it really
# depends on _InputT. For most Layers it is just InputSpec
# depends on _InputT_contra. For most Layers it is just InputSpec
# though. Maybe describable with HKT?
input_spec: InputSpec | Any
@@ -65,11 +65,13 @@ class Layer(tf.Module, Generic[_InputT, _OutputT]):
# *args/**kwargs are allowed, but have obscure footguns and tensorflow documentation discourages their usage.
# First argument will automatically be cast to layer's compute dtype, but any other tensor arguments will not be.
# Also various tensorflow tools/apis can misbehave if they encounter a layer with *args/**kwargs.
def __call__(self, inputs: _InputT, *, training: bool = False, mask: TensorCompatible | None = None) -> _OutputT: ...
def call(self, inputs: _InputT, /) -> _OutputT: ...
def __call__(
self, inputs: _InputT_contra, *, training: bool = False, mask: TensorCompatible | None = None
) -> _OutputT_co: ...
def call(self, inputs: _InputT_contra, /) -> _OutputT_co: ...
# input_shape's real type depends on _InputT, but we can't express that without HKT.
# For example _InputT tf.Tensor -> tf.TensorShape, _InputT dict[str, tf.Tensor] -> dict[str, tf.TensorShape].
# input_shape's real type depends on _InputT_contra, but we can't express that without HKT.
# For example _InputT_contra tf.Tensor -> tf.TensorShape, _InputT_contra dict[str, tf.Tensor] -> dict[str, tf.TensorShape].
def build(self, input_shape: Any, /) -> None: ...
@overload
def compute_output_shape(self: Layer[tf.Tensor, tf.Tensor], input_shape: tf.TensorShape, /) -> tf.TensorShape: ...
+11 -9
View File
@@ -9,14 +9,14 @@ import numpy.typing as npt
import tensorflow as tf
from tensorflow import Variable
from tensorflow._aliases import ContainerGeneric, ShapeLike, TensorCompatible
from tensorflow.keras.layers import Layer, _InputT, _OutputT
from tensorflow.keras.layers import Layer, _InputT_contra, _OutputT_co
from tensorflow.keras.optimizers import Optimizer
_Loss: TypeAlias = str | tf.keras.losses.Loss | Callable[[TensorCompatible, TensorCompatible], tf.Tensor]
_Metric: TypeAlias = str | tf.keras.metrics.Metric | Callable[[TensorCompatible, TensorCompatible], tf.Tensor] | None
# Missing keras.src.backend.tensorflow.trainer.TensorFlowTrainer as a base class, which is not exposed by tensorflow
class Model(Layer[_InputT, _OutputT]):
class Model(Layer[_InputT_contra, _OutputT_co]):
_train_counter: tf.Variable
_test_counter: tf.Variable
optimizer: Optimizer | None
@@ -27,13 +27,15 @@ class Model(Layer[_InputT, _OutputT]):
) -> tf.Tensor | None: ...
stop_training: bool
def __new__(cls, *args: Any, **kwargs: Any) -> Model[_InputT, _OutputT]: ...
def __new__(cls, *args: Any, **kwargs: Any) -> Model[_InputT_contra, _OutputT_co]: ...
def __init__(self, *args: Any, **kwargs: Any) -> None: ...
def __setattr__(self, name: str, value: Any) -> None: ...
def __reduce__(self): ...
def build(self, input_shape: ShapeLike) -> None: ...
def __call__(self, inputs: _InputT, *, training: bool = False, mask: TensorCompatible | None = None) -> _OutputT: ...
def call(self, inputs: _InputT, training: bool | None = None, mask: TensorCompatible | None = None) -> _OutputT: ...
def __call__(
self, inputs: _InputT_contra, *, training: bool = False, mask: TensorCompatible | None = None
) -> _OutputT_co: ...
def call(self, inputs: _InputT_contra, training: bool | None = None, mask: TensorCompatible | None = None) -> _OutputT_co: ...
# Ideally loss/metrics/output would share the same structure but higher kinded types are not supported.
def compile(
self,
@@ -106,8 +108,8 @@ class Model(Layer[_InputT, _OutputT]):
return_dict: bool = False,
**kwargs: Any,
) -> float | list[float]: ...
def predict_step(self, data: _InputT) -> _OutputT: ...
def make_predict_function(self, force: bool = False) -> Callable[[tf.data.Iterator[Incomplete]], _OutputT]: ...
def predict_step(self, data: _InputT_contra) -> _OutputT_co: ...
def make_predict_function(self, force: bool = False) -> Callable[[tf.data.Iterator[Incomplete]], _OutputT_co]: ...
def predict(
self,
x: TensorCompatible | tf.data.Dataset[Incomplete],
@@ -115,7 +117,7 @@ class Model(Layer[_InputT, _OutputT]):
verbose: Literal["auto", 0, 1, 2] = "auto",
steps: int | None = None,
callbacks: list[tf.keras.callbacks.Callback] | None = None,
) -> _OutputT: ...
) -> _OutputT_co: ...
def reset_metrics(self) -> None: ...
def train_on_batch(
self,
@@ -132,7 +134,7 @@ class Model(Layer[_InputT, _OutputT]):
sample_weight: npt.NDArray[np.float64] | None = None,
return_dict: bool = False,
) -> float | list[float]: ...
def predict_on_batch(self, x: Iterator[_InputT]) -> npt.NDArray[Incomplete]: ...
def predict_on_batch(self, x: Iterator[_InputT_contra]) -> npt.NDArray[Incomplete]: ...
@property
def trainable_weights(self) -> list[Variable]: ...
@property
@@ -10,7 +10,7 @@ from tensorflow.saved_model.experimental import VariablePolicy
from tensorflow.types.experimental import ConcreteFunction, PolymorphicFunction
_P = ParamSpec("_P")
_R = TypeVar("_R", covariant=True)
_R_co = TypeVar("_R_co", covariant=True)
class Asset:
@property
@@ -77,10 +77,10 @@ class SaveOptions:
def contains_saved_model(export_dir: str | Path) -> bool: ...
class _LoadedAttributes(Generic[_P, _R]):
signatures: Mapping[str, ConcreteFunction[_P, _R]]
class _LoadedAttributes(Generic[_P, _R_co]):
signatures: Mapping[str, ConcreteFunction[_P, _R_co]]
class _LoadedModel(AutoTrackable, _LoadedAttributes[_P, _R]):
class _LoadedModel(AutoTrackable, _LoadedAttributes[_P, _R_co]):
variables: list[tf.Variable]
trainable_variables: list[tf.Variable]
# TF1 model artifact specific
@@ -7,23 +7,23 @@ import tensorflow as tf
from tensorflow._aliases import ContainerGeneric
_P = ParamSpec("_P")
_R = TypeVar("_R", covariant=True)
_R_co = TypeVar("_R_co", covariant=True)
class Callable(Generic[_P, _R], metaclass=abc.ABCMeta):
def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R: ...
class Callable(Generic[_P, _R_co], metaclass=abc.ABCMeta):
def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R_co: ...
class ConcreteFunction(Callable[_P, _R], metaclass=abc.ABCMeta):
def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R: ...
class ConcreteFunction(Callable[_P, _R_co], metaclass=abc.ABCMeta):
def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R_co: ...
class PolymorphicFunction(Callable[_P, _R], metaclass=abc.ABCMeta):
class PolymorphicFunction(Callable[_P, _R_co], metaclass=abc.ABCMeta):
@overload
@abc.abstractmethod
def get_concrete_function(self, *args: _P.args, **kwargs: _P.kwargs) -> ConcreteFunction[_P, _R]: ...
def get_concrete_function(self, *args: _P.args, **kwargs: _P.kwargs) -> ConcreteFunction[_P, _R_co]: ...
@overload
@abc.abstractmethod
def get_concrete_function(
self, *args: ContainerGeneric[tf.TypeSpec[Any]], **kwargs: ContainerGeneric[tf.TypeSpec[Any]]
) -> ConcreteFunction[_P, _R]: ...
) -> ConcreteFunction[_P, _R_co]: ...
def experimental_get_compiler_ir(self, *args, **kwargs): ...
GenericFunction = PolymorphicFunction