Source code for scml.scml2020.components.production

__all__ = [
    "ProductionStrategy",
    "SupplyDrivenProductionStrategy",
    "DemandDrivenProductionStrategy",
    "TradeDrivenProductionStrategy",
]

from typing import Dict
from typing import List
from typing import Tuple

import numpy as np
from negmas import Contract

from scml.scml2020.common import NO_COMMAND


[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) 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, )