Reference
- class DatetimeUnit(start: datetime | time = datetime.time(0, 0), unit: timedelta = datetime.timedelta(seconds=60))
The time unit of the
Model
class.- Parameters:
start – The start time of the scheduling problem. Defaults to time().
unit – The time unit for the scheduling problem. Defaults to timedelta(minutes=1).
Examples
>>> from amplify_sched import * >>> dtime = DatetimeUnit(start=datetime(year=2023, month=1, day=1), unit=timedelta(seconds=60)) >>> dtime {'start': datetime.datetime(2023, 1, 1, 0, 0), 'unit': datetime.timedelta(seconds=60)}
- from_int_timedelta(int_timedelta: int) timedelta
convert
int
totimedelta
using the specified time unit- Parameters:
int_timedelta –
int
to convert totimedelta
- Returns:
A
timedelta
value representing int_timedelta converted using the specified time unit.
- to_int_datetime(dtime: datetime | time) int
Convert
datetime
toint
using the time unit.to_int_datetime converts dtime to an integer value representing the time duration between dtime and the specified start time, in the specified time unit.
- Parameters:
dtime –
datetime
object- Returns:
the time duration
- class DependentJob(parent: DependentJobs, job: Job | str, interval_times: tuple[int | None, int | None] | int | None = None)
DependentJob contains the properties of dependency
- Parameters:
parent – parent object which this object belong to
job – dependent job to append
interval_time – interval time or times (lower and upper bounds)
- class DependentJobs(parent: Job)
DependentJobs is container of dependent job.
DependentJobs is managed as list of Job.
- Parameters:
parent – parent object which this object belong to
- class Job(name: str = '')
A class managing a Job.
In a scheduling problem, a Job is a collection of multiple tasks. A job is completed by processing each task in turn.
- Parameters:
name – job name. Defaults to “”
- add_dependent_jobs(job: Job | str, interval_time: tuple[int | None, int | None] | int | None = None) None
add dependent jobs
- Parameters:
job – depandent job
interval_time – interval time
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs.add("Job B") >>> model.jobs.add("Job C") >>> model.jobs["Job A"].add_dependent_jobs("Job B") >>> model.jobs["Job A"].add_dependent_jobs("Job C", (5, 10)) >>> model.jobs["Job A"].dependent_jobs [{'job_id': 1, 'interval_time_lower_bound': None, 'interval_time_upper_bound': None}, {'job_id': 2, 'interval_time_lower_bound': 5, 'interval_time_upper_bound': 10}]
- append(task: Task | None = None) None
append task
- Parameters:
task – task for appending. Defaults to
None
.- Raises:
ValueError – If task is already registerd
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][0] {'index': 0, 'processing_times': {}, 'required_resources': None, 'required_buffer_size': None, 'transportation_times': None, 'release_time': None, 'deadline': None}
- property dependent_jobs: DependentJobs
dependent jobs for this job
Dependent jobs is managed as list. During registration, the existence of the job in the model is checked. Registering a machine name that does not exist in the model will result in a ValueError.
- Parameters:
job – depandent job
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs.add("Job B") >>> model.jobs["Job A"].dependent_jobs.append("Job B") >>> model.jobs["Job A"].dependent_jobs [{'job_id': 1, 'interval_time_lower_bound': None, 'interval_time_upper_bound': None}]
- property index: int
job index
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].index 0 >>> model.jobs.add("Job B") >>> model.jobs["Job B"].index 1
- property name: str
job name
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].name 'Job A'
- property no_wait: bool | None
no wait flag
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].no_wait = True >>> model.jobs["Job A"].no_wait True
- property num_process: int
number of processes
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].num_process 0 >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"].num_process 1
- task(process: int) Task
returns a task for the given process
- Parameters:
process – task index
- Returns:
a task for the given process
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"].task(0) {'index': 0, 'processing_times': {}, 'required_resources': None, 'required_buffer_size': None, 'transportation_times': None, 'release_time': None, 'deadline': None}
- class Jobs(parent: Model)
Jobs is container of
Job
.Jobs is managed as dict. Key is job name and value is
Job
.- Parameters:
parent – parent object which this object belong to
- add(job: str | Job) None
add job
- Parameters:
job – job to adding
- Raises:
ValueError – If job has no name
- items()
- keys()
- values()
- class Machine(name: str = '')
A class managing a machine
In a scheduling problem, a machine processes tasks in a specified time range.
- Parameters:
name – machine name. Defaults to “”
- add_maintenance_time(start: datetime | time | int, end: datetime | time | int) None
add maintenance time
- Parameters:
start – maintenance start time
end – maintenance end time
Examples
>>> model = Model() >>> model.machines.add("Machine A") >>> model.machines["Machine A"].add_maintenance_time(10, 15) >>> model.machines["Machine A"].maintenance_times [(10, 15)]
- add_setup_time(setup_time: int | timedelta, prev_task: Job | Task | str, next_task: Job | Task | str) None
add setup time
Add setup_time (time taken to switch from previous task to next task).
- Parameters:
setup_time – setup time
prev_task – previous task
next_task – next task
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs.add("Job B") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job B"].append(Task()) >>> model.machines.add("Machine A") >>> model.machines["Machine A"].add_setup_time(10, "Job A", "Job B") >>> model.machines["Machine A"].setup_times [(10, ['Job A'], ['Job B'])]
- property buffer_size: int | None
machine buffer size
Examples
>>> model = Model() >>> model.machines.add("Machine A") >>> model.machines["Machine A"].buffer_size = 10 >>> model.machines["Machine A"].buffer_size 10
- property index: int
machine index
Examples
>>> model = Model() >>> model.machines.add("Machine A") >>> model.machines["Machine A"].index 0 >>> model.machines.add("Machine B") >>> model.machines["Machine B"].index 1
- property maintenance_times: List[TimeRange]
machine maintenance times
Examples
>>> model = Model() >>> model.machines.add("Machine A") >>> model.machines["Machine A"].maintenance_times.append((10, 15)) >>> model.machines["Machine A"].maintenance_times [(10, 15)]
- property name: str
machine name
Examples
>>> model = Model() >>> model.machines.add("Machine A") >>> model.machines["Machine A"].name 'Machine A'
- property setup_times: SetupTimes
machine setup times
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs.add("Job B") >>> model.machines.add("Machine A") >>> model.machines["Machine A"].setup_times.append((10, "Job A", "Job B")) >>> model.machines["Machine A"].setup_times [(10, ['Job A'], ['Job B'])]
- class Machines(parent: Model)
Machines is container of
Machine
.Machines is managed as dict. Key is machine name and value is
Machine
.- Parameters:
parent – parent object which this object belong to
- add(machine: str | Machine) None
add machine as dict key is machine.name and value is machine itself
- Parameters:
machine – machine to adding. If machine type is str, create new
Machine
instance using machine as name.- Raises:
ValueError – If machine is no name
- items()
- keys()
- values()
- class Model
A class managing model.
A model defines a scheduling problem in terms of jobs, machines and resources. Use the solve method to solve the defined scheduling problem using the Amplify Scheduling Engine.
Examples:
from amplify_sched import * # Define Model model = Model() # Define Jobs model.jobs.add("Job A") model.jobs.add("Job B") model.jobs.add("Job C") model.jobs.add("Job D") # Define Machines model.machines.add("Machine X") model.machines.add("Machine Y") model.machines.add("Machine Z") # Add Task to Job model.jobs["Job A"].append(Task()) model.jobs["Job A"][0].processing_times["Machine X"] = 10 model.jobs["Job A"].append(Task()) model.jobs["Job A"][1].processing_times["Machine Y"] = 20 model.jobs["Job A"].append(Task()) model.jobs["Job A"][2].processing_times["Machine Z"] = 15 model.jobs["Job B"].append(Task()) model.jobs["Job B"][0].processing_times["Machine Y"] = 10 model.jobs["Job B"].append(Task()) model.jobs["Job B"][1].processing_times["Machine Z"] = 20 model.jobs["Job B"].append(Task()) model.jobs["Job B"][2].processing_times["Machine X"] = 15 model.jobs["Job C"].append(Task()) model.jobs["Job C"][0].processing_times["Machine Z"] = 10 model.jobs["Job C"].append(Task()) model.jobs["Job C"][1].processing_times["Machine X"] = 20 model.jobs["Job C"].append(Task()) model.jobs["Job C"][2].processing_times["Machine Y"] = 15 model.jobs["Job D"].append(Task()) model.jobs["Job D"][0].processing_times["Machine Z"] = 10 model.jobs["Job D"].append(Task()) model.jobs["Job D"][1].processing_times["Machine Y"] = 20 model.jobs["Job D"].append(Task()) model.jobs["Job D"][2].processing_times["Machine X"] = 15 # solve using Amplify Scheduling Engine token = "YOUR TOKEN" gantt = model.solve(token=token, timeout=1) # visualize solution gantt.timeline()
- property jobs: Jobs
job list
Jobs is managed as dict. Key is job name and value is
Job
object.Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].index 0 >>> model.jobs["Job A"].name 'Job A' >>> model.jobs["Job A"] {'tasks': [], 'no_wait': None, 'dependent_jobs': []}
- property machines: Machines
machine list
Machines is managed as dict. Key is machine name and value is
Machine
object.Examples
>>> model = Model() >>> model.machines.add("Machine A") >>> model.machines["Machine A"].index 0 >>> model.machines["Machine A"].name 'Machine A' >>> model.machines["Machine A"] {'maintenance_times': [], 'buffer_size': None, 'setup_times': []}
- property resources: Resources
resource list
Resources is managed as dict. Key is resource name and value is
Resource
object.Examples
>>> model = Model() >>> model.resources.add("Resource A") >>> model.resources["Resource A"].index 0 >>> model.resources["Resource A"].name 'Resource A' >>> model.resources["Resource A"] {'capacity': 1}
- solve(token: str, timeout: int, base_url: str = 'https://amplify.fixstars.com/api/schd') Schedule
solve the scheduling problem
- Parameters:
token – token of Amplify Scheduling Engine
timeout – search timeout seconds of Amplify Scheduling Engine
base_url – base url of Amplify Scheduling Engine
- Returns:
Solution obtained from Amplify Scheduling Engine
- class ProcessingTimes(parent: Task)
ProcessingTimes is container of processing time.
Processing times is managed as dict; key is
Machine
and value is processing time.- Parameters:
parent – parent object which this object belong to
- items()
Return machine and processing time.
- keys()
Return machine list registered in processing times.
- values()
Return processing times.
- class RequiredResources(parent: Task)
RequiredResources is container of required resource.
RequiredResources is managed as list of
Resource
.- Parameters:
parent – parent object which this object belong to
- class Resource(name: str = '')
A class managing a resource.
- Parameters:
name – resource name. Defaults to “”
- property capacity: int
resource capacity
Capacity must be positive number.
Examples
>>> model = Model() >>> model.resources.add("Resource A") >>> model.resources["Resource A"].capacity = 2 >>> model.resources["Resource A"].capacity 2
- class Resources(parent: Model)
Resources is container of resource.
Resources is managed as dict. Key is resource name and value is
Resource
.- Parameters:
parent – parent object which this object belong to
- add(resource: str | Resource) None
add resource as dict Key is resource.name and value is resource itself.
- Parameters:
resource – resource to adding. If resource type is str, create new
Resource
instance using resource as name.- Raises:
ValueError – If resource is no name.
- items()
- keys()
- values()
- class Schedule(result: list[dict[str, int]], status: str, model: Model)
Solution of a scheduling problem.
Schedule
is generated from solve method ofModel
.- resource(avail_view: bool = False, width: int | None = None, height: int | None = None) Figure
Solution resource figure.
- Parameters:
avail_view –
True
to show a timeline for available resources,False
for requirements. Defaults toFalse
.width – The figure width in pixels. Defaults to
None
.height – The figure height in pixels. Defaults to
None
.
- Returns:
resource figure
- Return type:
- property table: DataFrame
solution in table format
- timeline(machine_view: bool = False, separated_by_task: bool = False, width: int | None = None, height: int | None = None) Figure
Solution timeline figure.
- Parameters:
machine_view –
True
to show a timeline for each machine,False
for each job. Defaults toFalse
.separated_by_task – True to separate GanttChart by process index or False to not separate it. Defaults to
False
.width – The figure width in pixels. Defaults to
None
.height – The figure height in pixels. Defaults to
None
.
- Returns:
timeline figure
- Return type:
- class SetupTimes(parent: Machine)
SetupTimes is container of setup time.
Setup times is managed as list of tuple. Each element represents one setup time, and each setup time is a tuple with 1 to 3 elements.
- Parameters:
parent – parent object which this object belong to
- class Task
A class managing a task.
In a scheduling problem, a task is processed by a machine at a specified processing time. The main target for optimization is the task start time.
- add_required_resource(resource: Resource | str) None
add required resource
- Parameters:
resource – required resource
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.resources.add("Resource A") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][0].add_required_resource("Resource A") >>> model.jobs["Job A"][0].required_resources ['Resource A']
- add_transportation_time(transportation_time: int | timedelta, src_machine: Machine | str, dst_machine: Machine | str) None
add transportation time
add transportation time from a source machine to a destination machine
- Parameters:
transportation_time – transportation time
src_machine – source machine
dst_machine – destination machine
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.machines.add("Machine A") >>> model.machines.add("Machine B") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][0].add_transportation_time(10, "Machine A", "Machine B") >>> model.jobs["Job A"][0].transportation_times [(10, 'Machine A', 'Machine B')]
- property deadline: int | datetime | time | None
deadline for this task
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][0].deadline = 30 >>> model.jobs["Job A"][0].deadline 30
- property index: int
task index
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][0].index 0 >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][1].index 1
- property processing_times: ProcessingTimes
processing times for this task
Processing times is managed as dict; key is the machine name and value is processing time. During registration, the existence of the machine in the model is checked. Registering a machine name that does not exist in the model will result in a ValueError. If this task is not registered in the model, a ValueError will occur because the existence of the machine cannot be checked.
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.machines.add("Machine A") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][0].processing_times["Machine A"] = 10 >>> model.jobs["Job A"][0].processing_times["Machine A"] 10
- property release_time: int | datetime | time | None
release time size for this task
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][0].release_time = 20 >>> model.jobs["Job A"][0].release_time 20
- property required_buffer_size: int | None
required buffer size for this task
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][0].required_buffer_size = 2 >>> model.jobs["Job A"][0].required_buffer_size 2
- property required_resources: RequiredResources
required resources for this task
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.resources.add("Resource A") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][0].required_resources.append("Resource A") >>> model.jobs["Job A"][0].required_resources ['Resource A']
- property transportation_times: TransportationTimes
transportation time for this task
Examples
>>> model = Model() >>> model.jobs.add("Job A") >>> model.machines.add("Machine A") >>> model.machines.add("Machine B") >>> model.jobs["Job A"].append(Task()) >>> model.jobs["Job A"][0].transportation_times.append((10, "Machine A", "Machine B")) >>> model.jobs["Job A"][0].transportation_times [(10, 'Machine A', 'Machine B')]
- class TransportationTimes(parent: Task)
TransportationTimes is container of transportation time.
Transportation times is managed as list of tuple. Each element represents one transportation time, and each transportation time is a tuple with 1 to 3 elements.
- Parameters:
parent – parent object which this object belong to