"""
State classes for LINE queueing network models (pure Python).
This module provides state representation for network analysis.
"""
from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING
import numpy as np
if TYPE_CHECKING:
from .network import Network
[docs]
class State:
"""
State representation for stochastic network models.
Represents the system state including job populations at each node,
phase information for multi-phase processes, and other state variables.
"""
[docs]
def __init__(self, network: Optional['Network'] = None):
"""
Initialize a state for the network.
Args:
network: The network this state belongs to.
"""
self._network = network
self._state: Dict[int, np.ndarray] = {} # stateful_idx -> state vector
[docs]
def get(self, stateful_idx: int) -> Optional[np.ndarray]:
"""Get state for a stateful node."""
return self._state.get(stateful_idx)
[docs]
def set(self, stateful_idx: int, state: np.ndarray):
"""Set state for a stateful node."""
self._state[stateful_idx] = np.array(state)
[docs]
def toArray(self) -> np.ndarray:
"""Convert to a flat array representation."""
if not self._state:
return np.array([])
# Concatenate all state vectors
return np.concatenate([self._state[i] for i in sorted(self._state.keys())])
@staticmethod
def fromMarginal(model: 'Network', node_idx: int,
n: Union[List[int], np.ndarray]) -> np.ndarray:
"""
Generate state space with specific marginal job counts at a node.
Creates all possible network states where the specified node has
exactly n[r] jobs of class r.
Args:
model: Network model
node_idx: Node index (0-based)
n: Vector of job counts per class
Returns:
State space matrix where each row is a valid state.
"""
from ..api.state.marginal import fromMarginal as _fromMarginal
# Get network struct from model
if hasattr(model, 'get_struct'):
sn = model.get_struct()
elif hasattr(model, 'getStruct'):
sn = model.getStruct()
elif hasattr(model, '_sn'):
sn = model._sn
else:
raise ValueError("Cannot get network struct from model")
return _fromMarginal(sn, node_idx, n)
@staticmethod
def fromMarginalAndRunning(model: 'Network', node_idx: int,
n: Union[List[int], np.ndarray],
s: Union[List[int], np.ndarray]) -> np.ndarray:
"""
Generate state space with specific marginal and running job counts.
Creates states where node has n[r] jobs of class r total,
with s[r] jobs of class r currently in service (running).
Args:
model: Network model
node_idx: Node index (0-based)
n: Vector of total job counts per class
s: Vector of running job counts per class
Returns:
State space matrix where each row is a valid state.
"""
from ..api.state.marginal import fromMarginalAndRunning as _fromMarginalAndRunning
# Get network struct from model
if hasattr(model, 'get_struct'):
sn = model.get_struct()
elif hasattr(model, 'getStruct'):
sn = model.getStruct()
elif hasattr(model, '_sn'):
sn = model._sn
else:
raise ValueError("Cannot get network struct from model")
return _fromMarginalAndRunning(sn, node_idx, n, s)
@staticmethod
def fromMarginalAndStarted(model: 'Network', node_idx: int,
n: Union[List[int], np.ndarray],
s: Union[List[int], np.ndarray]) -> np.ndarray:
"""
Generate state space with specific marginal and started job counts.
Creates states where node has n[r] jobs of class r total,
with s[r] jobs of class r that have started service.
Args:
model: Network model
node_idx: Node index (0-based)
n: Vector of total job counts per class
s: Vector of started job counts per class
Returns:
State space matrix where each row is a valid state.
"""
from ..api.state.marginal import fromMarginalAndStarted as _fromMarginalAndStarted
# Get network struct from model
if hasattr(model, 'get_struct'):
sn = model.get_struct()
elif hasattr(model, 'getStruct'):
sn = model.getStruct()
elif hasattr(model, '_sn'):
sn = model._sn
else:
raise ValueError("Cannot get network struct from model")
return _fromMarginalAndStarted(sn, node_idx, n, s)
@staticmethod
def _struct_of(model):
"""Resolve a NetworkStruct from a model/struct argument."""
if hasattr(model, 'get_struct'):
return model.get_struct()
if hasattr(model, 'getStruct'):
return model.getStruct()
if hasattr(model, '_sn'):
return model._sn
# Already a struct
return model
@staticmethod
def toMarginal(model: 'Network', node_idx: int,
state_i: np.ndarray):
"""
Extract marginal job statistics from a state for a specific node.
Args:
model: Network model (or NetworkStruct).
node_idx: Node index (0-based).
state_i: State vector/matrix for the node.
Returns:
Tuple ``(ni, nir, sir, kir)``:
- ni: total jobs in the node,
- nir: total jobs per class,
- sir: jobs in service per class,
- kir: jobs in service per class and phase.
References:
MATLAB: matlab/src/lang/+State/toMarginal.m
"""
from ..api.state.marginal import toMarginal as _toMarginal
sn = State._struct_of(model)
return _toMarginal(sn, node_idx, state_i)
[docs]
@staticmethod
def fromMarginalBounds(model: 'Network', node_idx: int,
lb, ub, cap=None) -> np.ndarray:
"""
Generate all valid states whose per-class marginal lies between ``lb``
and ``ub`` (inclusive), subject to the node capacity ``cap``.
Args:
model: Network model (or NetworkStruct).
node_idx: Node index (0-based).
lb: Lower bound on the number of resident jobs (scalar total or
per-class vector). ``None`` is treated as all-zero.
ub: Upper bound on the number of resident jobs (scalar total or
per-class vector).
cap: Total capacity at the node (defaults to the station's total
capacity, or infinity if unbounded).
Returns:
State space matrix where each row is a valid state.
References:
MATLAB: matlab/src/lang/+State/fromMarginalBounds.m
"""
from ..api.state.ctmc_ssg import _from_marginal_bounds
from ..api.state.marginal import toMarginal as _toMarginal
sn = State._struct_of(model)
R = sn.nclasses
ub_vec = np.atleast_1d(ub).astype(int)
if ub_vec.size == 1:
ub_vec = np.full(R, int(ub_vec[0]))
# Determine capacity if not supplied.
if cap is None:
ist = int(sn.nodeToStation[node_idx]) if sn.isstation[node_idx] else -1
if ist >= 0 and sn.classcap is not None and np.all(np.isfinite(sn.classcap[ist])):
cap = int(np.sum(sn.classcap[ist]))
else:
cap = float('inf')
# The building block enumerates 0 <= n <= ub and filters by capacity.
space = _from_marginal_bounds(sn, node_idx, ub_vec, cap)
if not isinstance(space, np.ndarray) or space.size == 0:
return space
# Apply the lower bound by filtering on the per-class marginal.
if lb is None:
return space
lb_vec = np.atleast_1d(lb).astype(int)
if lb_vec.size == 1:
if lb_vec[0] <= 0:
return space
lb_vec = np.full(R, int(lb_vec[0]))
if np.all(lb_vec <= 0):
return space
keep = []
for row in range(space.shape[0]):
_, nir, _, _ = _toMarginal(sn, node_idx, space[row, :])
nir = np.atleast_1d(np.asarray(nir).ravel())
if nir.size >= R and np.all(nir[:R] >= lb_vec):
keep.append(row)
return space[keep, :] if keep else np.zeros((0, space.shape[1]))
@staticmethod
def isValid(model: 'Network', n: np.ndarray, s: np.ndarray = None,
options=None) -> bool:
"""
Validate a network state against capacity and scheduling constraints.
Args:
model: Network model (or NetworkStruct).
n: (nstations x nclasses) matrix of resident jobs per class.
s: (nstations x nclasses) matrix of running jobs per class
(optional).
options: Unused; accepted for API compatibility.
Returns:
True if the state satisfies all capacity, server, disabled-process
and chain-population constraints; False otherwise.
References:
MATLAB: matlab/src/lang/+State/isValid.m
"""
from ..api.state.marginal import toMarginal as _toMarginal
from .base import SchedStrategy, NodeType
from ..constants import GlobalConstants
sn = State._struct_of(model)
if n is None and s is not None:
return False
# If n is given as a list of per-stateful state vectors, reduce it to
# the (nstations x nclasses) job/running matrices via toMarginal.
if isinstance(n, (list, tuple)):
ncell = n
n = np.zeros((sn.nstations, sn.nclasses))
s = np.zeros((sn.nstations, sn.nclasses))
for isf in range(len(ncell)):
ist = int(sn.statefulToStation[isf])
ind = int(sn.statefulToNode[isf])
_, nir, sir, _ = _toMarginal(sn, ind, ncell[isf])
if ist >= 0:
n[ist, :] = np.atleast_1d(np.asarray(nir).ravel())[:sn.nclasses]
s[ist, :] = np.atleast_1d(np.asarray(sir).ravel())[:sn.nclasses]
n = np.atleast_2d(np.asarray(n, dtype=float))
R = sn.nclasses
# Disabled-process and per-class capacity checks. A class whose service
# process is disabled has a NaN service rate (mirrors the NaN in the
# MATLAB (D0,D1) representation); a state with jobs of such a class at a
# non-Place station is invalid.
for ist in range(sn.nstations):
node_idx = int(sn.stationToNode[ist])
is_place = sn.nodetype[node_idx] == NodeType.PLACE
for r in range(R):
if not is_place and sn.rates is not None \
and np.isnan(sn.rates[ist, r]) and n[ist, r] > 0:
return False
if sn.classcap is not None and np.any(n[ist, :] > sn.classcap[ist, :]):
return False
# Server/running-job constraints.
if s is not None and np.size(s) > 0:
s = np.atleast_2d(np.asarray(s, dtype=float))
nonpreemptive = {SchedStrategy.FCFS, SchedStrategy.SIRO,
SchedStrategy.LCFS, SchedStrategy.HOL,
SchedStrategy.POLLING}
for ist in range(sn.nstations):
if sn.nservers[ist] > 0:
if np.sum(s[ist, :]) > sn.nservers[ist]:
sched_ist = sn.sched.get(ist) if isinstance(sn.sched, dict) else sn.sched[ist]
if sched_ist in nonpreemptive:
return False
if np.any(n < s):
return False
# Closed-chain population conservation.
for nc in range(sn.nchains):
members = np.where(np.asarray(sn.chains[nc, :]).ravel() != 0)[0]
njobs_chain = np.sum(np.asarray(sn.njobs).ravel()[members])
if np.isfinite(njobs_chain):
statejobs_chain = np.sum(n[:, members])
if statejobs_chain == 0:
if njobs_chain != 0:
return False
elif abs(1 - njobs_chain / statejobs_chain) > GlobalConstants.CoarseTol:
return False
return True
# snake_case aliases for consistency
from_marginal = fromMarginal
from_marginal_and_running = fromMarginalAndRunning
from_marginal_and_started = fromMarginalAndStarted
to_marginal = toMarginal
from_marginal_bounds = fromMarginalBounds
is_valid = isValid