File size: 2,440 Bytes
e9315b2 6e3f176 e9315b2 6e3f176 e9315b2 6e3f176 e9315b2 6e3f176 e9315b2 |
1 2 3 4 5 6 7 8 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 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 |
"""
Data models for the JSSP Environment.
"""
from dataclasses import dataclass
from typing import Optional
from openenv_core import Action, Observation
JobT = list[tuple[int, int]] # (machine_index, processing_time)
@dataclass(kw_only=True)
class JSSPAction(Action):
"""Action for the JSSP environment."""
job_ids: list[int]
def __post_init__(self):
if isinstance(self.job_ids, str):
# For web app
self.job_ids = parse_job_ids(self.job_ids)
@dataclass(kw_only=True)
class MachineObservation:
"""Observation of a single machine in the JSSP environment."""
machine_id: int
busy_until: Optional[int]
current_job_id: Optional[int]
@dataclass
class JobObservation:
"""Observation of a given Job in the JSSP environment."""
job_id: int
operations: JobT # remaining operations to be scheduled
busy_until: Optional[int] # time until the current operation is complete (or none if available)
@dataclass(kw_only=True)
class JSSPObservation(Observation):
"""Observation from the JSSP environment - the echoed message."""
episode_id: str
step_count: int
machines: list[MachineObservation]
jobs: list[JobObservation]
def available_machines(self) -> list[MachineObservation]:
"""Get available machines from observation."""
return [m for m in self.machines if m.busy_until is None or m.busy_until <= self.step_count]
def available_jobs(self) -> list[JobObservation]:
"""Get available jobs from observation."""
available_machine_ids = [m.machine_id for m in self.available_machines()]
return [
job
for job in self.jobs
if (job.busy_until is None or job.busy_until <= self.step_count)
and len(job.operations) > 0
and job.operations[0][0] in available_machine_ids
]
def parse_job_ids(job_ids: str) -> list[int]:
"""Parse job_ids from string (error out if cannot be parsed)."""
try:
return [int(job_id) for job_id in job_ids.split(",") if job_id.strip()]
except ValueError as e:
raise ValueError(f"Invalid job_ids: {job_ids}") from e
@dataclass
class ScheduledEvent:
"""Represents a scheduled operation on a machine.
Used for plotting the schedule.
Not used for the environment / policy / solver.
"""
job_id: int
machine_id: int
start_time: int
end_time: int
|