Source code for scml.scml2020.common

import sys
from collections import namedtuple
from dataclasses import dataclass

import numpy as np

__all__ = [

"""ID of the system seller agent"""
"""ID of the system buyer agent"""
"""ID of the takeover agent"""
[docs] ANY_STEP = -1
"""Used to indicate any time-step"""
[docs] ANY_LINE = -1
"""Used to indicate any line"""
[docs] NO_COMMAND = -1
"""A constant indicating no command is scheduled on a factory line"""
[docs] INFINITE_COST = sys.maxsize // 2
"""A constant indicating an invalid cost for lines incapable of running some process"""
[docs] QUANTITY = 0
"""Index of quantity in negotiation issues"""
[docs] TIME = 1
"""Index of time in negotiation issues"""
[docs] UNIT_PRICE = 2
"""Index of unit price in negotiation issues"""
[docs] def is_system_agent(aid: str) -> bool: """ Checks whether an agent is a system agent or not Args: aid: Agent ID Returns: True if the ID is for a system agent. """ return ( aid.startswith(SYSTEM_SELLER_ID) or aid.startswith(SYSTEM_BUYER_ID) or aid.startswith(COMPENSATION_ID) )
ContractInfo = namedtuple( "ContractInfo", ["q", "u", "product", "is_seller", "partner", "contract"] ) """Information about a contract including a pointer to it""" @dataclass
[docs] class ExogenousContract: """Represents a contract to be revealed at revelation_time to buyer and seller between them that is not agreed upon through negotiation but is endogenously given"""
[docs] product: int
[docs] quantity: int
[docs] unit_price: int
"""Unit price"""
[docs] time: int
"""Delivery time"""
[docs] revelation_time: int
"""Time at which to reveal the contract to both buyer and seller"""
[docs] seller: int = -1
"""Seller index in the agents array (-1 means "system")"""
[docs] buyer: int = -1
"""Buyer index in the agents array (-1 means "system")"""
[docs] class FinancialReport: """A report published periodically by the system showing the financial standing of an agent"""
[docs] __slots__ = [ "agent_id", "step", "cash", "assets", "breach_prob", "breach_level", "is_bankrupt", "agent_name", ]
[docs] agent_id: str
"""Agent ID"""
[docs] step: int
"""Simulation step at the beginning of which the report was published."""
[docs] cash: int
"""Cash in the agent's wallet. Negative numbers indicate liabilities."""
[docs] assets: int
"""Value of the products in the agent's inventory @ catalog prices. """
[docs] breach_prob: float
"""Number of times the agent breached a contract over the total number of contracts it signed."""
[docs] breach_level: float
"""Sum of the agent's breach levels so far divided by the number of contracts it signed."""
[docs] is_bankrupt: bool
"""Whether the agent is already bankrupt (i.e. incapable of doing any more transactions)."""
[docs] agent_name: str
"""Agent name for printing purposes"""
[docs] def __str__(self): bankrupt = "BANKRUPT" if self.is_bankrupt else "" return ( f"{self.agent_name} @ {self.step} {bankrupt}: Cash: {}, Assets: {self.assets}, " f"breach_prob: {self.breach_prob}, breach_level: {self.breach_level} " f"{'(BANKRUPT)' if self.is_bankrupt else ''}" )
[docs] class FactoryProfile: """Defines all private information of a factory"""
[docs] __slots__ = ["costs"]
[docs] costs: np.ndarray
"""An n_lines * n_processes array giving the cost of executing any process (INVALID_COST indicates infinity)""" @property
[docs] def n_lines(self): return self.costs.shape[0]
[docs] def n_products(self): return self.costs.shape[1] + 1
[docs] def n_processes(self): return self.costs.shape[1]
[docs] def processes(self) -> np.ndarray: """The processes that have valid costs""" return np.nonzero(np.any(self.costs != INFINITE_COST, axis=0))[0]
[docs] def input_products(self) -> np.ndarray: """The input products to all processes runnable (See `processes` )""" return np.nonzero(np.any(self.costs != INFINITE_COST, axis=0))[0]
[docs] def output_products(self) -> np.ndarray: """The output products to all processes runnable (See `processes` )""" return np.nonzero(np.any(self.costs != INFINITE_COST, axis=0))[0] + 1
[docs] class Failure: """A production failure"""
[docs] __slots__ = ["is_inventory", "line", "step", "process"]
[docs] is_inventory: bool
"""True if the cause of failure was insufficient inventory. If False, the cause was insufficient funds. Note that if both conditions were true, only insufficient funds (is_inventory=False) will be reported."""
[docs] line: int
"""The line at which the failure happened"""
[docs] step: int
"""The step at which the failure happened"""
[docs] process: int
"""The process that failed to execute"""
[docs] class FactoryState:
[docs] inventory: np.ndarray
"""An n_products vector giving current quantity of every product in storage"""
[docs] balance: int
"""Current balance in the wallet"""
[docs] commands: np.ndarray
"""n_steps * n_lines array giving the process scheduled on each line at every step for the whole simulation"""
[docs] inventory_changes: np.ndarray
"""Changes in the inventory in the last step"""
[docs] balance_change: int
"""Change in the balance in the last step"""
[docs] contracts: list[list[ContractInfo]]
"""The An n_steps list of lists containing the contracts of this agent by time-step""" @property
[docs] def n_lines(self) -> int: return self.commands.shape[1]
[docs] def n_steps(self) -> int: return self.commands.shape[0]
[docs] def n_products(self) -> int: return len(self.inventory)
[docs] def n_processes(self) -> int: return len(self.inventory) - 1