跳转至

存储

CellClass module-attribute 

CellClass = TypeVar('CellClass', bound=ICell)

DEFAULT_PYDANTIC_MODEL_CONFIG module-attribute 

DEFAULT_PYDANTIC_MODEL_CONFIG = ConfigDict(strict=False, arbitrary_types_allowed=True, extra='ignore', validate_by_alias=True, validate_by_name=True, coerce_numbers_to_str=True, alias_generator=camelize, populate_by_name=True)

RackClass module-attribute 

RackClass = TypeVar('RackClass', bound=IRack)

RackGroupClass module-attribute 

RackGroupClass = TypeVar('RackGroupClass', bound=IRackGroup)

RackType module-attribute 

RackType = NewType('RackType', str)

StorageProperties module-attribute 

StorageProperties = Union[RackGroupProperties, RackProperties, CellProperties]

Unit module-attribute 

Unit = NewType('Unit', str)

AsrsProperties

立库属性信息

源代码位于： logis/biz/sim/storage/model/__init__.py 
208
209
210
211
212
213
class AsrsProperties(RackProperties):
    """
    立库属性信息
    """

    type: Literal["一货架，一堆垛机", "两货架，一堆垛机"] = Field(alias="立库系统类型")

type class-attribute instance-attribute 

type: Literal['一货架，一堆垛机', '两货架，一堆垛机'] = Field(alias='立库系统类型')

Capacity

源代码位于： logis/data_type/unitable.py 
150
151
class Capacity(QuantifiedValue):
    pass

CellProperties

储位属性

源代码位于： logis/biz/sim/storage/model/__init__.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class CellProperties(ContainerMetadata):
    """
    储位属性
    """

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

    # # 储位编码
    # cell_id: Optional[str] = None
    # 货架编码
    rack_id: Optional[str] = None
    # 货架组编码
    rack_group_id: Optional[str] = None

    # 储位存储的货物信息
    stocks: List["QuantifiedStock"] = []

    def store(self, v: QuantifiedStock) -> bool:
        """
        存储货物
        """
        ok = self.can_store(v)
        if ok:
            self.stocks.append(v)
        return ok

    def retrieve(self, v: QuantifiedStock) -> bool:
        """
        取出货物
        """
        ok = self.can_retrieve(v)
        if ok:
            self.stocks.remove(v)
        return ok

model_config class-attribute instance-attribute 

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

rack_group_id class-attribute instance-attribute 

rack_group_id: Optional[str] = None

rack_id class-attribute instance-attribute 

rack_id: Optional[str] = None

stocks class-attribute instance-attribute 

stocks: List[QuantifiedStock] = []

retrieve 

retrieve(v: QuantifiedStock) -> bool

取出货物

源代码位于： logis/biz/sim/storage/model/__init__.py 
159
160
161
162
163
164
165
166
def retrieve(self, v: QuantifiedStock) -> bool:
    """
    取出货物
    """
    ok = self.can_retrieve(v)
    if ok:
        self.stocks.remove(v)
    return ok

store 

store(v: QuantifiedStock) -> bool

存储货物

源代码位于： logis/biz/sim/storage/model/__init__.py 
150
151
152
153
154
155
156
157
def store(self, v: QuantifiedStock) -> bool:
    """
    存储货物
    """
    ok = self.can_store(v)
    if ok:
        self.stocks.append(v)
    return ok

ContainerMetadata

容器元数据

源代码位于： logis/biz/sim/storage/model/__init__.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class ContainerMetadata(SpatialProps):
    """
    容器元数据
    """

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG
    id: Optional[str] = Field(
        validation_alias=AliasChoices("id", "平堆区编号", "货架编号", "储位编码"),
        default=None,
    )

    name: Optional[str] = Field(
        validation_alias=AliasChoices("名称", "name"), default=None
    )

    capacity: Optional[Capacity] = None
    occupied_capacity: Optional[Capacity] = None

    @field_validator("width", "height", "depth", mode="before")
    def check_width(cls: Type[QuantifiedValue], v, validation_info: ValidationInfo):
        """
        其实这里为了避免重复逻辑，就要通过反射来获取字段的类型，麻烦了点但是也长了见识
        """
        field = cls.model_fields.get(validation_info.field_name)
        # field_type = get_origin(field.annotation)
        field_types = get_args(field.annotation)
        field_types: List[Type[QuantifiedValue]] = list(
            filter(lambda t: issubclass(t, QuantifiedValue), field_types)
        )
        if isinstance(v, str) and field_types:
            v = field_types[0].parse_str(v)
        return v

    @field_validator("rotation", mode="before")
    def check_rotation(cls, v, validation_info: ValidationInfo):
        if not v:
            return None
        try:
            return QuantifiedValue(quantity=float(v), unit=None)
        except Exception as e:
            logging.warning("convert failed：%s", e)
        return None

    @property
    def free_capacity(self) -> Capacity:
        return (
            self.capacity - self.occupied_capacity
            if self.occupied_capacity
            else self.capacity
        )

    disabled: Optional[bool] = False

    store_speed_time: Optional[Time] = None
    # 如果不指定默认使用store_speed_time
    retrieve_speed_time: Optional[Time] = None

    # # 是否是互斥访问
    # exclusive: bool = True

    # 内部可以放置的物品类型
    allowed_stock_type: Optional[Union[str, List[str]]] = None

    @property
    def allowed_stock_types(self):
        """
        这里为了方便访问，返回一个固定格式的列表
        """
        if isinstance(self.allowed_stock_type, str):
            return [self.allowed_stock_type]
        elif isinstance(self.allowed_stock_type, list):
            return self.allowed_stock_type
        return ["*"]

    def get_cells(self) -> List["CellProperties"]:
        """
        获取所有的储位
        """
        if hasattr(self, "cells"):
            return self.cells
        return [self]

    def group_cell_by_stock_type(self) -> Dict[str, List["CellProperties"]]:
        dc = defaultdict(list)
        for cell in self.get_cells():
            for allowed_type in cell.allowed_stock_types:
                items = dc[allowed_type]
                items.append(cell)
                dc[allowed_type] = items
        return dc

    def is_allowed_stock_type(self, stock_type: str) -> bool:
        tps = self.allowed_stock_types
        return "*" in tps or stock_type in tps

    def can_store(self, v: QuantifiedStock) -> bool:
        """
        判断当前存储单元是否可以存储指定货物
        1. 类型符合
        2. 空间足够
        """
        return self.is_allowed_stock_type(v.unique_id) and self.free_capacity >= v

    def can_retrieve(self, v: QuantifiedStock) -> bool:
        """
        判断当前存储单元是否可以提供指定货物
        1. 类型符合
        2. 空间足够
        3. 未被占用
        """
        if self.occupied_capacity is None:
            return False
        return self.is_allowed_stock_type(v.unique_id) and self.occupied_capacity >= v

allowed_stock_type class-attribute instance-attribute 

allowed_stock_type: Optional[Union[str, List[str]]] = None

allowed_stock_types property 

allowed_stock_types

这里为了方便访问，返回一个固定格式的列表

capacity class-attribute instance-attribute 

capacity: Optional[Capacity] = None

disabled class-attribute instance-attribute 

disabled: Optional[bool] = False

free_capacity property 

free_capacity: Capacity

id class-attribute instance-attribute 

id: Optional[str] = Field(validation_alias=AliasChoices('id', '平堆区编号', '货架编号', '储位编码'), default=None)

model_config class-attribute instance-attribute 

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

name class-attribute instance-attribute 

name: Optional[str] = Field(validation_alias=AliasChoices('名称', 'name'), default=None)

occupied_capacity class-attribute instance-attribute 

occupied_capacity: Optional[Capacity] = None

retrieve_speed_time class-attribute instance-attribute 

retrieve_speed_time: Optional[Time] = None

store_speed_time class-attribute instance-attribute 

store_speed_time: Optional[Time] = None

can_retrieve 

can_retrieve(v: QuantifiedStock) -> bool

判断当前存储单元是否可以提供指定货物 1. 类型符合 2. 空间足够 3. 未被占用

源代码位于： logis/biz/sim/storage/model/__init__.py 
121
122
123
124
125
126
127
128
129
130
def can_retrieve(self, v: QuantifiedStock) -> bool:
    """
    判断当前存储单元是否可以提供指定货物
    1. 类型符合
    2. 空间足够
    3. 未被占用
    """
    if self.occupied_capacity is None:
        return False
    return self.is_allowed_stock_type(v.unique_id) and self.occupied_capacity >= v

can_store 

can_store(v: QuantifiedStock) -> bool

判断当前存储单元是否可以存储指定货物 1. 类型符合 2. 空间足够

源代码位于： logis/biz/sim/storage/model/__init__.py 
113
114
115
116
117
118
119
def can_store(self, v: QuantifiedStock) -> bool:
    """
    判断当前存储单元是否可以存储指定货物
    1. 类型符合
    2. 空间足够
    """
    return self.is_allowed_stock_type(v.unique_id) and self.free_capacity >= v

check_rotation 

check_rotation(v, validation_info: ValidationInfo)
源代码位于： logis/biz/sim/storage/model/__init__.py 
51
52
53
54
55
56
57
58
59
@field_validator("rotation", mode="before")
def check_rotation(cls, v, validation_info: ValidationInfo):
    if not v:
        return None
    try:
        return QuantifiedValue(quantity=float(v), unit=None)
    except Exception as e:
        logging.warning("convert failed：%s", e)
    return None

check_width 

check_width(v, validation_info: ValidationInfo)

其实这里为了避免重复逻辑，就要通过反射来获取字段的类型，麻烦了点但是也长了见识

源代码位于： logis/biz/sim/storage/model/__init__.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@field_validator("width", "height", "depth", mode="before")
def check_width(cls: Type[QuantifiedValue], v, validation_info: ValidationInfo):
    """
    其实这里为了避免重复逻辑，就要通过反射来获取字段的类型，麻烦了点但是也长了见识
    """
    field = cls.model_fields.get(validation_info.field_name)
    # field_type = get_origin(field.annotation)
    field_types = get_args(field.annotation)
    field_types: List[Type[QuantifiedValue]] = list(
        filter(lambda t: issubclass(t, QuantifiedValue), field_types)
    )
    if isinstance(v, str) and field_types:
        v = field_types[0].parse_str(v)
    return v

get_cells 

get_cells() -> List[CellProperties]

获取所有的储位

源代码位于： logis/biz/sim/storage/model/__init__.py 
92
93
94
95
96
97
98
def get_cells(self) -> List["CellProperties"]:
    """
    获取所有的储位
    """
    if hasattr(self, "cells"):
        return self.cells
    return [self]

group_cell_by_stock_type 

group_cell_by_stock_type() -> Dict[str, List[CellProperties]]
源代码位于： logis/biz/sim/storage/model/__init__.py 
100
101
102
103
104
105
106
107
def group_cell_by_stock_type(self) -> Dict[str, List["CellProperties"]]:
    dc = defaultdict(list)
    for cell in self.get_cells():
        for allowed_type in cell.allowed_stock_types:
            items = dc[allowed_type]
            items.append(cell)
            dc[allowed_type] = items
    return dc

is_allowed_stock_type 

is_allowed_stock_type(stock_type: str) -> bool
源代码位于： logis/biz/sim/storage/model/__init__.py 
109
110
111
def is_allowed_stock_type(self, stock_type: str) -> bool:
    tps = self.allowed_stock_types
    return "*" in tps or stock_type in tps

DefaultCellSelectionStrategy

默认储位选择策略

源代码位于： logis/biz/sim/storage/iface/select.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class DefaultCellSelectionStrategy(ICellSelectionStrategy):
    """
    默认储位选择策略
    """

    def select_cells(
        self,
        operation: OperationType,
        stock: IStock,
        rack: Optional[RackClass] = None,
        strategy: Optional[StorageSelectionStrategy] = None,
        **kwargs,
    ) -> List[CellClass]:
        rack = rack or self.rack
        assert rack is not None, "未传入货架，无法选择储位"
        cells = [
            cell for cell in rack.cells if cell.is_able_to(operation, stock, **kwargs)
        ]
        if not cells:
            return cells
        if StorageSelectionStrategy.NumberAscend.matches(strategy):
            if OperationType.Store.matches(operation):
                cells.sort(
                    key=lambda x: (
                        1 if x.is_still_universal else 0,
                        -x.target_quantity.capacity + x.target_quantity.level,
                        x.id,
                    )
                )
            else:
                cells.sort(key=lambda x: (-x.target_quantity.level, x.id))
        elif StorageSelectionStrategy.DistanceAscend.matches(strategy):
            if OperationType.Store.matches(operation):
                cells.sort(
                    key=lambda x: (
                        0 if x.is_still_universal else 1,
                        x.target_quantity.capacity - x.target_quantity.level,
                        x.id,
                    ),
                    reverse=True,
                )
            else:
                cells.sort(
                    key=lambda x: (
                        x.target_quantity.level,
                        x.id,
                    ),
                    reverse=True,
                )
        else:
            raise NotImplementedError(f"尚未支持的储位选择策略:{strategy}")
        return cells

select_cells 

select_cells(operation: OperationType, stock: IStock, rack: Optional[RackClass] = None, strategy: Optional[StorageSelectionStrategy] = None, **kwargs) -> List[CellClass]
源代码位于： logis/biz/sim/storage/iface/select.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def select_cells(
    self,
    operation: OperationType,
    stock: IStock,
    rack: Optional[RackClass] = None,
    strategy: Optional[StorageSelectionStrategy] = None,
    **kwargs,
) -> List[CellClass]:
    rack = rack or self.rack
    assert rack is not None, "未传入货架，无法选择储位"
    cells = [
        cell for cell in rack.cells if cell.is_able_to(operation, stock, **kwargs)
    ]
    if not cells:
        return cells
    if StorageSelectionStrategy.NumberAscend.matches(strategy):
        if OperationType.Store.matches(operation):
            cells.sort(
                key=lambda x: (
                    1 if x.is_still_universal else 0,
                    -x.target_quantity.capacity + x.target_quantity.level,
                    x.id,
                )
            )
        else:
            cells.sort(key=lambda x: (-x.target_quantity.level, x.id))
    elif StorageSelectionStrategy.DistanceAscend.matches(strategy):
        if OperationType.Store.matches(operation):
            cells.sort(
                key=lambda x: (
                    0 if x.is_still_universal else 1,
                    x.target_quantity.capacity - x.target_quantity.level,
                    x.id,
                ),
                reverse=True,
            )
        else:
            cells.sort(
                key=lambda x: (
                    x.target_quantity.level,
                    x.id,
                ),
                reverse=True,
            )
    else:
        raise NotImplementedError(f"尚未支持的储位选择策略:{strategy}")
    return cells

DefaultRackSelectionStrategy

默认货架选择策略

源代码位于： logis/biz/sim/storage/iface/select.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class DefaultRackSelectionStrategy(IRackSelectionStrategy):
    """
    默认货架选择策略
    """

    def select_racks(
        self,
        operation: OperationType,
        stock: IStock,
        rack_group: Optional[RackGroupClass] = None,
        strategy: Optional[StorageSelectionStrategy] = None,
        **kwargs,
    ):
        """
        默认货架选择策略

        Args:
            operation: 操作类型
            stock: 货物
            rack_group: 货架组，如果不传则使用初始化时传的货架组
            strategy: 此方式实际上不是规范的策略模式，仅仅是为了少写代码

        Returns:
            符合操作需求的货架列表
        """
        racks = super().select_racks(
            operation=operation, stock=stock, rack_group=rack_group, **kwargs
        )
        if not racks:
            return racks
        if StorageSelectionStrategy.BusyLevelAscend.matches(strategy):
            racks.sort(key=lambda x: x.current_jobs)
        elif StorageSelectionStrategy.DistanceAscend.matches(strategy):
            racks.sort(key=lambda x: x.distance_to(stock=stock, **kwargs))
        return racks

select_racks 

select_racks(operation: OperationType, stock: IStock, rack_group: Optional[RackGroupClass] = None, strategy: Optional[StorageSelectionStrategy] = None, **kwargs)

默认货架选择策略

参数：

名称 类型 描述 默认
operation OperationType

操作类型

 必需
stock IStock

货物

 必需
rack_group Optional[RackGroupClass]

货架组，如果不传则使用初始化时传的货架组

 None
strategy Optional[StorageSelectionStrategy]

此方式实际上不是规范的策略模式，仅仅是为了少写代码

 None

返回：

类型 描述

符合操作需求的货架列表
源代码位于： logis/biz/sim/storage/iface/select.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def select_racks(
    self,
    operation: OperationType,
    stock: IStock,
    rack_group: Optional[RackGroupClass] = None,
    strategy: Optional[StorageSelectionStrategy] = None,
    **kwargs,
):
    """
    默认货架选择策略

    Args:
        operation: 操作类型
        stock: 货物
        rack_group: 货架组，如果不传则使用初始化时传的货架组
        strategy: 此方式实际上不是规范的策略模式，仅仅是为了少写代码

    Returns:
        符合操作需求的货架列表
    """
    racks = super().select_racks(
        operation=operation, stock=stock, rack_group=rack_group, **kwargs
    )
    if not racks:
        return racks
    if StorageSelectionStrategy.BusyLevelAscend.matches(strategy):
        racks.sort(key=lambda x: x.current_jobs)
    elif StorageSelectionStrategy.DistanceAscend.matches(strategy):
        racks.sort(key=lambda x: x.distance_to(stock=stock, **kwargs))
    return racks

ICell

储位

源代码位于： logis/biz/sim/storage/iface/device.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
class ICell(IStorage):
    """
    储位
    """

    @property
    @abstractmethod
    def id(self):
        pass

    @property
    @abstractmethod
    def is_still_universal(self) -> bool:
        pass

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.current_quantity: Optional[simpy.Container] = None
        self.target_quantity: Optional[simpy.Container] = None

    @abstractmethod
    def is_able_to(
        self, operation: OperationType, stock: IStock, strict: bool = False, **kwargs
    ) -> bool:
        """
        判断是否可操作

        Args:
            operation: 操作类型
            stock: 货物
            strict: 是否严格判断，即完全能满足所有货物的需求

        Returns:
            是否可操作
        """
        pass

    @property
    def rack(self) -> Optional["IRack"]:
        raise NotImplementedError("rack not implemented")

current_quantity instance-attribute 

current_quantity: Optional[Container] = None

id abstractmethod property 

id

is_still_universal abstractmethod property 

is_still_universal: bool

rack property 

rack: Optional[IRack]

target_quantity instance-attribute 

target_quantity: Optional[Container] = None

__init__ 

__init__(*args, **kwargs)
源代码位于： logis/biz/sim/storage/iface/device.py 
165
166
167
168
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.current_quantity: Optional[simpy.Container] = None
    self.target_quantity: Optional[simpy.Container] = None

is_able_to abstractmethod 

is_able_to(operation: OperationType, stock: IStock, strict: bool = False, **kwargs) -> bool

判断是否可操作

参数：

名称 类型 描述 默认
operation OperationType

操作类型

 必需
stock IStock

货物

 必需
strict bool

是否严格判断，即完全能满足所有货物的需求

 False

返回：

类型 描述
bool

是否可操作
源代码位于： logis/biz/sim/storage/iface/device.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@abstractmethod
def is_able_to(
    self, operation: OperationType, stock: IStock, strict: bool = False, **kwargs
) -> bool:
    """
    判断是否可操作

    Args:
        operation: 操作类型
        stock: 货物
        strict: 是否严格判断，即完全能满足所有货物的需求

    Returns:
        是否可操作
    """
    pass

ICellSelectionStrategy

储位选择策略，从若干储位中筛选出符合操作需求的储位

源代码位于： logis/biz/sim/storage/iface/select.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class ICellSelectionStrategy(IExpose):
    """
    储位选择策略，从若干储位中筛选出符合操作需求的储位
    """

    def __init__(self, rack: Optional[RackClass] = None, **kwargs) -> None:
        """
        初始化储位选择策略

        Args:
            rack: 货架，如果此策略仅针对特定货架则建议传，否则忽略即可
        """
        super().__init__(**kwargs)
        self.rack = rack

    @abstractmethod
    def select_cells(
        self,
        operation: OperationType,
        stock: IStock,
        rack: Optional[RackClass] = None,
        **kwargs,
    ) -> List[CellClass]:
        """
        从所有可操作的储位中筛选出符合操作需求的储位

        Args:
            operation: 操作类型
            stocks: 货物列表
            rack: 货架，如果不传则使用初始化时传的货架

        Returns:
            符合操作需求的储位列表
        """
        pass

rack instance-attribute 

rack = rack

__init__ 

__init__(rack: Optional[RackClass] = None, **kwargs) -> None

初始化储位选择策略

参数：

名称 类型 描述 默认
rack Optional[RackClass]

货架，如果此策略仅针对特定货架则建议传，否则忽略即可

 None
源代码位于： logis/biz/sim/storage/iface/select.py 
 99
100
101
102
103
104
105
106
107
def __init__(self, rack: Optional[RackClass] = None, **kwargs) -> None:
    """
    初始化储位选择策略

    Args:
        rack: 货架，如果此策略仅针对特定货架则建议传，否则忽略即可
    """
    super().__init__(**kwargs)
    self.rack = rack

select_cells abstractmethod 

select_cells(operation: OperationType, stock: IStock, rack: Optional[RackClass] = None, **kwargs) -> List[CellClass]

从所有可操作的储位中筛选出符合操作需求的储位

参数：

名称 类型 描述 默认
operation OperationType

操作类型

 必需
stocks

货物列表

 必需
rack Optional[RackClass]

货架，如果不传则使用初始化时传的货架

 None

返回：

类型 描述
List[CellClass]

符合操作需求的储位列表
源代码位于： logis/biz/sim/storage/iface/select.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@abstractmethod
def select_cells(
    self,
    operation: OperationType,
    stock: IStock,
    rack: Optional[RackClass] = None,
    **kwargs,
) -> List[CellClass]:
    """
    从所有可操作的储位中筛选出符合操作需求的储位

    Args:
        operation: 操作类型
        stocks: 货物列表
        rack: 货架，如果不传则使用初始化时传的货架

    Returns:
        符合操作需求的储位列表
    """
    pass

IExpose

用于通过SDK向外暴露参数

源代码位于： logis/biz/sim/iface/base.py 
14
15
16
17
18
19
20
class IExpose(ABC):
    """
    用于通过SDK向外暴露参数
    """

    def __init__(self, ctx: "Context", **kwargs) -> None:
        self.ctx = ctx

ctx instance-attribute 

ctx = ctx

__init__ 

__init__(ctx: Context, **kwargs) -> None
源代码位于： logis/biz/sim/iface/base.py 
19
20
def __init__(self, ctx: "Context", **kwargs) -> None:
    self.ctx = ctx

IRack

货架

源代码位于： logis/biz/sim/storage/iface/device.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
class IRack(IStorage):
    """
    货架
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    @property
    @abstractmethod
    def cells(self) -> List[ICell]:
        """
        获取本货架下的所有储位
        """
        pass

    @abstractmethod
    def is_able_to(
        self, operation: OperationType, stock: IStock, strict: bool = False, **kwargs
    ) -> bool:
        """
        判断是否能完成某个操作

        Args:
            operation: 操作类型
            stock: 货物
            strict: 是否严格判断，即完全能满足所有货物的需求

        Returns:
            是否能完成操作
        """
        pass

    @property
    def rack_group(self) -> Optional["IRackGroup"]:
        raise NotImplementedError("rack_group not implemented")

cells abstractmethod property 

cells: List[ICell]

获取本货架下的所有储位

rack_group property 

rack_group: Optional[IRackGroup]

__init__ 

__init__(*args, **kwargs)
源代码位于： logis/biz/sim/storage/iface/device.py 
200
201
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)

is_able_to abstractmethod 

is_able_to(operation: OperationType, stock: IStock, strict: bool = False, **kwargs) -> bool

判断是否能完成某个操作

参数：

名称 类型 描述 默认
operation OperationType

操作类型

 必需
stock IStock

货物

 必需
strict bool

是否严格判断，即完全能满足所有货物的需求

 False

返回：

类型 描述
bool

是否能完成操作
源代码位于： logis/biz/sim/storage/iface/device.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
@abstractmethod
def is_able_to(
    self, operation: OperationType, stock: IStock, strict: bool = False, **kwargs
) -> bool:
    """
    判断是否能完成某个操作

    Args:
        operation: 操作类型
        stock: 货物
        strict: 是否严格判断，即完全能满足所有货物的需求

    Returns:
        是否能完成操作
    """
    pass

IRackGroup

货架组

源代码位于： logis/biz/sim/storage/iface/device.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
class IRackGroup(IStorage):
    """
    货架组
    """

    @property
    @abstractmethod
    def racks(self) -> List[IRack]:
        """
        获取本货架组下的所有货架
        """
        pass

    def handle_operation(
        self,
        operation: Union[str, OperationType],
        stock: IStock,
        strategy: Optional[str] = None,
        agent: Optional["IAgent"] = None,
        force_assign_location: bool = False,
        block: bool = False,
        rack_selection_strategy: Optional["IRackSelectionStrategy"] = None,
        cell_selection_strategy: Optional["ICellSelectionStrategy"] = None,
        **kwargs,
    ) -> Generator[simpy.Event, Any, Tuple[List["IStock"], Optional["IStock"]]]:
        """
        完成指定操作

        Args:
            operation: 操作类型
            stock: 货物
            strategy: 操作策略,此字段已废弃，仅保留兼容性
            agent: 所使用的智能体，可选
            force_assign_location: 是否强制分配位置
            block: 是否等待存取完成
            rack_selection_strategy: 货架选择策略
            cell_selection_strategy: 储位选择策略

        Returns:
            (List["IStock"], Optional["IStock"]): (成功操作的货物列表, 失败的货物)
        """
        raise NotImplementedError("handle_operation not implemented")

    def handle_operation_until(
        self,
        operation: OperationType,
        stock: "IStock",
        try_interval=5,
        max_try: Optional[int] = 1,
        stop_event: Optional[simpy.Event] = None,
        **kwargs,
    ):
        """
        内部调用handle_operation，直到成功或超过最大尝试次数

        Args:
            operation: 操作类型
            stock: 货物
            try_interval: 尝试间隔，单位秒
            max_try: 最大尝试次数,默认1次
            stop_event: 停止事件，触发后停止尝试
            **kwargs: 其他参数，传递给handle_operation

        """
        try_forever = max_try is None
        result = []
        try_count = 0
        while True:
            if stop_event and stop_event.triggered:
                break
            if not try_forever and try_count >= max_try:
                break
            stocks, stock_left = yield from self.handle_operation(
                operation, stock, **kwargs
            )
            result.extend(stocks)
            stock = stock_left
            if not try_forever or not stock_left:
                break
            try_count += 1
            if try_forever or try_count < max_try:
                yield self.env.timeout(try_interval)
        return result, stock

racks abstractmethod property 

racks: List[IRack]

获取本货架组下的所有货架

handle_operation 

handle_operation(operation: Union[str, OperationType], stock: IStock, strategy: Optional[str] = None, agent: Optional[IAgent] = None, force_assign_location: bool = False, block: bool = False, rack_selection_strategy: Optional[IRackSelectionStrategy] = None, cell_selection_strategy: Optional[ICellSelectionStrategy] = None, **kwargs) -> Generator[simpy.Event, Any, Tuple[List[IStock], Optional[IStock]]]

完成指定操作

参数：

名称 类型 描述 默认
operation Union[str, OperationType]

操作类型

 必需
stock IStock

货物

 必需
strategy Optional[str]

操作策略,此字段已废弃，仅保留兼容性

 None
agent Optional[IAgent]

所使用的智能体，可选

 None
force_assign_location bool

是否强制分配位置

 False
block bool

是否等待存取完成

 False
rack_selection_strategy Optional[IRackSelectionStrategy]

货架选择策略

 None
cell_selection_strategy Optional[ICellSelectionStrategy]

储位选择策略

 None

返回：

类型 描述
(List[IStock], Optional[IStock])

(成功操作的货物列表, 失败的货物)
源代码位于： logis/biz/sim/storage/iface/device.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def handle_operation(
    self,
    operation: Union[str, OperationType],
    stock: IStock,
    strategy: Optional[str] = None,
    agent: Optional["IAgent"] = None,
    force_assign_location: bool = False,
    block: bool = False,
    rack_selection_strategy: Optional["IRackSelectionStrategy"] = None,
    cell_selection_strategy: Optional["ICellSelectionStrategy"] = None,
    **kwargs,
) -> Generator[simpy.Event, Any, Tuple[List["IStock"], Optional["IStock"]]]:
    """
    完成指定操作

    Args:
        operation: 操作类型
        stock: 货物
        strategy: 操作策略,此字段已废弃，仅保留兼容性
        agent: 所使用的智能体，可选
        force_assign_location: 是否强制分配位置
        block: 是否等待存取完成
        rack_selection_strategy: 货架选择策略
        cell_selection_strategy: 储位选择策略

    Returns:
        (List["IStock"], Optional["IStock"]): (成功操作的货物列表, 失败的货物)
    """
    raise NotImplementedError("handle_operation not implemented")

handle_operation_until 

handle_operation_until(operation: OperationType, stock: IStock, try_interval=5, max_try: Optional[int] = 1, stop_event: Optional[Event] = None, **kwargs)

内部调用handle_operation，直到成功或超过最大尝试次数

参数：

名称 类型 描述 默认
operation OperationType

操作类型

 必需
stock IStock

货物

 必需
try_interval

尝试间隔，单位秒

 5
max_try Optional[int]

最大尝试次数,默认1次

 1
stop_event Optional[Event]

停止事件，触发后停止尝试

 None
**kwargs

其他参数，传递给handle_operation

 {}
源代码位于： logis/biz/sim/storage/iface/device.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def handle_operation_until(
    self,
    operation: OperationType,
    stock: "IStock",
    try_interval=5,
    max_try: Optional[int] = 1,
    stop_event: Optional[simpy.Event] = None,
    **kwargs,
):
    """
    内部调用handle_operation，直到成功或超过最大尝试次数

    Args:
        operation: 操作类型
        stock: 货物
        try_interval: 尝试间隔，单位秒
        max_try: 最大尝试次数,默认1次
        stop_event: 停止事件，触发后停止尝试
        **kwargs: 其他参数，传递给handle_operation

    """
    try_forever = max_try is None
    result = []
    try_count = 0
    while True:
        if stop_event and stop_event.triggered:
            break
        if not try_forever and try_count >= max_try:
            break
        stocks, stock_left = yield from self.handle_operation(
            operation, stock, **kwargs
        )
        result.extend(stocks)
        stock = stock_left
        if not try_forever or not stock_left:
            break
        try_count += 1
        if try_forever or try_count < max_try:
            yield self.env.timeout(try_interval)
    return result, stock

IRackSelectionStrategy

货架选择策略，从若干货架中筛选出符合操作需求的货架

源代码位于： logis/biz/sim/storage/iface/select.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class IRackSelectionStrategy(IExpose):
    """
    货架选择策略，从若干货架中筛选出符合操作需求的货架
    """

    def __init__(
        self, rack_group: Union[RackGroupClass, None] = None, **kwargs
    ) -> None:
        """
        初始化货架选择策略

        Args:
            rack_group: 货架组，如果此策略仅针对特定货架组则建议传，否则忽略即可
        """
        super().__init__(**kwargs)
        self.rack_group = rack_group

    @abstractmethod
    def select_racks(
        self,
        operation: OperationType,
        stock: IStock,
        rack_group: Optional[RackGroupClass] = None,
        **kwargs,
    ) -> List[IRack]:
        """
        从所有可操作的货架中筛选出符合操作需求的货架

        Args:
            operation: 操作类型
            stocks: 货物列表
            rack_group: 货架组，如果不传则使用初始化时传的货架组

        Returns:
            符合操作需求的货架列表
        """
        rack_group = rack_group or self.rack_group
        assert rack_group is not None, "未传入货架组，无法选择货架"
        result: List[RackClass] = []
        for rack in rack_group.racks:
            if not rack.is_able_to(operation, stock):
                continue
            result.append(rack)
        return result

rack_group instance-attribute 

rack_group = rack_group

__init__ 

__init__(rack_group: Union[RackGroupClass, None] = None, **kwargs) -> None

初始化货架选择策略

参数：

名称 类型 描述 默认
rack_group Union[RackGroupClass, None]

货架组，如果此策略仅针对特定货架组则建议传，否则忽略即可

 None
源代码位于： logis/biz/sim/storage/iface/select.py 
16
17
18
19
20
21
22
23
24
25
26
def __init__(
    self, rack_group: Union[RackGroupClass, None] = None, **kwargs
) -> None:
    """
    初始化货架选择策略

    Args:
        rack_group: 货架组，如果此策略仅针对特定货架组则建议传，否则忽略即可
    """
    super().__init__(**kwargs)
    self.rack_group = rack_group

select_racks abstractmethod 

select_racks(operation: OperationType, stock: IStock, rack_group: Optional[RackGroupClass] = None, **kwargs) -> List[IRack]

从所有可操作的货架中筛选出符合操作需求的货架

参数：

名称 类型 描述 默认
operation OperationType

操作类型

 必需
stocks

货物列表

 必需
rack_group Optional[RackGroupClass]

货架组，如果不传则使用初始化时传的货架组

 None

返回：

类型 描述
List[IRack]

符合操作需求的货架列表
源代码位于： logis/biz/sim/storage/iface/select.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@abstractmethod
def select_racks(
    self,
    operation: OperationType,
    stock: IStock,
    rack_group: Optional[RackGroupClass] = None,
    **kwargs,
) -> List[IRack]:
    """
    从所有可操作的货架中筛选出符合操作需求的货架

    Args:
        operation: 操作类型
        stocks: 货物列表
        rack_group: 货架组，如果不传则使用初始化时传的货架组

    Returns:
        符合操作需求的货架列表
    """
    rack_group = rack_group or self.rack_group
    assert rack_group is not None, "未传入货架组，无法选择货架"
    result: List[RackClass] = []
    for rack in rack_group.racks:
        if not rack.is_able_to(operation, stock):
            continue
        result.append(rack)
    return result

IStock

库存\货物接口

源代码位于： logis/biz/sim/stock/model/__init__.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class IStock(NumberUnit, metaclass=ABCMeta):
    # class IStock(QuantifiedStock):
    """
    库存\货物接口
    """

    order_id: Optional[str] = None
    task_id: Optional[str] = None

    name: str
    code: Optional[str] = None

    stage: Optional[str] = None

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.__stage__: Optional[Literal["pickup", "delivery"]] = kwargs.get(
            "__stage__"
        )
        """内部状态变量"""

    @property
    def unique_id(self):
        return self.code or self.name

    @property
    def target_location(self) -> LocationType:
        raise NotImplementedError()

    @target_location.setter
    def target_location(self, location: LocationType):
        raise NotImplementedError()

    @property
    def current_location(self) -> LocationType:
        raise NotImplementedError()

    @current_location.setter
    def current_location(self, location: LocationType):
        raise NotImplementedError()

    @property
    def final_target_location(self) -> LocationType:
        """
        起初只有当前位置和目标位置，但是后来发现有些场景下目标位置只是短期目标，短期目标会经常更换
        """
        raise NotImplementedError()

    @final_target_location.setter
    def final_target_location(self, loc: LocationType):
        raise NotImplementedError()

    @property
    def storage_key(self) -> Tuple[str, str]:
        """
        用于存储索引的key，格式为(name, unit)，
        """
        return (self.name, self.unit)

__stage__ instance-attribute 

__stage__: Optional[Literal['pickup', 'delivery']] = get('__stage__')

内部状态变量

code class-attribute instance-attribute 

code: Optional[str] = None

current_location property writable 

current_location: LocationType

final_target_location property writable 

final_target_location: LocationType

起初只有当前位置和目标位置，但是后来发现有些场景下目标位置只是短期目标，短期目标会经常更换

name instance-attribute 

name: str

order_id class-attribute instance-attribute 

order_id: Optional[str] = None

stage class-attribute instance-attribute 

stage: Optional[str] = None

storage_key property 

storage_key: Tuple[str, str]

用于存储索引的key，格式为(name, unit)，

target_location property writable 

target_location: LocationType

task_id class-attribute instance-attribute 

task_id: Optional[str] = None

unique_id property 

unique_id

__init__ 

__init__(*args, **kwargs)
源代码位于： logis/biz/sim/stock/model/__init__.py 
54
55
56
57
58
59
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.__stage__: Optional[Literal["pickup", "delivery"]] = kwargs.get(
        "__stage__"
    )
    """内部状态变量"""

IStorage

存储设备

源代码位于： logis/biz/sim/storage/iface/device.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class IStorage(Storable):
    """
    存储设备
    """

    store_strategy: Optional[StoreStrategy] = StoreStrategy()
    retrieve_strategy: Optional[RetrieveStrategy] = RetrieveStrategy()

    props: Optional[StorageProperties] = None

    __resource__: Optional[simpy.Resource] = None

    # 表明当前存储空间是否被占用
    __occupied__: Optional[simpy.Resource] = None

    __container__: Optional[QuantifiedContainer] = None

    log: logging.Logger

    @property
    @abstractmethod
    def env(self) -> simpy.Environment:
        pass

    def __init__(
        self,
        props: Optional[StorageProperties] = None,
        exclusive: bool = True,
        **kwargs
    ):
        """

        Args:
            props: 存储设备的属性
            exclusive: 是否独占，适用于储位、货架、货架组等
        """
        self.props = props
        # 这里以一个不可能达到的数值表示共享
        self.__occupied__ = simpy.Resource(self.env, 1 if exclusive else 10**12)
        if props:
            self.__container__ = QuantifiedContainer(props.capacity, env=self.env)
        else:
            self.__container__ = None
        self.center_point: Optional[Point] = None
        self.current_jobs: int = 0

    def decrease_jobs(self, *args, **kwargs):
        """
        减少作业数
        """
        self.current_jobs -= 1

    def increase_jobs(self, *args, **kwargs):
        """
        增加作业数
        """
        self.current_jobs += 1

    @deprecated("use agent.distance_to instead")
    def distance_to(self, **kwargs) -> float:
        """
        获取到货物源头的距离
        TODO：其实这里按理说应该包括取货距离和送货距离，且放在这里似乎也不太合适
        """
        from logis.biz.sim.agent import IAgent

        # TODO: 考虑过stock不再继承IAgent，所以这里也要考虑改变
        x: IAgent = kwargs.get("agent", None) or kwargs.get("stock", None)
        return euclid_distance(self.center_point, x.resolve_center_point())

    @property
    def level(self):
        return self.__container__.level

    @property
    def capacity(self):
        return self.__container__.capacity

    @property
    def unit(self):
        return self.__container__.unit

    @property
    def free_capacity(self):
        return self.__container__.free_capacity

    def store(self, v: QuantifiedStock, *args, **kwargs):
        """
        存储
        """
        log = self.log
        v = unify_quantified_value(v, self.unit)
        cells = self.store_strategy.find_location(self.props, [v])
        with self.__occupied__.request() as req:
            # yield req
            if self.free_capacity < v.quantity:
                log.warning("%s < %s, skip store", self.free_capacity, v.quantity)
                yield self.env.timeout(0)
                return
            for cell in cells:
                cell.store(v)
            yield self.__container__.put(v)
            t = self.props.store_speed_time
            yield self.env.timeout(0 if not t else t.value)

    def retrieve(self, v: QuantifiedStock, *args, **kwargs):
        """
        取回
        """
        log = self.log
        v = unify_quantified_value(v, self.unit)
        cells = self.retrieve_strategy.find_location(self.props, [v])
        with self.__occupied__.request() as req:
            yield req
            if v.quantity > self.level:
                log.warning("%s < %s, skip retrieve", self.level, v.quantity)
                yield self.env.timeout(0)
                return
            for cell in cells:
                cell.retrieve(v)
            yield self.__container__.get(v)
            t = self.props.retrieve_speed_time or self.props.store_speed_time
            yield self.env.timeout(0 if not t else t.value)

__container__ class-attribute instance-attribute 

__container__: Optional[QuantifiedContainer] = None

__occupied__ class-attribute instance-attribute 

__occupied__: Optional[Resource] = Resource(env, 1 if exclusive else 10 ** 12)

__resource__ class-attribute instance-attribute 

__resource__: Optional[Resource] = None

capacity property 

capacity

center_point instance-attribute 

center_point: Optional[Point] = None

current_jobs instance-attribute 

current_jobs: int = 0

env abstractmethod property 

env: Environment

free_capacity property 

free_capacity

level property 

level

log instance-attribute 

log: Logger

props class-attribute instance-attribute 

props: Optional[StorageProperties] = props

retrieve_strategy class-attribute instance-attribute 

retrieve_strategy: Optional[RetrieveStrategy] = RetrieveStrategy()

store_strategy class-attribute instance-attribute 

store_strategy: Optional[StoreStrategy] = StoreStrategy()

unit property 

unit

__init__ 

__init__(props: Optional[StorageProperties] = None, exclusive: bool = True, **kwargs)

参数：

名称 类型 描述 默认
props Optional[StorageProperties]

存储设备的属性

 None
exclusive bool

是否独占，适用于储位、货架、货架组等

 True
源代码位于： logis/biz/sim/storage/iface/device.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(
    self,
    props: Optional[StorageProperties] = None,
    exclusive: bool = True,
    **kwargs
):
    """

    Args:
        props: 存储设备的属性
        exclusive: 是否独占，适用于储位、货架、货架组等
    """
    self.props = props
    # 这里以一个不可能达到的数值表示共享
    self.__occupied__ = simpy.Resource(self.env, 1 if exclusive else 10**12)
    if props:
        self.__container__ = QuantifiedContainer(props.capacity, env=self.env)
    else:
        self.__container__ = None
    self.center_point: Optional[Point] = None
    self.current_jobs: int = 0

decrease_jobs 

decrease_jobs(*args, **kwargs)

减少作业数

源代码位于： logis/biz/sim/storage/iface/device.py 
71
72
73
74
75
def decrease_jobs(self, *args, **kwargs):
    """
    减少作业数
    """
    self.current_jobs -= 1

distance_to 

distance_to(**kwargs) -> float

获取到货物源头的距离 TODO：其实这里按理说应该包括取货距离和送货距离，且放在这里似乎也不太合适

源代码位于： logis/biz/sim/storage/iface/device.py 
83
84
85
86
87
88
89
90
91
92
93
@deprecated("use agent.distance_to instead")
def distance_to(self, **kwargs) -> float:
    """
    获取到货物源头的距离
    TODO：其实这里按理说应该包括取货距离和送货距离，且放在这里似乎也不太合适
    """
    from logis.biz.sim.agent import IAgent

    # TODO: 考虑过stock不再继承IAgent，所以这里也要考虑改变
    x: IAgent = kwargs.get("agent", None) or kwargs.get("stock", None)
    return euclid_distance(self.center_point, x.resolve_center_point())

increase_jobs 

increase_jobs(*args, **kwargs)

增加作业数

源代码位于： logis/biz/sim/storage/iface/device.py 
77
78
79
80
81
def increase_jobs(self, *args, **kwargs):
    """
    增加作业数
    """
    self.current_jobs += 1

retrieve 

retrieve(v: QuantifiedStock, *args, **kwargs)

取回

源代码位于： logis/biz/sim/storage/iface/device.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def retrieve(self, v: QuantifiedStock, *args, **kwargs):
    """
    取回
    """
    log = self.log
    v = unify_quantified_value(v, self.unit)
    cells = self.retrieve_strategy.find_location(self.props, [v])
    with self.__occupied__.request() as req:
        yield req
        if v.quantity > self.level:
            log.warning("%s < %s, skip retrieve", self.level, v.quantity)
            yield self.env.timeout(0)
            return
        for cell in cells:
            cell.retrieve(v)
        yield self.__container__.get(v)
        t = self.props.retrieve_speed_time or self.props.store_speed_time
        yield self.env.timeout(0 if not t else t.value)

store 

store(v: QuantifiedStock, *args, **kwargs)

存储

源代码位于： logis/biz/sim/storage/iface/device.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def store(self, v: QuantifiedStock, *args, **kwargs):
    """
    存储
    """
    log = self.log
    v = unify_quantified_value(v, self.unit)
    cells = self.store_strategy.find_location(self.props, [v])
    with self.__occupied__.request() as req:
        # yield req
        if self.free_capacity < v.quantity:
            log.warning("%s < %s, skip store", self.free_capacity, v.quantity)
            yield self.env.timeout(0)
            return
        for cell in cells:
            cell.store(v)
        yield self.__container__.put(v)
        t = self.props.store_speed_time
        yield self.env.timeout(0 if not t else t.value)

OperationType

操作类型

源代码位于： logis/biz/sim/const/__init__.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class OperationType(Enum):
    """
    操作类型
    """

    Pick = "pick"
    Store = "store"

    def matches(self, operation: Union[str, "OperationType"]):
        """
        判断是否匹配操作类型，此方法以非严格模式简化判断、兼容历史逻辑

        Args:
            operation: 操作类型

        Returns:
            是否匹配
        """
        return self == operation or self.value == operation

Pick class-attribute instance-attribute 

Pick = 'pick'

Store class-attribute instance-attribute 

Store = 'store'

matches 

matches(operation: Union[str, OperationType])

判断是否匹配操作类型，此方法以非严格模式简化判断、兼容历史逻辑

参数：

名称 类型 描述 默认
operation Union[str, OperationType]

操作类型

 必需

返回：

类型 描述

是否匹配
源代码位于： logis/biz/sim/const/__init__.py 
72
73
74
75
76
77
78
79
80
81
82
def matches(self, operation: Union[str, "OperationType"]):
    """
    判断是否匹配操作类型，此方法以非严格模式简化判断、兼容历史逻辑

    Args:
        operation: 操作类型

    Returns:
        是否匹配
    """
    return self == operation or self.value == operation

Point

此类不通用，历史遗留，不建议使用，如有需要建议使用GenericPoint 支持x,y,z三维坐标点，也可当作二维坐标使用 TODO: 处理单位

源代码位于： logis/data_type/point.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
class Point(GenericPoint[float]):
    """
    此类不通用，历史遗留，不建议使用，如有需要建议使用GenericPoint
    支持x,y,z三维坐标点，也可当作二维坐标使用
    TODO: 处理单位
    """

    __counter = count(0, step=1)
    __global_precision__ = 3

    @classmethod
    def try_parse(cls, dc: dict, **kwargs) -> "Point":
        """
        尝试读取X、Y、Z坐标，返回一个Point对象
        """
        if not isinstance(dc, dict):
            raise ValueError("Expected a dictionary for Point parsing.")
        xyz = [dc.get(k, None) for k in ("X", "Y", "Z")]
        return Point(xyz, **kwargs)

    @classmethod
    def from_tuple(cls, args: Tuple[Number], **kwargs):
        """
        从元组创建点
        """
        return cls(args, **kwargs)

    @classmethod
    def of(
        cls,
        x: Optional[Number] = None,
        y: Optional[Number] = None,
        z: Optional[Number] = None,
        **kwargs,
    ) -> "Point":
        p = Point(x, y, z, **kwargs)
        return p

    def __init__(self, *args, precision: Optional[int] = None, **kwargs):
        """
        支持[x,y]、x,y、[x,y,z]、x,y,z传参赋值
        默认x=y=z=0

        TODO: 不应该自动转float、不应该自动保留两位小数
        """
        self._id = next(self.__counter)
        self.unit: Optional[str] = kwargs.get("unit", None)

        x: Optional[Number] = kwargs.get("x", None)
        y: Optional[Number] = kwargs.get("y", None)
        z: Optional[Number] = kwargs.get("z", None)
        if len(args) == 1:
            assert isinstance(
                args[0], (list, tuple)
            ), "single argument must be a list or tuple."
            x = cast_if_not_none(args[0][0], float)
            y = cast_if_not_none(args[0][1], float)
            if len(args[0]) > 2 and (z := args[0][2]) is not None:
                z = cast_if_not_none(z, float)

        elif len(args) >= 2:
            x = cast_if_not_none(args[0], float)
            y = cast_if_not_none(args[1], float)
            if len(args) > 2 and (z := args[2]) is not None:
                z = cast_if_not_none(z, float)

        # 调试过程中发现z有时候是None有时候是0，因此这里统一给了默认值
        # 实际上应该外层规范
        x = x if x is not None else 0
        y = y if y is not None else 0
        z = z if z is not None else 0

        kwargs.update(x=x, y=y, z=z)

        super().__init__(precision=precision, **kwargs)

    def __repr__(self):
        return f"Point({self.x}, {self.y}, {self.z})"

    def __hash__(self):

        def s(v: Optional[Number]):
            if not self.precision:
                return v
            if v is None:
                return v
            return v
            # return str(Decimal(v).quantize(Decimal(f"0.{'0'*self.precision}")))

        # if self._id is not None:
        #     return hash((self.__class__.__name__, self._id))
        return hash((s(self.x), s(self.y), s(self.z), self.precision, self.unit))

    def __lt__(self, other):
        """
        FIXME: 考虑z
        """
        if isinstance(other, Point):
            # TODO: 只要任何一个大于就行？
            return (self.x, self.y) < (other.x, other.y)
        return NotImplemented

    def __sub__(self, other: "Point"):
        def sub(a: Number, b: Number):
            if a is None and b is None:
                return None
            a = a or 0
            b = b or 0
            return a - b

        assert self.unit == other.unit, "unit must be same"
        return Point.of(
            x=sub(self.x, other.x),
            y=sub(self.y, other.y),
            z=sub(self.z, other.z),
            unit=self.unit,
        )

    def to_tuple(self) -> TuplePoint:
        return (
            self.x,
            self.y,
            self.z,
        )

    def __str__(self):
        return f"Point(id={self._id},x={self.x},y={self.y},z={self.z})"

__counter class-attribute instance-attribute 

__counter = count(0, step=1)

__global_precision__ class-attribute instance-attribute 

__global_precision__ = 3

unit instance-attribute 

unit: Optional[str] = get('unit', None)

__hash__ 

__hash__()
源代码位于： logis/data_type/point.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
def __hash__(self):

    def s(v: Optional[Number]):
        if not self.precision:
            return v
        if v is None:
            return v
        return v
        # return str(Decimal(v).quantize(Decimal(f"0.{'0'*self.precision}")))

    # if self._id is not None:
    #     return hash((self.__class__.__name__, self._id))
    return hash((s(self.x), s(self.y), s(self.z), self.precision, self.unit))

__init__ 

__init__(*args, precision: Optional[int] = None, **kwargs)

支持[x,y]、x,y、[x,y,z]、x,y,z传参赋值 默认x=y=z=0

TODO: 不应该自动转float、不应该自动保留两位小数

源代码位于： logis/data_type/point.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def __init__(self, *args, precision: Optional[int] = None, **kwargs):
    """
    支持[x,y]、x,y、[x,y,z]、x,y,z传参赋值
    默认x=y=z=0

    TODO: 不应该自动转float、不应该自动保留两位小数
    """
    self._id = next(self.__counter)
    self.unit: Optional[str] = kwargs.get("unit", None)

    x: Optional[Number] = kwargs.get("x", None)
    y: Optional[Number] = kwargs.get("y", None)
    z: Optional[Number] = kwargs.get("z", None)
    if len(args) == 1:
        assert isinstance(
            args[0], (list, tuple)
        ), "single argument must be a list or tuple."
        x = cast_if_not_none(args[0][0], float)
        y = cast_if_not_none(args[0][1], float)
        if len(args[0]) > 2 and (z := args[0][2]) is not None:
            z = cast_if_not_none(z, float)

    elif len(args) >= 2:
        x = cast_if_not_none(args[0], float)
        y = cast_if_not_none(args[1], float)
        if len(args) > 2 and (z := args[2]) is not None:
            z = cast_if_not_none(z, float)

    # 调试过程中发现z有时候是None有时候是0，因此这里统一给了默认值
    # 实际上应该外层规范
    x = x if x is not None else 0
    y = y if y is not None else 0
    z = z if z is not None else 0

    kwargs.update(x=x, y=y, z=z)

    super().__init__(precision=precision, **kwargs)

__lt__ 

__lt__(other)

FIXME: 考虑z

源代码位于： logis/data_type/point.py 
251
252
253
254
255
256
257
258
def __lt__(self, other):
    """
    FIXME: 考虑z
    """
    if isinstance(other, Point):
        # TODO: 只要任何一个大于就行？
        return (self.x, self.y) < (other.x, other.y)
    return NotImplemented

__repr__ 

__repr__()
源代码位于： logis/data_type/point.py 
234
235
def __repr__(self):
    return f"Point({self.x}, {self.y}, {self.z})"

__str__ 

__str__()
源代码位于： logis/data_type/point.py 
283
284
def __str__(self):
    return f"Point(id={self._id},x={self.x},y={self.y},z={self.z})"

__sub__ 

__sub__(other: Point)
源代码位于： logis/data_type/point.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def __sub__(self, other: "Point"):
    def sub(a: Number, b: Number):
        if a is None and b is None:
            return None
        a = a or 0
        b = b or 0
        return a - b

    assert self.unit == other.unit, "unit must be same"
    return Point.of(
        x=sub(self.x, other.x),
        y=sub(self.y, other.y),
        z=sub(self.z, other.z),
        unit=self.unit,
    )

from_tuple classmethod 

from_tuple(args: Tuple[Number], **kwargs)

从元组创建点

源代码位于： logis/data_type/point.py 
178
179
180
181
182
183
@classmethod
def from_tuple(cls, args: Tuple[Number], **kwargs):
    """
    从元组创建点
    """
    return cls(args, **kwargs)

of classmethod 

of(x: Optional[Number] = None, y: Optional[Number] = None, z: Optional[Number] = None, **kwargs) -> Point
源代码位于： logis/data_type/point.py 
185
186
187
188
189
190
191
192
193
194
@classmethod
def of(
    cls,
    x: Optional[Number] = None,
    y: Optional[Number] = None,
    z: Optional[Number] = None,
    **kwargs,
) -> "Point":
    p = Point(x, y, z, **kwargs)
    return p

to_tuple 

to_tuple() -> TuplePoint
源代码位于： logis/data_type/point.py 
276
277
278
279
280
281
def to_tuple(self) -> TuplePoint:
    return (
        self.x,
        self.y,
        self.z,
    )

try_parse classmethod 

try_parse(dc: dict, **kwargs) -> Point

尝试读取X、Y、Z坐标，返回一个Point对象

源代码位于： logis/data_type/point.py 
168
169
170
171
172
173
174
175
176
@classmethod
def try_parse(cls, dc: dict, **kwargs) -> "Point":
    """
    尝试读取X、Y、Z坐标，返回一个Point对象
    """
    if not isinstance(dc, dict):
        raise ValueError("Expected a dictionary for Point parsing.")
    xyz = [dc.get(k, None) for k in ("X", "Y", "Z")]
    return Point(xyz, **kwargs)

QuantifiedContainer

初衷是统一simpy的Resource、Container、Store等类型并增加单位的概念，但目前还未完全实现

引入单位的概念，能够简化对simpy的操作，可实现单位换算、检验等功能

源代码位于： logis/biz/sim/storage/iface/container.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class QuantifiedContainer:
    """
    初衷是统一simpy的Resource、Container、Store等类型并增加单位的概念，但目前还未完全实现

    引入单位的概念，能够简化对simpy的操作，可实现单位换算、检验等功能

    """

    __container__: Optional[simpy.Container] = None

    def __init__(
        self, capacity: QuantifiedValue, env: simpy.Environment, *args, **kwargs
    ):
        self.__quantified_value__ = capacity
        self.__container__ = simpy.Container(env, capacity.quantity)

    @property
    def unit(self) -> Unit:
        return self.__quantified_value__.unit

    @property
    def capacity(self):
        return self.__quantified_value__.quantity

    @property
    def level(self):
        return self.__container__.level

    @property
    def free_capacity(self) -> Number:
        """
        空闲空间
        """
        return self.capacity - self.level

    def get(self, v: QuantifiedValue):
        assert v.unit == self.unit, "unit must be the same"
        return self.__container__.get(v.quantity)

    def put(self, v: QuantifiedValue):
        assert v.unit == self.unit, "unit must be the same"
        return self.__container__.put(v.quantity)

__container__ class-attribute instance-attribute 

__container__: Optional[Container] = Container(env, quantity)

__quantified_value__ instance-attribute 

__quantified_value__ = capacity

capacity property 

capacity

free_capacity property 

free_capacity: Number

空闲空间

level property 

level

unit property 

unit: Unit

__init__ 

__init__(capacity: QuantifiedValue, env: Environment, *args, **kwargs)
源代码位于： logis/biz/sim/storage/iface/container.py 
19
20
21
22
23
def __init__(
    self, capacity: QuantifiedValue, env: simpy.Environment, *args, **kwargs
):
    self.__quantified_value__ = capacity
    self.__container__ = simpy.Container(env, capacity.quantity)

get 

get(v: QuantifiedValue)
源代码位于： logis/biz/sim/storage/iface/container.py 
44
45
46
def get(self, v: QuantifiedValue):
    assert v.unit == self.unit, "unit must be the same"
    return self.__container__.get(v.quantity)

put 

put(v: QuantifiedValue)
源代码位于： logis/biz/sim/storage/iface/container.py 
48
49
50
def put(self, v: QuantifiedValue):
    assert v.unit == self.unit, "unit must be the same"
    return self.__container__.put(v.quantity)

QuantifiedStock

基础货物属性

源代码位于： logis/biz/sim/stock/model/__init__.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class QuantifiedStock(QuantifiedValue):
    """
    基础货物属性
    """

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

    sku: Optional[str] = None
    name: str
    code: Optional[str] = None

    # 默认数量为1
    quantity: Number = 1

    order_id: Optional[str] = None
    task_id: Optional[str] = None

    @property
    def unique_id(self):
        return self.sku or self.code or self.name

    @property
    def storage_key(self) -> Tuple[str, str]:
        """
        用于存储索引的key，格式为(name, unit)，
        """
        return (self.name, self.unit)

code class-attribute instance-attribute 

code: Optional[str] = None

model_config class-attribute instance-attribute 

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

name instance-attribute 

name: str

order_id class-attribute instance-attribute 

order_id: Optional[str] = None

quantity class-attribute instance-attribute 

quantity: Number = 1

sku class-attribute instance-attribute 

sku: Optional[str] = None

storage_key property 

storage_key: Tuple[str, str]

用于存储索引的key，格式为(name, unit)，

task_id class-attribute instance-attribute 

task_id: Optional[str] = None

unique_id property 

unique_id

QuantifiedValue

量化值，相比于接口NumberUnit，适合作为数据结构使用

源代码位于： logis/data_type/unitable.py 
138
139
140
141
142
143
144
145
146
147
class QuantifiedValue(BaseModel, NumberUnit):
    """
    量化值，相比于接口NumberUnit，适合作为数据结构使用
    """
    model_config = MODEL_CONFIG

    name: Optional[str] = None

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

model_config class-attribute instance-attribute 

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

name class-attribute instance-attribute 

name: Optional[str] = None

__init__ 

__init__(**kwargs)
源代码位于： logis/data_type/unitable.py 
146
147
def __init__(self, **kwargs):
    super().__init__(**kwargs)

RackGroupProperties

货架组

源代码位于： logis/biz/sim/storage/model/__init__.py 
197
198
199
200
201
202
203
204
205
class RackGroupProperties(ContainerMetadata):
    """
    货架组
    """

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

    rack_ids: List[str] = Field(default_factory=list)
    racks: List[RackProperties] = Field(default_factory=list)

model_config class-attribute instance-attribute 

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

rack_ids class-attribute instance-attribute 

rack_ids: List[str] = Field(default_factory=list)

racks class-attribute instance-attribute 

racks: List[RackProperties] = Field(default_factory=list)

RackProperties

货架的属性信息

源代码位于： logis/biz/sim/storage/model/__init__.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
class RackProperties(ContainerMetadata):
    """
    货架的属性信息
    """

    rack_group_id: Optional[str] = None

    type: Optional[RackType] = Field(
        None, validation_alias=AliasChoices("类型", "type")
    )
    # TODO: 考虑去除
    is_obstacle: Optional[bool] = Field(
        validation_alias=AliasChoices("是否障碍", "is_obstacle", "是障碍"), default=None
    )

    row_count: int = Field(
        validation_alias=AliasChoices("层数", "row_count"), default=1
    )
    col_count: Optional[int] = Field(
        validation_alias=AliasChoices("单元格数", "列数", "col_count"), default=None
    )

    cell_metadata: Optional[ContainerMetadata] = None

    cell_ids: List[str] = Field(default_factory=list)
    cells: List[CellProperties] = Field(default_factory=list)

cell_ids class-attribute instance-attribute 

cell_ids: List[str] = Field(default_factory=list)

cell_metadata class-attribute instance-attribute 

cell_metadata: Optional[ContainerMetadata] = None

cells class-attribute instance-attribute 

cells: List[CellProperties] = Field(default_factory=list)

col_count class-attribute instance-attribute 

col_count: Optional[int] = Field(validation_alias=AliasChoices('单元格数', '列数', 'col_count'), default=None)

is_obstacle class-attribute instance-attribute 

is_obstacle: Optional[bool] = Field(validation_alias=AliasChoices('是否障碍', 'is_obstacle', '是障碍'), default=None)

rack_group_id class-attribute instance-attribute 

rack_group_id: Optional[str] = None

row_count class-attribute instance-attribute 

row_count: int = Field(validation_alias=AliasChoices('层数', 'row_count'), default=1)

type class-attribute instance-attribute 

type: Optional[RackType] = Field(None, validation_alias=AliasChoices('类型', 'type'))

RetrieveStrategy

取回策略

源代码位于： logis/biz/sim/storage/iface/retrieve.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
class RetrieveStrategy(StorageStrategy):
    """
    取回策略
    """

    def find_location(
        self, storage: StorageProperties, stocks: List[QuantifiedStock], *args, **kwargs
    ) -> List[CellProperties]:
        result = []
        type_cells_map = storage.group_cell_by_stock_type()
        for stock in stocks:
            # 过滤出可以容纳此货物的所有储位
            candidates = type_cells_map.get(stock.unique_id, []) + type_cells_map.get(
                "*", []
            )
            for candidate in candidates:
                if candidate.can_retrieve(stock):
                    result.append(candidate)
        return result

find_location 

find_location(storage: StorageProperties, stocks: List[QuantifiedStock], *args, **kwargs) -> List[CellProperties]
源代码位于： logis/biz/sim/storage/iface/retrieve.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def find_location(
    self, storage: StorageProperties, stocks: List[QuantifiedStock], *args, **kwargs
) -> List[CellProperties]:
    result = []
    type_cells_map = storage.group_cell_by_stock_type()
    for stock in stocks:
        # 过滤出可以容纳此货物的所有储位
        candidates = type_cells_map.get(stock.unique_id, []) + type_cells_map.get(
            "*", []
        )
        for candidate in candidates:
            if candidate.can_retrieve(stock):
                result.append(candidate)
    return result

SpatialProps

空间属性

源代码位于： logis/data_type/spatial.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class SpatialProps(BaseModel):
    """
    空间属性
    """

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

    width: Optional[Length] = Field(
        default=None,
        validation_alias=AliasChoices("单元格宽度", "width"),
    )
    height: Optional[Length] = Field(
        default=None, validation_alias=AliasChoices("层高", "height")
    )
    depth: Optional[Length] = Field(
        validation_alias=AliasChoices("货架深度", "托盘货架深度"), default=None
    )
    rotation: Optional[QuantifiedValue] = Field(
        validation_alias=AliasChoices("旋转", "rotate"), default=None
    )
    center_point: Optional[Point] = None

    @field_validator("center_point", mode="before")
    def center_point_validator_before(cls, v: Optional[Point]):
        if v is None:
            return None
        if isinstance(v, Point):
            return v
        if isinstance(v, dict):
            return Point.model_validate(v)
        elif isinstance(v, (tuple, list)):
            return Point.from_tuple(v)
        raise ValueError("only dict, tuple, list type is supported")

center_point class-attribute instance-attribute 

center_point: Optional[Point] = None

depth class-attribute instance-attribute 

depth: Optional[Length] = Field(validation_alias=AliasChoices('货架深度', '托盘货架深度'), default=None)

height class-attribute instance-attribute 

height: Optional[Length] = Field(default=None, validation_alias=AliasChoices('层高', 'height'))

model_config class-attribute instance-attribute 

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

rotation class-attribute instance-attribute 

rotation: Optional[QuantifiedValue] = Field(validation_alias=AliasChoices('旋转', 'rotate'), default=None)

width class-attribute instance-attribute 

width: Optional[Length] = Field(default=None, validation_alias=AliasChoices('单元格宽度', 'width'))

center_point_validator_before 

center_point_validator_before(v: Optional[Point])
源代码位于： logis/data_type/spatial.py 
32
33
34
35
36
37
38
39
40
41
42
@field_validator("center_point", mode="before")
def center_point_validator_before(cls, v: Optional[Point]):
    if v is None:
        return None
    if isinstance(v, Point):
        return v
    if isinstance(v, dict):
        return Point.model_validate(v)
    elif isinstance(v, (tuple, list)):
        return Point.from_tuple(v)
    raise ValueError("only dict, tuple, list type is supported")

Storable

可存储的

使用鸭子类型实现，相比ABC更灵活

源代码位于： logis/iface/container.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@runtime_checkable
class Storable(Protocol):
    # class Storable(metaclass=ABCMeta):
    """
    可存储的

    使用鸭子类型实现，相比ABC更灵活
    """

    def pre_store(self, *args, **kwargs):
        """
        预存储，一般用来实现资源检查、校验
        """
        pass

    def pre_retrieve(self, *args, **kwargs):
        """
        预取回，一般用来实现资源检查、校验
        """
        pass

    def store(self, *args, **kwargs) -> StoreResult:
        """
        真正的存储操作
        """
        pass

    def retrieve(self, *args, **kwargs) -> RetrieveResult:
        """
        真正的取回操作
        """
        pass

pre_retrieve 

pre_retrieve(*args, **kwargs)

预取回，一般用来实现资源检查、校验

源代码位于： logis/iface/container.py 
45
46
47
48
49
def pre_retrieve(self, *args, **kwargs):
    """
    预取回，一般用来实现资源检查、校验
    """
    pass

pre_store 

pre_store(*args, **kwargs)

预存储，一般用来实现资源检查、校验

源代码位于： logis/iface/container.py 
39
40
41
42
43
def pre_store(self, *args, **kwargs):
    """
    预存储，一般用来实现资源检查、校验
    """
    pass

retrieve 

retrieve(*args, **kwargs) -> RetrieveResult

真正的取回操作

源代码位于： logis/iface/container.py 
57
58
59
60
61
def retrieve(self, *args, **kwargs) -> RetrieveResult:
    """
    真正的取回操作
    """
    pass

store 

store(*args, **kwargs) -> StoreResult

真正的存储操作

源代码位于： logis/iface/container.py 
51
52
53
54
55
def store(self, *args, **kwargs) -> StoreResult:
    """
    真正的存储操作
    """
    pass

StorageSelectionStrategy

存储选择策略枚举，此枚举试图统一并兼容历史逻辑

源代码位于： logis/biz/sim/const/__init__.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class StorageSelectionStrategy(Enum):
    """
    存储选择策略枚举，此枚举试图统一并兼容历史逻辑
    """

    BusyLevelAscend = "按作业数量少的优先"
    DistanceAscend = "按距离近的优先"
    NumberAscend = "按编号从小到大"
    NumberDescend = "按编号从大到小"
    Custom = "自定义"

    def matches(
        self,
        strategy: Union[
            "StorageSelectionStrategy", LegencyStorageSelectionStrategyName
        ],
    ):
        # 下面有一些历史兼容逻辑
        if strategy == "按储位编号从大到小":
            return self == StorageSelectionStrategy.NumberDescend
        if strategy == "按储位编号从小到大":
            return self == StorageSelectionStrategy.NumberAscend
        if strategy == "按作业数量少的货架优先":
            return self == StorageSelectionStrategy.BusyLevelAscend
        if strategy == "按距离近的货架优先":
            return self == StorageSelectionStrategy.DistanceAscend
        return self == strategy or self.value == strategy

BusyLevelAscend class-attribute instance-attribute 

BusyLevelAscend = '按作业数量少的优先'

Custom class-attribute instance-attribute 

Custom = '自定义'

DistanceAscend class-attribute instance-attribute 

DistanceAscend = '按距离近的优先'

NumberAscend class-attribute instance-attribute 

NumberAscend = '按编号从小到大'

NumberDescend class-attribute instance-attribute 

NumberDescend = '按编号从大到小'

matches 

matches(strategy: Union[StorageSelectionStrategy, LegencyStorageSelectionStrategyName])
源代码位于： logis/biz/sim/const/__init__.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def matches(
    self,
    strategy: Union[
        "StorageSelectionStrategy", LegencyStorageSelectionStrategyName
    ],
):
    # 下面有一些历史兼容逻辑
    if strategy == "按储位编号从大到小":
        return self == StorageSelectionStrategy.NumberDescend
    if strategy == "按储位编号从小到大":
        return self == StorageSelectionStrategy.NumberAscend
    if strategy == "按作业数量少的货架优先":
        return self == StorageSelectionStrategy.BusyLevelAscend
    if strategy == "按距离近的货架优先":
        return self == StorageSelectionStrategy.DistanceAscend
    return self == strategy or self.value == strategy

StorageStrategy

存取策略

源代码位于： logis/biz/sim/storage/iface/base.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
@runtime_checkable
class StorageStrategy(Protocol):
    """
    存取策略
    """

    def find_location(
        self, storage: StorageProperties, stocks: List[QuantifiedStock], *args, **kwargs
    ) -> List[CellProperties]:
        """
        寻找储位位置，存储获取或者取出货物
        """
        raise NotImplementedError()

find_location 

find_location(storage: StorageProperties, stocks: List[QuantifiedStock], *args, **kwargs) -> List[CellProperties]

寻找储位位置，存储获取或者取出货物

源代码位于： logis/biz/sim/storage/iface/base.py 
14
15
16
17
18
19
20
def find_location(
    self, storage: StorageProperties, stocks: List[QuantifiedStock], *args, **kwargs
) -> List[CellProperties]:
    """
    寻找储位位置，存储获取或者取出货物
    """
    raise NotImplementedError()

StoreStrategy

存储策略

源代码位于： logis/biz/sim/storage/iface/store.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class StoreStrategy(StorageStrategy):
    """
    存储策略
    """

    def find_location(
        self, storage: StorageProperties, stocks: List[QuantifiedStock], *args, **kwargs
    ) -> List[CellProperties]:
        result = []
        type_cells_map = storage.group_cell_by_stock_type()
        for stock in stocks:
            # 过滤出可以容纳此货物的所有储位
            candidates = type_cells_map.get(stock.unique_id, []) + type_cells_map.get(
                "*", []
            )
            for candidate in candidates:
                if candidate.can_store(stock):
                    result.append(candidate)
                    break
        return result

find_location 

find_location(storage: StorageProperties, stocks: List[QuantifiedStock], *args, **kwargs) -> List[CellProperties]
源代码位于： logis/biz/sim/storage/iface/store.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def find_location(
    self, storage: StorageProperties, stocks: List[QuantifiedStock], *args, **kwargs
) -> List[CellProperties]:
    result = []
    type_cells_map = storage.group_cell_by_stock_type()
    for stock in stocks:
        # 过滤出可以容纳此货物的所有储位
        candidates = type_cells_map.get(stock.unique_id, []) + type_cells_map.get(
            "*", []
        )
        for candidate in candidates:
            if candidate.can_store(stock):
                result.append(candidate)
                break
    return result

Time

时间

源代码位于： logis/data_type/unitable.py 
191
192
193
194
195
196
class Time(QuantifiedValue):
    """
    时间
    """

    pass

euclid_distance 

euclid_distance(pos1: Point, pos2: Point, precision: Optional[int] = None) -> float

计算欧几里得距离

源代码位于： logis/math/__init__.py 
19
20
21
22
def euclid_distance(pos1: Point, pos2: Point, precision: Optional[int] = None) -> float:
    """计算欧几里得距离"""
    v = math.sqrt((pos1.x - pos2.x) ** 2 + (pos1.y - pos2.y) ** 2)
    return round(v, precision) if precision is not None else v

unify_quantified_value 

unify_quantified_value(self: QuantifiedValue, target_unit: Optional[Unit] = None, unit_config: UnitConfig = DEFAULT_UNIT_CONFIG, modify_ref: bool = False)

单位转换

源代码位于： logis/data_type/unitable.py 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
def unify_quantified_value(
    self: QuantifiedValue,
    target_unit: Optional[Unit] = None,
    unit_config: UnitConfig = DEFAULT_UNIT_CONFIG,
    modify_ref: bool = False,
):
    """
    单位转换
    """
    same_unit = self.unit == target_unit
    if unit_config is None:
        assert same_unit, "unit must be the same if no radio_computer given"
    else:
        ratio = 1 if same_unit else unit_config.get_ratio(self.unit, target_unit)
        if modify_ref:
            self.quantity = self.quantity * ratio
            self.unit = target_unit
            return self
        # 上面这种方式能尽可能避免属性丢失，但修改的是原始对象
        dc = {
            **self.model_dump(),
            **dict(quantity=self.quantity * ratio, unit=target_unit),
        }
        return type(self)(**dc)