scml.scml2019.common

Common data-structures and objects used throughout the SCM world implementation

Module Contents

Classes

Product
InputOutput	An input/output to a production process
Process
ManufacturingProfile	The costs/time required for running a process on a line (with associated cancellation costs etc). This
FactoryStatusUpdate
RunningCommandInfo
Job	Describes a job to be run on one production line of a Factory.
ProductionNeed	Describes some quantity of a product that is needed to honor a (sell) contract.
MissingInput
ProductionFailure
ProductionReport
SCMLAgreement
CFP	A Call for proposal upon which a negotiation can start
SCMLAction
ManufacturingProfileCompiled	The costs/time required for running a process on a line (with associated cancellation costs etc).
ProductManufacturingInfo	Gives full information about a manufacturing process that can generate or consume a product.
FinancialReport	Reports that financial standing of an agent at a given time in the simulation
Loan
InsurancePolicy
FactoryState	Read Only State of a factory
Factory	Represents a factory within an SCML world. It is only accessed by the SCML2020World so it need not be made public.

Attributes

INVALID_STEP
NO_PRODUCTION
DEFAULT_NEGOTIATOR
QUANTITY	Index of quantity in negotiation issues
TIME	Index of time in negotiation issues
UNIT_PRICE	Index of unit price in negotiation issues

scml.scml2019.common.INVALID_STEP

scml.scml2019.common.NO_PRODUCTION

scml.scml2019.common.DEFAULT_NEGOTIATOR = 'negmas.sao.AspirationNegotiator'

scml.scml2019.common.QUANTITY = 0

Index of quantity in negotiation issues

scml.scml2019.common.TIME = 1

Index of time in negotiation issues

scml.scml2019.common.UNIT_PRICE = 2

Index of unit price in negotiation issues

class scml.scml2019.common.Product

__slots__ = ['id', 'production_level', 'name', 'expires_in', 'catalog_price']

A product that can be transacted in.

id: int

Product index. Must be set during construction and MUST be unique for products in the same world

production_level: int

The level of this product in the production graph.

name: str

Object name

expires_in: int | None

Number of steps within which the product must be consumed. None means never

catalog_price: float | None

Catalog price of the product.

__str__()

String representation is simply the name

__post_init__()

__hash__()

Return hash(self).

class scml.scml2019.common.InputOutput

An input/output to a production process

__slots__ = ['product', 'quantity', 'step']

product: int

Index of the product used as input or output

quantity: int

Quantity needed/produced

step: float

Relative time within the production at which the input is needed (output is produced)

class scml.scml2019.common.Process

__slots__ = ['id', 'production_level', 'name', 'inputs', 'outputs', 'historical_cost']

id: int

A manufacturing process.

production_level: int

The level of this process in the production graph

name: str

Object name

inputs: List[InputOutput]

list of input product name + quantity required and time of consumption relative to the time required for production (value from 0 to 1)

outputs: List[InputOutput]

list of output product names, quantity required and when it becomes available relative to the time required for production (value from 0 to 1)

historical_cost: float | None

Average cost for running this process in some world. Filled by the world

__str__()

String representation is simply the name

__post_init__()

__hash__()

The hash depends only on the name

class scml.scml2019.common.ManufacturingProfile

The costs/time required for running a process on a line (with associated cancellation costs etc). This data-structure carries full information about the Process es instead of just its index as in ManufacturingProfileCompiled. It is intended to be used to construct factories

See also

Factory

__slots__ = ['n_steps', 'cost', 'initial_pause_cost', 'running_pause_cost', 'resumption_cost',...

n_steps: int

Number of steps needed to complete the manufacturing

cost: float

Cost of manufacturing

initial_pause_cost: float

Cost of pausing incurred only at the step a pause is started

running_pause_cost: float

Running cost of pausing

resumption_cost: float

Cost of resuming a process

cancellation_cost: float

Cost of cancelling the process before the last step

line: int

The line index

process: Process

The Process associated with this profile

class scml.scml2019.common.FactoryStatusUpdate

property is_empty

__slots__ = ['balance', 'storage']

balance: float

The update to the balance

storage: Dict[int, int]

The updates to be applied to the storage after this step

__post_init__()

make_empty() → None

Makes the update an empty one.

combine(other: FactoryStatusUpdate) → None

Combines this status update with another one in place

Parameters:

other – The other status update

Returns:

None

classmethod combine_sets(dst: Dict[int, FactoryStatusUpdate], src: Dict[int, FactoryStatusUpdate])

Combines a set of updates over time with another in place (overriding first) :param dst: First set of updates to be combined into :param src: second set of updates to be combined from

Returns:

classmethod empty()

__str__()

Return str(self).

class scml.scml2019.common.RunningCommandInfo

property n_steps: int

property is_none

__slots__ = ['profile', 'beg', 'end', 'action', 'updates', 'step', 'paused']

profile: ManufacturingProfile

The manufacturing profile associated with this command. Most importantly, it gives the process and line

beg: int

The time the command is to be executed

end: int

The number of steps starting at beg for this command to end (it ends at end - 1)

step: int

The time-step relative to beg at the factory is currently executing the Process indicated in profile. step will always go up by one every simulation step except if the command is paused where it does not change

paused: bool

True if the command is paused

action: str

The command type. For the current implementation it will always be run or none for no command

updates: Dict[int, FactoryStatusUpdate]

The status updates implied by this command with their times relative to beg

ended_before(t: int)

started_on_or_after(t: int)

__str__()

Return str(self).

classmethod do_nothing()

class scml.scml2019.common.Job

Describes a job to be run on one production line of a Factory.

__slots__ = ['profile', 'time', 'line', 'action', 'contract', 'override']

profile: int

The process for run commands

time: int

The time the command is to be executed

line: int

Index of the line on which the job is to be scheduled. Notice that it will be ignored for run actions.

action: str

The command type. For the current implementation it can be run/pause/resume/stop/cancel with cancel cancelling any other command type.

contract: negmas.situated.Contract | None

The sell contract associated with the command

override: bool

Whether to override existing commands when the job is to be executed.

__str__()

Return str(self).

is_cancelling(job: Job) → bool

Determines if the given jobs cancels this one

Parameters:

job –

Returns:

class scml.scml2019.common.ProductionNeed

Describes some quantity of a product that is needed to honor a (sell) contract.

__slots__ = ['product', 'needed_for', 'quantity_to_buy', 'quantity_in_storage', 'step']

product: int

The product needed

needed_for: negmas.situated.Contract

The contract for which the product is needed

quantity_to_buy: int

The quantity need to be bought

quantity_in_storage: int

The quantity already found in storage

step: int

The time step at which the product is needed

__str__()

String representation is simply the name

class scml.scml2019.common.MissingInput

__slots__ = ['product', 'quantity']

product: int

quantity: int

__str__()

Return str(self).

class scml.scml2019.common.ProductionFailure

__slots__ = ['line', 'command', 'missing_inputs', 'missing_money', 'missing_space']

line: int

ID of the line that failed

command: RunningCommandInfo

Information about the command that failed

missing_inputs: List[MissingInput]

The missing inputs if any with their quantities

missing_money: float

The amount of money needed for production that is not available

missing_space: int

The amount space needed in storage but not found

__str__()

Return str(self).

class scml.scml2019.common.ProductionReport

property failed

property is_empty

property no_production

line: int

ID of the line

started: RunningCommandInfo | None

Commands started

continuing: RunningCommandInfo | None

Command that is continuing

finished: RunningCommandInfo | None

Command finished

failure: ProductionFailure | None

Failures

updates: FactoryStatusUpdate

Updates applied to the factory

__str__()

Return str(self).

class scml.scml2019.common.SCMLAgreement

time: int

delivery time

unit_price: float

unit price

quantity: int

quantity

penalty: float | None

penalty

signing_delay: int

Delay between agreement conclusion and signing it to be binding

__getitem__(k)

get(k, default=None)

asdict()

to_dict()

keys()

values()

items()

class scml.scml2019.common.CFP

A Call for proposal upon which a negotiation can start

property issues

Returns the set of issues associated with this CFP. Notice that some of the issues may have a single value

property outcomes

property min_time

property max_time

property min_quantity

property max_quantity

property min_unit_price

property max_unit_price

property min_signing_delay

property max_signing_delay

property min_penalty

property max_penalty

is_buy: bool

If true, the author wants to buy otherwise to sell. Non-negotiable.

publisher: str

the publisher name. Non-negotiable.

product: int

product ID. Non-negotiable.

time: int | Tuple[int, int] | List[int]

delivery time. May be negotiable.

unit_price: float | Tuple[float, float] | List[float]

unit price. May be negotiable.

quantity: int | Tuple[int, int] | List[int]

quantity. May be negotiable.

penalty: float | Tuple[float, float] | List[float] | None

penalty per missing item in case the seller cannot provide the required quantity. May be negotiable.

signing_delay: int | Tuple[int, int] | List[int] | None

The grace period after which the agents are asked to confirm signing the contract

money_resolution: float | None

If not None then it is the minimum unit of money (e.g. 1 for dollar, 0.01 for cent, etc)

id: str

Unique CFP ID

__str__()

Return str(self).

satisfies(query: Dict[str, Any]) → bool

Tests whether the CFP satisfies the conditions set by the query

Parameters:

query – A dictionary given the conditions. See Remarks for details

Remarks:

• The query dictionary can be used to specify any conditions that are required in the CFP. Only CFPs that satisfy ALL the conditions specified in the query are considered satisfying the query. The following keys can be set with corresponding meanings:

  is_buy

  True or False. If both are OK, just do not add this key

  publisher

  A string or SCML2019Agent specifying a specific publisher

  publishers

  A list of publishers (see publisher key)

  product

  A string specifying a product name

  products

  A list of products (see product key)

  time

  A number, list or 2-items-tuple (range) specifying possible times to consider satisfactory

  unit_price

  A number, list or 2-items-tuple (range) specifying possible prices to consider satisfactory

  quantity

  A number, list or 2-items-tuple (range) specifying possible quantities to consider OK

  penalty

  A number, list or 2-items-tuple (range) specifying possible penalties to consider satisfactory

to_dict()

classmethod from_dict(idict: Dict[str, Any], class_name: str | None = None) → CFP

class scml.scml2019.common.SCMLAction

line: str

Line to execute the action on (need not be given if the profile is given

profile: int | None

Index of the profile to execute

action: str

The action which may be start, stop, pause, resume

time: int = 0

Time to execute the action at

class scml.scml2019.common.ManufacturingProfileCompiled

The costs/time required for running a process on a line (with associated cancellation costs etc).

See also

Factory

__slots__ = ['n_steps', 'cost', 'initial_pause_cost', 'running_pause_cost', 'resumption_cost',...

n_steps: int

Number of steps needed to complete the manufacturing

cost: float

Cost of manufacturing

initial_pause_cost: float

Cost of pausing incurred only at the step a pause is started

running_pause_cost: float

Running cost of pausing

resumption_cost: float

Cost of resuming a process

cancellation_cost: float

Cost of cancelling the process before the last step

line: int

The line index

process: int

The Process index

classmethod from_manufacturing_profile(profile: ManufacturingProfile, process2ind: Dict[Process, int])

class scml.scml2019.common.ProductManufacturingInfo

Gives full information about a manufacturing process that can generate or consume a product.

See also

consuming and producing of Factory

__slots__ = ['profile', 'quantity', 'step']

profile: int

The ManufacturingProfile index

quantity: int

The quantity generated/consumed by running this manufacturing info

step: int

The step from the beginning at which the Product is received/consumed

class scml.scml2019.common.FinancialReport

Reports that financial standing of an agent at a given time in the simulation

property balance

The balance of the agent defined as the difference between its available cash + inventory and its liabilities

Remarks:

• If the inventory was not calculated (due to having at least one product with unknown catalog price), it is used as zero in the equation.

agent: str

Agent ID

step: int

Time of the report

cash: float

Cash at hand

liabilities: float

Total liabilities (loans)

inventory: float

Value of everything in the inventory priced at catalog prices.

credit_rating: float

The agent’s credit rating as a fraction of the maximum credit rating (1 indicates highest credit rating).

class scml.scml2019.common.Loan

amount: float

Loan amount

starts_at: int

The time-step at which payment starts

total: float

The total to be paid including the amount + interests

interest: float

The interest rate per step

installment: float

The amount to be paid in one installment

n_installments: int

The number of installments

__str__()

Return str(self).

class scml.scml2019.common.InsurancePolicy

premium: float

contract: negmas.situated.Contract

at_time: int

against: scml.scml2019.agent.SCML2019Agent

class scml.scml2019.common.FactoryState

Read Only State of a factory

max_storage: int

Maximum storage allowed in this factory

line_schedules: numpy.ndarray

An array of n_lines * n_steps giving the line schedules

storage: Dict[int, int]

Mapping from product index to the amount available in the inventory

wallet: float

Money available for purchases

hidden_money: float

Amount of money hidden by the agent

hidden_storage: Dict[int, int]

Mapping from product index to the amount hidden by the agent

loans: float

The total money owned as loans

n_lines: int

The number of lines in the factory, will be set using the profiles input

profiles: List[ManufacturingProfile]

A list of profiles used to initialize the factory

next_step: int

Next simulation step for this factory

commands: numpy.ndarray

The production command currently running

jobs: Dict[Tuple[int, int], Job]

The jobs waiting to be run on the factory indexed by (time, line) tuples

class scml.scml2019.common.Factory

Represents a factory within an SCML world. It is only accessed by the SCML2020World so it need not be made public.

property hidden_money: float

property hidden_storage: Dict[int, int]

property n_lines: int

property jobs: Dict[Tuple[int, int], Job]

property commands: numpy.ndarray

property line_schedules: numpy.ndarray

property wallet: float

property storage: Dict[int, int]

property loans: float

property total_storage: int

property balance: float

The total balance of the factory

property total_balance: float

total balance including hidden money

property next_step: int

initial_storage: dataclasses.InitVar[Dict[int, int]]

Initial storage

initial_wallet: dataclasses.InitVar[float] = 0.0

Initial Wallet

id: str

Object name

profiles: List[ManufacturingProfile]

A list of profiles used to initialize the factory

max_storage: int

Maximum storage allowed in this factory

min_storage: int = 0

Minimum allowed storage per product

min_balance: int | float = 0

Minimum allowed balance

initial_balance: float

Initial balance of the factory

_commands: numpy.ndarray

The production command currently running

_line_schedules: numpy.ndarray

_storage: Dict[int, int]

Mapping from product index to the amount available in the inventory

_total_storage: int

Total storage

_wallet: float

Money available for purchases

_hidden_money: float

Amount of money hidden by the agent

_hidden_storage: Dict[int, int]

Mapping from product index to the amount hidden by the agent

_loans: float

The total money owned as loans

_n_lines: int

The number of lines in the factory, will be set using the profiles input

_jobs: Dict[Tuple[int, int], Job]

The jobs waiting to be run on the factory indexed by (time, line) tuples

_next_step: int

Current simulation step

_carried_updates: FactoryStatusUpdate

Carried updates from last executed command

_world: negmas.situated.World

attach_to_world(world)

__post_init__(initial_storage: Dict[int, int], initial_wallet=0.0)

add_loan(total: float) → None

receive(payment: float) → None

pay(payment: float) → None

transport_to(product: int, quantity: int) → None

buy(product: int, quantity: int, price: float) → None

sell(product: int, quantity: int, price: float) → None

transport_from(product: int, quantity: int) → None

hide_funds(amount: float) → None

hide_product(product: int, quantity: int) → None

unhide_funds(amount: float) → None

unhide_product(product: int, quantity: int) → None

schedule(job: Job, override=False) → None

Schedules the given job at its time and line optionally overriding whatever was already scheduled :param job: :param override:

Returns:

Success/failure

_apply_updates(updates: FactoryStatusUpdate) → None

step() → List[ProductionReport]

_run(profile: ManufacturingProfile, override=True) → None

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

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

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

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

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