from __future__ import annotations
from typing import TYPE_CHECKING
import numpy as np
from negmas import Contract
from scml.scml2020.common import NO_COMMAND
if TYPE_CHECKING:
from scml.scml2020.agent import SCML2020Agent
__all__ = [
"ProductionStrategy",
"SupplyDrivenProductionStrategy",
"DemandDrivenProductionStrategy",
"TradeDrivenProductionStrategy",
]
[docs]
class ProductionStrategy:
"""Represents a strategy for controlling production.
Provides:
- `schedule_range` : A mapping from contract ID to a tuple of the first and last steps at which some lines
are occupied to produce the quantity specified by the contract and whether it is a sell contract
- `can_be_produced` : Given a contract, it returns whether or not it is possible to produce the quantity
entailed by it (which means that there is enough vacant production line slots before/after the contracts
delivery time for sell/buy contracts).
Hooks Into:
- `on_contract_breached`
- `on_contract_executed`
Remarks:
- `Attributes` section describes the attributes that can be used to construct the component (passed to its
`__init__` method).
- `Provides` section describes the attributes (methods, properties, data-members) made available by this
component directly. Note that everything provided by the bases of this components are also available to the
agent (Check the `Bases` section above for all the bases of this component).
- `Requires` section describes any requirements from the agent using this component. It defines a set of methods
or properties/data-members that must exist in the agent that uses this component. These requirement are
usually implemented as abstract methods in the component
- `Abstract` section describes abstract methods that MUST be implemented by any descendant of this component.
- `Hooks Into` section describes the methods this component overrides calling `super` () which allows other
components to hook into the same method (by overriding it). Usually callbacks starting with `on_` are
hooked into this way.
- `Overrides` section describes the methods this component overrides without calling `super` effectively
disallowing any other components after it in the MRO to call this method. Usually methods that do some
action (i.e. not starting with `on_`) are overridden this way.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
[docs]
self.schedule_range: dict[str, tuple[int, int, bool]] = dict()
"""Gives the range of steps at which the production needed for a given contract are scheduled"""
[docs]
def can_be_produced(self, contract_id: str):
"""Returns True if the SELL contract given can be honored in principle given the production
capacity of the agent (n. lines). It does not check for the availability of inputs or enough money
to run the production process.
Remarks:
- Cannot be called before calling on_contracts_finalized
"""
earliest, latest, is_sell = self.schedule_range.get(
contract_id, (-1, -1, False)
)
return (
earliest >= self.awi.current_step
if is_sell
else (latest <= self.awi.n_steps - 2 and earliest >= 0)
)
[docs]
def on_contract_executed(self, contract: Contract) -> None:
super().on_contract_executed(contract)
if contract.id in self.schedule_range.keys():
del self.schedule_range[contract.id]
[docs]
def on_contract_breached(self, contract: Contract, breaches, resolution) -> None:
if contract.id in self.schedule_range.keys():
del self.schedule_range[contract.id]
[docs]
class SupplyDrivenProductionStrategy(ProductionStrategy):
"""A production strategy that converts all inputs to outputs
Hooks Into:
- `step`
- `on_contracts_finalized`
Remarks:
- `Attributes` section describes the attributes that can be used to construct the component (passed to its
`__init__` method).
- `Provides` section describes the attributes (methods, properties, data-members) made available by this
component directly. Note that everything provided by the bases of this components are also available to the
agent (Check the `Bases` section above for all the bases of this component).
- `Requires` section describes any requirements from the agent using this component. It defines a set of methods
or properties/data-members that must exist in the agent that uses this component. These requirement are
usually implemented as abstract methods in the component
- `Abstract` section describes abstract methods that MUST be implemented by any descendant of this component.
- `Hooks Into` section describes the methods this component overrides calling `super` () which allows other
components to hook into the same method (by overriding it). Usually callbacks starting with `on_` are
hooked into this way.
- `Overrides` section describes the methods this component overrides without calling `super` effectively
disallowing any other components after it in the MRO to call this method. Usually methods that do some
action (i.e. not starting with `on_`) are overridden this way.
"""
[docs]
def step(self):
super().step()
commands = NO_COMMAND * np.ones(self.awi.n_lines, dtype=int)
inputs = min(self.awi.state.inventory[self.awi.my_input_product], len(commands))
commands[:inputs] = self.awi.my_input_product
commands[inputs:] = NO_COMMAND
self.awi.set_commands(commands)
[docs]
def on_contracts_finalized(
self: SCML2020Agent,
signed: list[Contract],
cancelled: list[Contract],
rejectors: list[list[str]],
) -> None:
super().on_contracts_finalized(signed, cancelled, rejectors)
latest = self.awi.n_steps - 2
earliest_production = self.awi.current_step
for contract in signed:
is_seller = contract.annotation["seller"] == self.id
if is_seller:
continue
step = contract.agreement["time"]
# find the earliest time I can do anything about this contract
if step > latest + 1 or step < earliest_production:
continue
# if I am a seller, I will schedule production
input_product = contract.annotation["product"]
steps, _ = self.awi.schedule_production(
process=input_product,
repeats=contract.agreement["quantity"],
step=(step, latest),
line=-1,
partial_ok=True,
)
self.schedule_range[contract.id] = (
min(steps) if len(steps) > 0 else -1,
max(steps) if len(steps) > 0 else -1,
is_seller,
)
[docs]
class DemandDrivenProductionStrategy(ProductionStrategy):
"""A production strategy that produces ONLY when a contract is secured
Hooks Into:
- `on_contract_finalized`
Remarks:
- `Attributes` section describes the attributes that can be used to construct the component (passed to its
`__init__` method).
- `Provides` section describes the attributes (methods, properties, data-members) made available by this
component directly. Note that everything provided by the bases of this components are also available to the
agent (Check the `Bases` section above for all the bases of this component).
- `Requires` section describes any requirements from the agent using this component. It defines a set of methods
or properties/data-members that must exist in the agent that uses this component. These requirement are
usually implemented as abstract methods in the component
- `Abstract` section describes abstract methods that MUST be implemented by any descendant of this component.
- `Hooks Into` section describes the methods this component overrides calling `super` () which allows other
components to hook into the same method (by overriding it). Usually callbacks starting with `on_` are
hooked into this way.
- `Overrides` section describes the methods this component overrides without calling `super` effectively
disallowing any other components after it in the MRO to call this method. Usually methods that do some
action (i.e. not starting with `on_`) are overridden this way.
"""
[docs]
def on_contracts_finalized(
self: SCML2020Agent,
signed: list[Contract],
cancelled: list[Contract],
rejectors: list[list[str]],
) -> None:
super().on_contracts_finalized(signed, cancelled, rejectors)
for contract in signed:
is_seller = contract.annotation["seller"] == self.id
if not is_seller:
continue
step = contract.agreement["time"]
# find the earliest time I can do anything about this contract
earliest_production = self.awi.current_step
if step > self.awi.n_steps - 1 or step < earliest_production:
continue
# if I am a seller, I will schedule production
output_product = contract.annotation["product"]
input_product = output_product - 1
steps, _ = self.awi.schedule_production(
process=input_product,
repeats=contract.agreement["quantity"],
step=(earliest_production, step - 1),
line=-1,
partial_ok=True,
)
self.schedule_range[contract.id] = (
min(steps) if len(steps) > 0 else -1,
max(steps) if len(steps) > 0 else -1,
is_seller,
)
[docs]
class TradeDrivenProductionStrategy(ProductionStrategy):
"""A production strategy that produces ONLY for contracts that the agent did not initiate.
Hooks Into:
- `on_contract_finalized`
Remarks:
- `Attributes` section describes the attributes that can be used to construct the component (passed to its
`__init__` method).
- `Provides` section describes the attributes (methods, properties, data-members) made available by this
component directly. Note that everything provided by the bases of this components are also available to the
agent (Check the `Bases` section above for all the bases of this component).
- `Requires` section describes any requirements from the agent using this component. It defines a set of methods
or properties/data-members that must exist in the agent that uses this component. These requirement are
usually implemented as abstract methods in the component
- `Abstract` section describes abstract methods that MUST be implemented by any descendant of this component.
- `Hooks Into` section describes the methods this component overrides calling `super` () which allows other
components to hook into the same method (by overriding it). Usually callbacks starting with `on_` are
hooked into this way.
- `Overrides` section describes the methods this component overrides without calling `super` effectively
disallowing any other components after it in the MRO to call this method. Usually methods that do some
action (i.e. not starting with `on_`) are overridden this way.
"""
[docs]
def on_contracts_finalized(
self: SCML2020Agent,
signed: list[Contract],
cancelled: list[Contract],
rejectors: list[list[str]],
) -> None:
super().on_contracts_finalized(signed, cancelled, rejectors)
for contract in signed:
if contract.annotation["caller"] == self.id:
continue
is_seller = contract.annotation["seller"] == self.id
step = contract.agreement["time"]
# find the earliest time I can do anything about this contract
earliest_production = self.awi.current_step
if step > self.awi.n_steps - 1 or step < earliest_production:
continue
# if I am a seller, I will schedule production
output_product = contract.annotation["product"]
input_product = output_product - 1
if is_seller:
steps, _ = self.awi.schedule_production(
process=input_product,
repeats=contract.agreement["quantity"],
step=(earliest_production, step - 2),
line=-1,
partial_ok=True,
)
else:
steps, _ = self.awi.schedule_production(
process=input_product,
repeats=contract.agreement["quantity"],
step=(step, self.awi.n_steps - 2),
line=-1,
partial_ok=True,
)
self.schedule_range[contract.id] = (
min(steps) if len(steps) > 0 else -1,
max(steps) if len(steps) > 0 else -1,
is_seller,
)