Типизация yield через Generator[]

Я пишу цифрового двойника для завода и использую фреймворк Simpy . Я написал функцию для получения деталей из складского буфера и хочу указать типизацию этой функции. Написав Generator[simpy.Event, None, Part] получаю предупреждение от mypy:

(Polygon) PS D:\Projects\Programming\Polygon\src\polygon\models> mypy buffer.py
buffer.py:46: error: Incompatible types in assignment (expression has type "None", variable has type "Part")  [assignment]
buffer.py:91: error: Signature of "put_item" incompatible with supertype "Buffer"  [override]
buffer.py:91: note:      Superclass:
buffer.py:91: note:          def put_item(self, item: Any, **kwargs: Any) -> Any
buffer.py:91: note:      Subclass:
buffer.py:91: note:          def put_item(self, count: int | float = ..., **kwargs: Any) -> Any
Found 2 errors in 1 file (checked 1 source file)

Как правильно в данном случае типизировать функцию такого формата?

Код buffer.py:

import logging
from typing import List, Any, Generator, Union
from abc import ABC, abstractmethod
import simpy
from .part import Part

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class Buffer(ABC):
    @abstractmethod
    def get_item(self, **kwargs) -> Any:
        """Получаем объект из буфера"""
        raise NotImplementedError

    @abstractmethod
    def put_item(self, item: Any, **kwargs) -> Any:
        """Кладем объект в буфер"""
        raise NotImplementedError

    @abstractmethod
    def get_buffer_level(self) -> Union[int, float]:
        """Текущее количество объектов в буфере"""
        raise NotImplementedError

    def __repr__(self):
        return f"{self.__class__.__name__}()"

    def __str__(self):
        return self.__class__.__name__


class BufferStore(Buffer):
    def __init__(self, env: simpy.Environment, buffer_config):
        # Передаем окружение и настройки буфера
        self.env = env
        self.buffer_config = buffer_config

        # Определяем тип буфера
        capacity = buffer_config.capacity if buffer_config.capacity is not None else simpy.core.Infinity
        self.buffer = simpy.Store(self.env, capacity)

    def get_item(self, **kwargs) -> Generator[simpy.Event, None, Part]:
        """Получаем объект из буфера"""
        item: Part = yield self.buffer.get()
        logger.info(f"Время: {self.env.now}, объект: {item.part_config.name} получена из "
                    f"буфера: {self.buffer_config.name}")
        return item

    def put_item(self, item: Part, **kwargs) -> Generator[simpy.Event, None, None]:
        """Кладем объект в буфер"""
        yield self.buffer.put(item)
        logger.info(f"Время: {self.env.now}, объект: {item.part_config.name} помещена в "
                    f"буфера: {self.buffer_config.name}")

    def get_buffer_level(self) -> Union[int, float]:
        """Текущее количество объектов в буфере"""
        return len(self.buffer.items)

    def get_items_list(self) -> List[Part]:
        """Список объектов в буфере"""
        return self.buffer.items.copy()

    def __repr__(self) -> str:
        return f"Buffer({self.env, self.buffer_config})\n"

    def __str__(self) -> str:
        return (f"id: {self.buffer_config.id}\n"
                f"name: {self.buffer_config.name}\n"
                f"capacity: {self.buffer_config.capacity}\n"
                f"metadata: {self.buffer_config.metadata}\n")


class BufferContainer(Buffer):
    def __init__(self, env: simpy.Environment, buffer_config):
        # Передаем окружение и настройки буфера
        self.env = env
        self.buffer_config = buffer_config

        # Определяем тип буфера
        capacity = buffer_config.capacity if buffer_config.capacity is not None else simpy.core.Infinity
        self.buffer = simpy.Container(self.env, self.buffer_config.init, capacity)

    def get_item(self, count: Union[int, float] = 1, **kwargs):
        """Получаем объект из буфера"""
        item = yield self.buffer.get(count)
        logger.info(f"Время: {self.env.now}, получено {item} объектов из буфера: {self.buffer_config.name}")
        return item

    def put_item(self, count: Union[int, float] = 1, **kwargs):
        """Кладем объект в буфер"""
        yield self.buffer.put(count)
        logger.info(f"Время: {self.env.now}, помещено {count} объектов в буфера: {self.buffer_config.name}")

    def get_buffer_level(self) -> Union[int, float]:
        """Текущее количество объектов в буфере"""
        return self.buffer.level

    def __repr__(self) -> str:
        return f"Buffer({self.env, self.buffer_config})\n"

    def __str__(self) -> str:
        return (f"id: {self.buffer_config.id}\n"
                f"name: {self.buffer_config.name}\n"
                f"capacity: {self.buffer_config.capacity}\n"
                f"metadata: {self.buffer_config.metadata}\n")

Код part.py:

import logging
from ..utils.validators import PartConfig

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class Part:
    def __init__(self, part_config: PartConfig):
        self.part_config = part_config

    def __repr__(self):
        return f"Part({self.part_config})"

    def __str__(self):
        return (f"id: {self.part_config.id}\n"
                f"name: {self.part_config.name}\n"
                f"path: {self.part_config.path}\n"
                f"metadata: {self.part_config.metadata}\n")

Код общего файла валидации validator.py:

from uuid import UUID, uuid4
from typing import Dict, List, Any, Optional, Union
from pydantic import BaseModel, Field


class BaseDataConfig(BaseModel):
    id: UUID = Field(default_factory=uuid4, description="ID объекта")
    name: str = Field(..., min_length=3, max_length=50, description="Имя объекта")
    metadata: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Остальные данные")


class ProcessConfig(BaseDataConfig):
    timeout: Union[int, float] = Field(..., description="Задержка имитирующая работу с объектом")
    strategies: Any = Field(description="Стратегия взаимодействия с объектами")


class BufferConfig(BaseDataConfig):
    capacity: Optional[Union[int, float]] = Field(default=None, gt=0, description="Емкость буфера")
    init: Optional[Union[int, float]] = Field(default=0, ge=0, description="Начальный уровень заполнения (для типа Container)")


class PartConfig(BaseDataConfig):
    path: List[UUID] = Field(default_factory=list, description="Путь детали по симуляции")

Ответы (1 шт):

Автор решения: Vitalizzare

Проблема с типизацией генератора, состоит в том, что у вас не согласован задекларированный в сигнатуре тип объекта, посылаемого генератору методом send, и тип переменной, которой вы передаёте этот объект (None vs. Part):

def get_item(self, **kwargs) -> Generator[simpy.Event, None, Part]:
                              #                        ^^^^
                              # Generator[YieldType, SendType, ReturnType]
    item: Part = yield self.buffer.get()    # <- YieldType
    #     ^^^^
    # SendType: здесь ожидается Part, хотя в сигнатуре заявлен None
    return item    # <- ReturnType

По идее, должно быть так:

def get_item(self, **kwargs) -> Generator[simpy.Event, Part, Part]:
                                                     # ^^^^
    ...

Вторая проблема с кодом состоит в отличии сигнатуры метода put_item в производном классе. В абстрактном методе родительского класса вы заявили имя item, тогда как в классе наследнике используете count. Поменяйте имя count на item:

 91     def put_item(self, item: Union[int, float] = 1, **kwargs):                                                                                 
                         # ^^^^
 92         """Кладем объект в буфер"""                                                                                                            
 93         yield self.buffer.put(item)                                                                                                            
                                # ^^^^
 94         logger.info(f"Время: {self.env.now}, помещено {item} объектов в буфера: {self.buffer_config.name}") 
                                                         # ^^^^
→ Ссылка