scml.scml2019.world
Classes
Represents a factory within an SCML world. It is only accessed by the SCML2020World so it need not be made public. |
|
The |
Module Contents
- class scml.scml2019.world.Factory[source]
Represents a factory within an SCML world. It is only accessed by the SCML2020World so it need not be made public.
- profiles: List[ManufacturingProfile] = []
A list of profiles used to initialize the factory
- _commands: numpy.ndarray
The production command currently running
- _line_schedules: numpy.ndarray
Amount of money hidden by the agent
Mapping from product index to the amount hidden by the agent
- _jobs: Dict[Tuple[int, int], Job]
The jobs waiting to be run on the factory indexed by (time, line) tuples
- _carried_updates: FactoryStatusUpdate
Carried updates from last executed command
- _world: negmas.situated.World = None
- property commands: numpy.ndarray
- property line_schedules: numpy.ndarray
- schedule(job: Job, override=False) None [source]
Schedules the given job at its
time
andline
optionally overriding whatever was already scheduled :param job: :param override:- Returns:
Success/failure
- _apply_updates(updates: FactoryStatusUpdate) None [source]
- step() List[ProductionReport] [source]
- _run(profile: ManufacturingProfile, override=True) None [source]
running is executed at the beginning of the step t
- Parameters:
profile – the profile to start giving both the line and process
override – If true, override any running processes paying cancellation cost for these processes
Remarks:
The output of a process that runs from step t to step t + n - 1 will only be in storage at step t + n
- _pause(line: int) None [source]
pausing is executed at the end of the step
- Parameters:
line – the line on which the process is running
- Returns:
The status updated for all times that need to be updated to cancel the command if it is not None. If None is returned then scheduling failed.
- Return type:
Optional[Dict[int, FactoryStatusUpdate]]
Remarks:
Not implemented yet
pausing when nothing is running is not an error and will return an empty status update
- _resume(line: int) None [source]
resumption is executed at the end of the step (starting next step count down)
- Parameters:
line – the line on which the process is running
- Returns:
The status updated for all times that need to be updated to cancel the command if it is not None. If None is returned then scheduling failed.
- Return type:
Optional[Dict[int, FactoryStatusUpdate]]
Remarks:
Not implemented yet
resuming when nothing is paused is not an error and will return an empty status update
- _stop(line: int) None [source]
stopping is executed at the beginning of the current step
- Parameters:
line – the line on which the process is running
- Returns:
The status updated for all times that need to be updated to cancel the command if it is not None. If None is returned then scheduling failed.
- Return type:
Optional[Dict[int, FactoryStatusUpdate]]
Remarks:
stopping when nothing is running is not an error and will just return an empty schedule
- _step_line(line: int) ProductionReport [source]
Steps the line to the time-step
t
assuming that it is already stepped to time-step t-1 given the storage- Parameters:
line – the line to step
- Returns:
ProductionReport
- class scml.scml2019.world.SCML2019World(products: Collection[scml.scml2019.common.Product], processes: Collection[scml.scml2019.common.Process], factories: List[scml.scml2019.common.Factory], consumers: List[scml.scml2019.consumers.Consumer], miners: List[scml.scml2019.miners.Miner], factory_managers: List[scml.scml2019.factory_managers.builtins.FactoryManager] | None = None, n_steps=100, time_limit=60 * 90, mechanisms: Dict[str, Dict[str, Any]] | None = None, neg_n_steps=20, neg_time_limit=2 * 60, neg_step_time_limit=60, negotiation_speed=21, no_bank=False, minimum_balance=0, interest_rate=0.1, interest_max=0.3, installment_interest=0.2, interest_time_increment=0.02, balance_at_max_interest=None, loan_installments=1, no_insurance=False, premium=0.03, premium_time_increment=0.03, premium_breach_increment=0.001, max_allowed_breach_level=None, breach_processing=BreachProcessing.VICTIM_THEN_PERPETRATOR, breach_penalty_society=0.1, breach_penalty_society_min=0.0, breach_penalty_victim=0.0, breach_move_max_product=True, initial_wallet_balances: int | None = None, money_resolution=0.5, default_signing_delay=0, transportation_delay: int = 0, transfer_delay: int = 0, start_negotiations_immediately=False, catalog_profit=0.15, avg_process_cost_is_public=True, catalog_prices_are_public=True, strip_annotations=True, financial_reports_period=10, ignore_negotiated_penalties=False, prevent_cfp_tampering=False, default_price_for_products_without_one=1, compensation_fraction=0.5, compact=False, log_folder=None, log_to_file: bool = False, log_to_screen: bool = False, log_file_level=logging.DEBUG, log_screen_level=logging.ERROR, log_file_name: str = 'log.txt', log_ufuns: bool = False, log_negotiations: bool = False, save_mechanism_state_in_contract=False, save_signed_contracts: bool = True, save_cancelled_contracts: bool = True, save_negotiations: bool = True, save_resolved_breaches: bool = True, save_unresolved_breaches: bool = True, ignore_agent_exceptions: bool = False, ignore_contract_execution_exceptions: bool = False, name: str | None = None, **kwargs)[source]
Bases:
negmas.situated.TimeInAgreementMixin
,negmas.situated.World
The
SCML2020World
class running a simulation of supply chain management.- agents: Dict[str, scml.scml2019.agent.SCML2019Agent][source]
- products: List[scml.scml2019.common.Product] = [][source]
- processes: List[scml.scml2019.common.Process] = [][source]
- factory_managers: List[scml.scml2019.factory_managers.builtins.FactoryManager] = [][source]
- miners: List[scml.scml2019.miners.Miner] = [][source]
- consumers: List[scml.scml2019.consumers.Consumer] = [][source]
- _report_receivers: Dict[str, Set[scml.scml2019.agent.SCML2019Agent]][source]
- f2a: Dict[str, scml.scml2019.agent.SCML2019Agent][source]
- a2f: Dict[str, scml.scml2019.common.Factory][source]
- __interested_agents: List[List[scml.scml2019.agent.SCML2019Agent]][source]
- _transfer: Dict[int, List[Tuple[scml.scml2019.agent.SCML2019Agent, float]]][source]
- join(x: negmas.situated.Agent, simulation_priority: int = 0)[source]
Add an agent to the world.
- Parameters:
x – The agent to be registered
simulation_priority – The simulation priority. Entities with lower priorities will be stepped first during
Returns:
- save_config(file_name: str) None [source]
Saves the config of the world as a yaml file
- Parameters:
file_name – Name of file to save the config to
Returns:
- assign_managers(factory_managers=Iterable[Union[str, Type[FactoryManager], FactoryManager]], params: Iterable[Dict[str, Any]] | None = None) None [source]
Assigns existing factories to new factory managers created from the given types and parameters or manager objects.
- Parameters:
factory_managers – An iterable of
FactoryManager
objects type names orFactoryManager
types to assign toparams – parameters of the newly created managers
Remarks:
factories are assigned in the same order they exist in the local
factories
attribute cycling through the input managers or types/paramsIf a
FactoryManager
object is given instead of a type or a string in thefactory_managers
collection, and the number offactory_managers
is less than the number of factories in the world causing this object to cycle for more than one factory, it is assigned to the first such factory but then deep copies of it with new ids and names are assigned to the rest of the factories. That ensures that each manager has exactly one factory and that all factories are assigned exactly one unique manager.
- classmethod random_small(n_production_levels: int = 1, n_factories: int = 10, factory_kwargs: Dict[str, Any] = None, miner_kwargs: Dict[str, Any] = None, consumer_kwargs: Dict[str, Any] = None, **kwargs)[source]
- classmethod chain_world(n_intermediate_levels=0, n_miners=5, n_factories_per_level=5, n_consumers: int | Tuple[int, int] | List[int] = 5, n_steps=100, n_lines_per_factory=10, n_max_assignable_factories=None, log_file_name: str = None, agent_names_reveal_type: bool = False, negotiator_type: str = DEFAULT_NEGOTIATOR, miner_type: str | Type[scml.scml2019.miners.Miner] = ReactiveMiner, consumer_type: str | Type[scml.scml2019.consumers.Consumer] = JustInTimeConsumer, max_storage: int = sys.maxsize, default_manager_params: Dict[str, Any] = None, miner_kwargs: Dict[str, Any] = None, consumption: int | Tuple[int, int] = (0, 5), consumer_kwargs: Dict[str, Any] = None, negotiation_speed: int | None = 21, manager_types: Sequence[Type[scml.scml2019.factory_managers.builtins.FactoryManager]] = (GreedyFactoryManager,), manager_params: Sequence[Dict[str, Any]] | None = None, n_default_per_level: int = 0, default_factory_manager_type: Type[scml.scml2019.factory_managers.builtins.FactoryManager] = GreedyFactoryManager, randomize: bool = True, initial_wallet_balances=1000, process_cost: float | Tuple[float, float] = (1.0, 5.0), process_time: int | Tuple[int, int] = 1, interest_rate=float('inf'), interest_max=float('inf'), shared_profile_per_factory=False, **kwargs)[source]
Creates a very small world in which only one raw material and one final product. The production graph is a series with
n_intermediate_levels
intermediate levels between the single raw material and single final product- Parameters:
n_max_assignable_factories – The maximum number of factories assigned to managers other than the default
randomize – If true, the factory assignment is randomized
n_default_per_level – The number of
GreedyFactoryManager
objects guaranteed at every leveldefault_factory_manager_type – The
FactoryManager
type to use as the base for default_factory_managers. You can specify how many of this type exist at every level by specifyingn_default_per_level
. Ifn_default_per_level
is zero, this parameter has no effect.manager_types – A sequence of factory manager types to control the factories.
manager_params – An optional sequence of dictionaries giving the parameters to pass to
manager_types
.consumer_type – Consumer type to use for all consumers
miner_type – Miner type to use for all miners
consumption – Consumption schedule
n_intermediate_levels – The number of intermediate products
n_miners – number of miners of the single raw material
n_factories_per_level – number of factories at every production level
n_consumers – number of consumers of the final product
n_steps – number of simulation steps
n_lines_per_factory – number of lines in each factory
process_cost – The range of process costs. A uniform distribution will be used
process_time – The range of process times. A uniform distribution will be used
log_file_name – File name to store the logs
agent_names_reveal_type – If true, agent names will start with a snake_case version of their type name
negotiator_type – The negotiation factory used to create all negotiators
max_storage – maximum storage capacity for all factory managers If None then it is unlimited
default_manager_params – keyword arguments to be used for constructing factory managers
consumer_kwargs – keyword arguments to be used for constructing consumers
miner_kwargs – keyword arguments to be used for constructing miners
negotiation_speed – The number of negotiation steps per simulation step. None means infinite
interest_max – Maximum interest rate
interest_rate – Minimum interest rate
initial_wallet_balances – initial wallet balances for all factories
shared_profile_per_factory – If true, all lines in the same factory will have the same profile costs
kwargs – Any other parameters are just passed to the world constructor
- Returns:
SCML2019World ready to run
Remarks:
Every production level n has one process only that takes n steps to complete
- classmethod random(n_raw_materials: int | Tuple[int, int] = (5, 10), raw_material_price: float | Tuple[float, float] = (1.0, 30.0), n_final_products: int | Tuple[int, int] = (3, 5), n_production_levels: int | Tuple[int, int] = (3, 5), n_products_per_level: int | Tuple[int, int] = (3, 5), n_processes_per_level: int | Tuple[int, int] = (6, 10), n_inputs_per_process: int | Tuple[int, int] = (2, 5), bias_toward_last_level_products: float = 0.0, quantity_per_input: int | Tuple[int, int] = (1, 10), input_step: float | Tuple[float, float] = 0.0, quantity_per_output: int | Tuple[int, int] = (1, 1), output_step: float | Tuple[float, float] = 1.0, process_relative_cost: float | Tuple[float, float] = (0.05, 0.4), n_outputs_per_process: int | Tuple[int, int] = (1, 1), n_lines: int | Tuple[int, int] = (3, 5), lines_are_similar: bool = False, n_processes_per_line: int | Tuple[int, int] = None, cost_for_line: float | Tuple[float, float] = (5.0, 50.0), n_production_steps: int | Tuple[int, int] = (2, 10), max_storage: int | Tuple[int, int] = 2000, n_factories: int | Tuple[int, int] = 20, n_consumers: int | Tuple[int, int] = 5, n_products_per_consumer: int | Tuple[int, int] = None, n_miners: int | Tuple[int, int] = 5, n_products_per_miner: int | Tuple[int, int] | None = None, factory_manager_types: Type[scml.scml2019.factory_managers.builtins.FactoryManager] | List[Type[scml.scml2019.factory_managers.builtins.FactoryManager]] = GreedyFactoryManager, consumer_types: Type[scml.scml2019.consumers.Consumer] | List[Type[scml.scml2019.consumers.Consumer]] = JustInTimeConsumer, miner_types: Type[scml.scml2019.miners.Miner] | List[Type[scml.scml2019.miners.Miner]] = ReactiveMiner, negotiator_type=DEFAULT_NEGOTIATOR, initial_wallet_balance: float | Tuple[float, float] = 1000, factory_kwargs: Dict[str, Any] = None, miner_kwargs: Dict[str, Any] = None, consumer_kwargs: Dict[str, Any] = None, **kwargs)[source]
Creates a random SCML scenario with adjustable parameters.
- Parameters:
n_raw_materials – Number of raw materials. Can be a value or a range.
raw_material_price – Catalog prices for raw materials. Can be a value or a range.
n_final_products – Number of final products. Can be a value or a range.
n_production_levels – How deep is the production graph (number of intermediate products). Can be a value or
range. (miner. Can be a value or a)
n_products_per_level – How many intermediate products per intermediate level. Can be a value or a range.
n_processes_per_level – Number of processes in intermediate levels. Can be a value or a range.
n_inputs_per_process – Number of inputs per process. Can be a value or a range.
bias_toward_last_level_products – How biased are production processes toward using products from the last
them (level below)
quantity_per_input – How many items are needed for each input to a process. Can be a value or a range.
input_step – When are inputs consumed during the production process. Can be a value or a range. Default 0
quantity_per_output – How many items are produced per output. Can be a value or a range.
output_step – When are outputs created during the production process. Can be a value or a range. Default 1
process_relative_cost – Intrinsic relative cost of processes [Outputs will be produced
sum (at a cost of)
n_outputs_per_process – Number of outputs per process. Can be a value or a range.
n_lines – Number of lines per factory. Can be a value or a range.
lines_are_similar – If true then all lins of the same factory will have the same production processes.
n_processes_per_line – Number of processes that can be run on each line per factory. Can be a value or a
range.
cost_for_line – Cost for running a process on a line. Can be a value or a range.
n_production_steps – Number of production steps per line. Can be a value or a range.
max_storage – Maximum storage per factory. Can be a value or a range.
n_factories – Number of factories. Can be a value or a range.
n_consumers – Number of consumers. Can be a value or a range.
n_products_per_consumer – Number of products per miner. If None then all final products will be assigned to
range.
n_miners – Number of miners. Can be a value or a range.
n_products_per_miner – Number of products per miner. If None then all raw materials will be assigned to every
range.
factory_manager_types – A callable for creating factory managers for the factories
consumer_types – A callable for creating
Consumer
objectsminer_types – A callable for creating
Miner
objectsnegotiator_type – A string that can be `eval`uated to a negotiator.
initial_wallet_balance – The initial balance of all wallets
factory_kwargs – keyword arguments to be used for constructing factory managers
consumer_kwargs – keyword arguments to be used for constructing consumers
miner_kwargs – keyword arguments to be used for constructing miners
**kwargs
- Returns:
SCML2019World
The random world generated
Remarks:
Most parameters accept either a single value or a 2-valued tuple. In the later case, it will sample a value within the range specified by the tuple (low, high) inclusive. For example the number of lines (n_lines) follows this pattern
- _update_dynamic_product_process_info()[source]
Updates the catalog prices of all products based on the prices of their inputs
- set_consumers(consumers: List[scml.scml2019.consumers.Consumer])[source]
- set_miners(miners: List[scml.scml2019.miners.Miner])[source]
- set_factory_managers(factory_managers: List[scml.scml2019.factory_managers.builtins.FactoryManager] | None)[source]
- set_processes(processes: Collection[scml.scml2019.common.Process])[source]
- set_products(products: Collection[scml.scml2019.common.Product])[source]
- order_contracts_for_execution(contracts: Collection[negmas.situated.Contract])[source]
Orders the contracts in a specific time-step that are about to be executed
- execute_action(action: negmas.situated.Action, agent: negmas.situated.Agent, callback: Callable[[negmas.situated.Action, bool], Any] = None) bool [source]
Executes the given action by the given agent
- get_private_state(agent: negmas.situated.Agent) scml.scml2019.common.FactoryState [source]
Reads the private state of the given agent
- receive_financial_reports(agent: scml.scml2019.agent.SCML2019Agent, receive: bool, agents: List[str] | None)[source]
Registers interest/disinterest in receiving financial reports
- pre_step_stats()[source]
Called at the beginning of the simulation step to prepare stats or update them
Kept for backward compatibility and will be dropped. Override
update_stats
instead
- start_contract_execution(contract: negmas.situated.Contract) Set[negmas.situated.Breach] [source]
Tries to execute the contract
- Parameters:
contract
- Returns:
The set of breaches committed if any. If there are no breaches return an empty set
- Return type:
Set[Breach]
Remarks:
You must call super() implementation of this method before doing anything
It is possible to return None which indicates that the contract was nullified (i.e. not executed due to a reason other than an execution exeception).
- _move_product(buyer: scml.scml2019.agent.SCML2019Agent, seller: scml.scml2019.agent.SCML2019Agent, product_id: int, quantity: int, money: float)[source]
Moves as much product and money between the buyer and seller
- complete_contract_execution(contract: negmas.situated.Contract, breaches: List[negmas.situated.Breach], resolution: negmas.situated.Contract | None)[source]
The resolution can either be None or a contract with the following items:
The issues can be any or all of the following:
immediate_quantity: int immediate_unit_price: float later_quantity: int later_unit_price: int later_penalty: float later_time: int
- _move_product_force(buyer: scml.scml2019.agent.SCML2019Agent, seller: scml.scml2019.agent.SCML2019Agent, product_id: int, quantity: int, money: float)[source]
Moves as much product and money between the buyer and seller
- register_interest(agent: scml.scml2019.agent.SCML2019Agent, products: List[int]) None [source]
- unregister_interest(agent: scml.scml2019.agent.SCML2019Agent, products: List[int]) None [source]
- make_bankrupt(agent: scml.scml2019.agent.SCML2019Agent, amount: float, beneficiary: negmas.situated.Agent, contract: negmas.situated.Contract | None) None [source]
Marks the agent as bankrupt
- evaluate_insurance(contract: negmas.situated.Contract, agent: scml.scml2019.agent.SCML2019Agent, t: int = None) float | None [source]
Can be called to evaluate the premium for insuring the given contract against breachs committed by others
- Parameters:
agent – The agent buying the contract
contract – hypothetical contract
t – time at which the policy is to be bought. If None, it means current step
- buy_insurance(contract: negmas.situated.Contract, agent: scml.scml2019.agent.SCML2019Agent) bool [source]
Buys insurance for the contract by the premium calculated by the insurance company.
- Remarks:
The agent can call
evaluate_insurance
to find the premium that will be used.
- _process_annotation(annotation: Dict[str, Any] | None) Dict[str, Any] | None [source]
Processes an annotation stripping any extra information not allowed if necessary. Will return None if the annotation is suspecious
- run_negotiation(caller: negmas.situated.Agent, issues: Collection[negmas.outcomes.Issue], partners: Collection[negmas.situated.Agent], negotiator: negmas.Negotiator, ufun: negmas.UtilityFunction = None, caller_role: str = None, roles: Collection[str] = None, annotation: Dict[str, Any] | None = None, mechanism_name: str = None, mechanism_params: Dict[str, Any] = None) Tuple[negmas.situated.Contract, negmas.NegotiatorMechanismInterface] | None [source]
Runs a negotiation until completion
- Parameters:
caller – The agent requesting the negotiation
partners – A list of partners to participate in the negotiation. Note that the caller itself may not be in this list which makes it possible for an agent to request a negotaition that it does not participate in. If that is not to be allowed in some world, override this method and explicitly check for these kinds of negotiations and return False. If partners is passed as a single string/
Agent
or as a list containing a single string/Agent
, then he caller will be added at the beginning of the list. This will only be done ifroles
was passed as None.negotiator – The negotiator to be used in the negotiation
preferences – The utility function. Only needed if the negotiator does not already know it
caller_role – The role of the caller in the negotiation
issues – Negotiation issues
annotation – Extra information to be passed to the
partners
when asking them to join the negotiationpartners – A list of partners to participate in the negotiation
roles – The roles of different partners. If None then each role for each partner will be None
mechanism_name – Name of the mechanism to use. It must be one of the mechanism_names that are supported by the
None (must also be)
my_role (then roles and)
None
mechanism_params – A dict of parameters used to initialize the mechanism object
- Returns:
A Tuple of a contract and the nmi of the mechanism used to get it in case of success. None otherwise
- run_negotiations(caller: negmas.situated.Agent, issues: List[negmas.outcomes.Issue] | List[List[negmas.outcomes.Issue]], partners: List[List[negmas.situated.Agent]], negotiators: List[negmas.Negotiator], ufuns: List[negmas.UtilityFunction] = None, caller_roles: List[str] = None, roles: List[List[str] | None] | None = None, annotations: List[Dict[str, Any] | None] | None = None, mechanism_names: str | List[str] | None = None, mechanism_params: Dict[str, Any] | List[Dict[str, Any]] | None = None, all_or_none: bool = False) List[Tuple[negmas.situated.Contract, negmas.NegotiatorMechanismInterface]] [source]
Requests to run a set of negotiations simultaneously. Returns after all negotiations are run to completion
- Parameters:
caller – The agent requesting the negotiation
partners – A list of list of partners to participate in the negotiation. Note that the caller itself may not be in this list which makes it possible for an agent to request a negotaition that it does not participate in. If that is not to be allowed in some world, override this method and explicitly check for these kinds of negotiations and return False. If partners[i] is passed as a single string/
Agent
or as a list containing a single string/Agent
, then he caller will be added at the beginning of the list. This will only be done ifroles
was passed as None.issues – Negotiation issues
negotiators – The negotiator to be used in the negotiation
ufuns – The utility function. Only needed if the negotiator does not already know it
caller_roles – The role of the caller in the negotiation
annotations – Extra information to be passed to the
partners
when asking them to join the negotiationpartners – A list of partners to participate in the negotiation
roles – The roles of different partners. If None then each role for each partner will be None
mechanism_names – Name of the mechanism to use. It must be one of the mechanism_names that are supported by the
None (must also be)
my_role (then roles and)
None
mechanism_params – A dict of parameters used to initialize the mechanism object
all_of_none – If True, ALL partners must agree to negotiate to go through.
- Returns:
contract (None for failure) and nmi (The mechanism info [None if the partner refused the negotiation])
- Return type:
A list of tuples each with two values
- request_negotiation_about(req_id: str, caller: negmas.situated.Agent, issues: List[negmas.outcomes.Issue], partners: List[negmas.situated.Agent], roles: List[str] = None, annotation: Dict[str, Any] | None = None, mechanism_name: str = None, mechanism_params: Dict[str, Any] = None, group=None)[source]
Requests to start a negotiation with some other agents
- Parameters:
req_id – An ID For the request that is unique to the caller
caller – The agent requesting the negotiation
partners – A list of partners to participate in the negotiation. Note that the caller itself may not be in this list which makes it possible for an agent to request a negotaition that it does not participate in. If that is not to be allowed in some world, override this method and explicitly check for these kinds of negotiations and return False. If partners is passed as a single string/
Agent
or as a list containing a single string/Agent
, then he caller will be added at the beginning of the list. This will only be done ifroles
was passed as None.issues – Negotiation issues
annotation – Extra information to be passed to the
partners
when asking them to join the negotiationpartners – A list of partners to participate in the negotiation
roles – The roles of different partners. If None then each role for each partner will be None
mechanism_name – Name of the mechanism to use. It must be one of the mechanism_names that are supported by the
None (must also be)
my_role (then roles and)
None
mechanism_params – A dict of parameters used to initialize the mechanism object
group – An identifier for the group to which the negotiation belongs. This is not not used by the system.
- Returns:
None. The caller will be informed by a callback function
on_neg_request_accepted
oron_neg_request_rejected
about the status of the negotiation.
- on_event(event: negmas.events.Event, sender: negmas.events.EventSource) None [source]
Called whenever an event is raised for which the
SCML2020World
is registered asa listener- Parameters:
event – The event
sender – The sender
- Returns:
None
- contract_record(contract: negmas.situated.Contract) Dict[str, Any] [source]
Converts a contract to a record suitable for permanent storage