Типизация 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 шт):
Проблема с типизацией генератора, состоит в том, что у вас не согласован задекларированный в сигнатуре тип объекта, посылаемого генератору методом 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}")
# ^^^^