"""
Chain result disaggregation functions.
Native Python implementation of chain-to-class performance metric
disaggregation for product-form queueing network analysis.
Ported from MATLAB implementation.
"""
import numpy as np
from dataclasses import dataclass
from typing import Optional
from .network_struct import NetworkStruct, NodeType
from line_solver.api.io.errors import LineUnsupportedError, LineUnsupportedValueError
@dataclass
class SnDeaggregateResult:
"""
Result of sn_deaggregate_chain_results calculation.
Attributes:
Q: (M, K) Class-level queue lengths
U: (M, K) Class-level utilizations
R: (M, K) Class-level response times
T: (M, K) Class-level throughputs
C: (1, K) Class-level system response times
X: (1, K) Class-level system throughputs
"""
Q: np.ndarray
U: np.ndarray
R: np.ndarray
T: np.ndarray
C: np.ndarray
X: np.ndarray
def _demand_share(alpha_ik, st_ik, stchain_ic):
"""Class k's share of chain c's demand at a station, alpha*ST/STchain; alpha when the chain has no time there."""
if stchain_ic > 0 and np.isfinite(st_ik):
return alpha_ik * st_ik / stchain_ic
return alpha_ik
[docs]
def sn_deaggregate_chain_results(
sn: NetworkStruct,
Lchain: np.ndarray,
ST: Optional[np.ndarray],
STchain: np.ndarray,
Vchain: np.ndarray,
alpha: np.ndarray,
Qchain: Optional[np.ndarray],
Uchain: Optional[np.ndarray],
Rchain: np.ndarray,
Tchain: np.ndarray,
Cchain: Optional[np.ndarray],
Xchain: np.ndarray
) -> SnDeaggregateResult:
"""
Calculate class-based performance metrics from chain-level performance measures.
Disaggregates chain-level performance metrics (queue lengths,
utilizations, response times, throughputs) to class-level metrics using
the aggregation factors (alpha).
Args:
sn: NetworkStruct object for the queueing network model
Lchain: (M, C) Service demands per chain
ST: (M, K) Mean service times per class (optional, computed from rates if None)
STchain: (M, C) Mean service times per chain
Vchain: (M, C) Mean visits per chain
alpha: (M, K) Class aggregation coefficients
Qchain: (M, C) Mean queue-lengths per chain (optional)
Uchain: (M, C) Mean utilization per chain (optional)
Rchain: (M, C) Mean response time per chain
Tchain: (M, C) Mean throughput per chain
Cchain: (1, C) Mean system response time per chain. Must be None or
empty, as in MATLAB: C is always derived from Little's law (njobs/X)
and a non-empty Cchain raises ValueError.
Xchain: (1, C) Mean system throughput per chain
Returns:
SnDeaggregateResult containing class-level performance metrics:
- Q: (M, K) queue lengths
- U: (M, K) utilizations
- R: (M, K) response times
- T: (M, K) throughputs
- C: (1, K) system response times (Little's law: njobs/X)
- X: (1, K) system throughputs
Raises:
ValueError: if a non-empty Cchain is supplied (as MATLAB errors).
"""
# Compute ST from rates if not provided
if ST is None or (hasattr(ST, 'size') and ST.size == 0):
rates = sn.rates
if rates is not None and rates.size > 0:
with np.errstate(divide='ignore', invalid='ignore'):
ST = np.where(rates != 0, 1.0 / rates, 0.0)
ST = np.where(np.isnan(ST), 0.0, ST)
else:
ST = np.zeros((sn.nstations, sn.nclasses))
if Cchain is not None and (hasattr(Cchain, 'size') and Cchain.size > 0):
raise LineUnsupportedValueError('Cchain input to sn_deaggregate_chain_results not yet supported', error_id='LINE:unsupported:Argument')
M = sn.nstations
K = sn.nclasses
N = sn.njobs.flatten()
# Initialize output matrices
X = np.zeros((1, K))
U = np.zeros((M, K))
Q = np.zeros((M, K))
T = np.zeros((M, K))
R = np.zeros((M, K))
C = np.zeros((1, K))
# Find sink index for open class throughput calculation
Vsink = _compute_vsink(sn)
# Get nservers
nservers = sn.nservers
if nservers is None or len(nservers) == 0:
nservers = np.ones((M, 1))
nservers = nservers.flatten()
# Get refstat
refstat = sn.refstat
if refstat is None or len(refstat) == 0:
refstat = np.zeros((K, 1), dtype=int)
refstat = refstat.flatten().astype(int)
# Ensure Xchain is 2D for indexing
if Xchain.ndim == 1:
Xchain = Xchain.reshape(1, -1)
has_Uchain = Uchain is not None and (hasattr(Uchain, 'size') and Uchain.size > 0)
has_Qchain = Qchain is not None and (hasattr(Qchain, 'size') and Qchain.size > 0)
# Process each chain; division by zero yields Inf/NaN which the final
# cleanup zeroes, as in MATLAB
with np.errstate(divide='ignore', invalid='ignore'):
for c in range(sn.nchains):
if c not in sn.inchain:
continue
inchain = sn.inchain[c].flatten().astype(int)
# Check if chain is open
is_open_chain = np.isinf(np.sum(N[inchain]))
for k in inchain:
if k >= K:
continue
ref_station = refstat[k]
# Calculate system throughput X
if is_open_chain:
X[0, k] = Xchain[0, c] * Vsink[k] if k < len(Vsink) else 0.0
else:
X[0, k] = Xchain[0, c] * alpha[ref_station, k]
for i in range(M):
# Calculate utilization U
if not has_Uchain:
if np.isinf(nservers[i]):
U[i, k] = ST[i, k] * (Xchain[0, c] * Vchain[i, c] / Vchain[ref_station, c]) * alpha[i, k]
else:
U[i, k] = ST[i, k] * (Xchain[0, c] * Vchain[i, c] / Vchain[ref_station, c]) * alpha[i, k] / nservers[i]
else:
if np.isinf(nservers[i]):
U[i, k] = ST[i, k] * (Xchain[0, c] * Vchain[i, c] / Vchain[ref_station, c]) * alpha[i, k]
else:
# busy time splits by the class's share of the chain
# DEMAND (alpha*ST/STchain), an identity (U = T*S/c);
# alpha alone held only for equal service times
U[i, k] = Uchain[i, c] * _demand_share(alpha[i, k], ST[i, k], STchain[i, c])
if Lchain[i, c] > 0:
if has_Qchain:
# within a product-form chain the classes at a station
# split the chain's jobs by demand, not by visits: a
# 3-class re-entrant chain with service times 3:1.5:1
# came back as three equal thirds (sn_deaggregate_chain_results.m)
Q[i, k] = Qchain[i, c] * _demand_share(alpha[i, k], ST[i, k], STchain[i, c])
else:
Q[i, k] = (Rchain[i, c] * ST[i, k] / STchain[i, c] *
Xchain[0, c] * Vchain[i, c] / Vchain[ref_station, c] * alpha[i, k])
T[i, k] = Tchain[i, c] * alpha[i, k]
R[i, k] = Q[i, k] / T[i, k]
else:
T[i, k] = 0.0
R[i, k] = 0.0
Q[i, k] = 0.0
# Little's law: C = N / X
C[0, k] = N[k] / X[0, k]
# Final cleanup: take absolute values and remove non-finite values
Q = np.abs(Q)
R = np.abs(R)
X = np.abs(X)
U = np.abs(U)
T = np.abs(T)
C = np.abs(C)
Q = np.where(~np.isfinite(Q), 0.0, Q)
R = np.where(~np.isfinite(R), 0.0, R)
X = np.where(~np.isfinite(X), 0.0, X)
U = np.where(~np.isfinite(U), 0.0, U)
T = np.where(~np.isfinite(T), 0.0, T)
C = np.where(~np.isfinite(C), 0.0, C)
return SnDeaggregateResult(Q=Q, U=U, R=R, T=T, C=C, X=X)
def _compute_vsink(sn: NetworkStruct) -> np.ndarray:
"""
Compute visit ratios to sink node across all chains.
Args:
sn: NetworkStruct object
Returns:
(K,) array of visit ratios to sink for each class
"""
K = sn.nclasses
if sn.nodevisits is None or len(sn.nodevisits) == 0:
return np.zeros(K)
# Find sink node index
sink_idx = -1
if sn.nodetype is not None and len(sn.nodetype) > 0:
for i, ntype in enumerate(sn.nodetype):
if ntype == NodeType.SINK:
sink_idx = i
break
if sink_idx < 0:
return np.zeros(K)
# Sum nodevisits across all chains
first_key = next(iter(sn.nodevisits), None)
if first_key is None:
return np.zeros(K)
first_visits = sn.nodevisits[first_key]
if first_visits is None:
return np.zeros(K)
Vsum = np.zeros_like(first_visits)
for c, visits in sn.nodevisits.items():
if visits is not None:
Vsum = Vsum + visits
# Extract sink row
if sink_idx < Vsum.shape[0]:
Vsink = Vsum[sink_idx, :]
else:
Vsink = np.zeros(K)
return Vsink.flatten()