面向对象编程：通过方法输出更新类属性及设备任务分配问题

本人是面向对象编程（OOP）新手，构建了多个相互关联的类，尝试通过方法输出更新类属性，但无法确定语法是否正确，未得到预期结果。

类定义代码 #

from dataclasses import dataclass
from typing import List, Any

# 假设UNASSIGNED_FLAG为预定义常量，例如-1
UNASSIGNED_FLAG = -1

@dataclass
class ProductionTask:
    id: int
    processingtime: int
    duration: float
    done: bool
    start_time: int = UNASSIGNED_FLAG
    end_time: int = UNASSIGNED_FLAG
    production_equipment: Any = UNASSIGNED_FLAG
    stage: Any = UNASSIGNED_FLAG


@dataclass
class TransferTask:
    id: int
    min_transfer_duration: float
    max_transfer_duration: float
    done: bool
    start_time: int = UNASSIGNED_FLAG
    end_time: int = UNASSIGNED_FLAG
    equipment_item: Any = UNASSIGNED_FLAG
    stage: Any = UNASSIGNED_FLAG

@dataclass
class ProductionEquipment:
    id: int
    production_tasks: List[ProductionTask] | None = None
    stage: Any = UNASSIGNED_FLAG


@dataclass
class TransferEquipment:
    id: int
    transfer_tasks: List[TransferTask] | None = None
    stage: Any = UNASSIGNED_FLAG

@dataclass
class Stage:
    id: int
    production_equipment: List[ProductionEquipment] | None = None
    transfer_equipment: List[TransferEquipment] | None = None

Stage类中的现有方法 #

def production_equipment_available(self, time):
    if self.id > 1:
        if self.production_equipment is not None:
            for equipment in self.production_equipment:
                if equipment.production_tasks is not None:
                    if all(
                        (
                            task.end_time < time
                            for task in equipment.production_tasks
                        )
                    ):
                        available = True
                    else:
                        available = False
                    return available


def transfer_task_passed_minimum_from_unassigned_production_tasks(self):
    production_task_ids = self.unassigned_production_tasks_in_stage2and3()
    transfer_task_ids = self.completed_transfer_tasks(self._time)
    for id in production_task_ids:
        if id in transfer_task_ids:
            return id
    return None

期望实现需求 #

1. 修改production_equipment_available方法，使其返回可用设备ID而非仅布尔值；假设存在可用设备，且需要分配的任务ID为2。
2. 将任务ID为2的任务分配给可用设备，同时设置该任务的start_time为分配时间。

一、现有代码的逻辑缺陷 #

1. production_equipment_available方法的遍历问题：

   • 循环仅检查第一个设备就直接返回结果，不会遍历后续设备，导致漏掉其他可能可用的设备。
   • 若设备的production_tasks为None，会直接跳过该设备，未将其判定为可用（无任务的设备应视为可用）。
   • 返回值为布尔值，不符合返回设备ID的需求。

2. 空列表默认值的处理问题：
   数据类中列表类型属性默认值设为None，导致后续需要频繁判断is not None，增加代码冗余。

二、需求实现步骤 #

1. 修改production_equipment_available方法，返回可用设备ID #

修改后方法会遍历所有设备，返回第一个可用设备的ID（如需返回所有可用设备ID，可改为返回列表），同时修复空任务列表的判定逻辑：

def production_equipment_available(self, time):
    if self.id > 1 and self.production_equipment is not None:
        for equipment in self.production_equipment:
            # 设备无任务列表或所有任务已结束，判定为可用
            if equipment.production_tasks is None or all(task.end_time < time for task in equipment.production_tasks):
                return equipment.id
        # 无可用设备时返回None
        return None
    # 阶段ID不满足条件或无生产设备时返回None
    return None

若需要返回所有可用设备的ID列表，可改为：

def production_equipment_available(self, time):
    available_ids = []
    if self.id > 1 and self.production_equipment is not None:
        for equipment in self.production_equipment:
            if equipment.production_tasks is None or all(task.end_time < time for task in equipment.production_tasks):
                available_ids.append(equipment.id)
    return available_ids

2. 实现任务分配逻辑 #

新增方法完成任务分配，或在业务逻辑中调用上述方法获取可用设备后执行分配：

def assign_production_task(self, task_id, assign_time):
    # 获取可用设备ID
    available_equip_id = self.production_equipment_available(assign_time)
    if available_equip_id is None:
        return False  # 无可用设备，分配失败
    
    # 找到目标任务（假设Stage类有存储ProductionTask的属性production_tasks）
    target_task = next((task for task in self.production_tasks if task.id == task_id), None)
    if target_task is None:
        return False  # 未找到目标任务
    
    # 找到对应可用设备
    target_equip = next((equip for equip in self.production_equipment if equip.id == available_equip_id), None)
    if target_equip is None:
        return False
    
    # 初始化设备的任务列表（若为None）
    if target_equip.production_tasks is None:
        target_equip.production_tasks = []
    
    # 完成分配：设置任务的开始时间与关联设备
    target_task.start_time = assign_time
    target_task.production_equipment = target_equip
    # 将任务添加到设备的任务列表
    target_equip.production_tasks.append(target_task)
    
    return True  # 分配成功

3. 数据类初始化优化（可选） #

为避免频繁判断None，将数据类中列表类型的默认值改为空列表（需导入field）：

from dataclasses import dataclass, field

@dataclass
class ProductionEquipment:
    id: int
    production_tasks: List[ProductionTask] = field(default_factory=list)
    stage: Any = UNASSIGNED_FLAG


@dataclass
class TransferEquipment:
    id: int
    transfer_tasks: List[TransferTask] = field(default_factory=list)
    stage: Any = UNASSIGNED_FLAG

@dataclass
class Stage:
    id: int
    production_equipment: List[ProductionEquipment] = field(default_factory=list)
    transfer_equipment: List[TransferEquipment] = field(default_factory=list)
    # 新增存储任务的属性
    production_tasks: List[ProductionTask] = field(default_factory=list)

三、调用示例 #

# 初始化测试数据
stage = Stage(id=2)
equip1 = ProductionEquipment(id=1)
equip2 = ProductionEquipment(id=2)
stage.production_equipment = [equip1, equip2]

task = ProductionTask(id=2, processingtime=5, duration=2.0, done=False)
stage.production_tasks.append(task)

# 执行分配
assign_success = stage.assign_production_task(task_id=2, assign_time=10)
if assign_success:
    print(f"任务2分配成功，开始时间：{task.start_time}，分配设备ID：{task.production_equipment.id}")
else:
    print("任务分配失败")

内容的提问来源于stack exchange，提问作者Margi Shah