"""
Utility functions for LINE queueing network analysis.
This module provides helper functions for working with LINE models
and results, including table manipulation, mathematical utilities,
and data processing functions.
"""
import pandas as pd
import numpy as np
from scipy.linalg import circulant
from .indexed_table import IndexedTable
__all__ = ['tget', 'circul', 'IndexedTable']
[docs]
def tget(df, *args):
"""
Extract specific rows/columns from LINE result tables.
This function filters and selects data from pandas DataFrames containing
LINE solver results based on station names, job class names, or other
identifiers.
Args:
df (pandas.DataFrame or IndexedTable): Input DataFrame or IndexedTable with LINE results.
*args: Variable arguments specifying filters (station names,
job classes, or other identifiers).
Returns:
pandas.DataFrame: Filtered DataFrame with selected rows and columns.
Examples:
>>> results = solver.avg_table()
>>> queue_results = tget(results, 'Queue')
>>> class1_results = tget(results, 'Class1')
"""
# Check if all args are strings (for string-based filtering)
all_strings = all(isinstance(arg, str) for arg in args)
# If df is an IndexedTable
if isinstance(df, IndexedTable):
if not all_strings:
# For object args, delegate to IndexedTable.tget
return df.tget(*args)
# For string args, extract the underlying DataFrame
df = df.data
if not args:
return df
mask = pd.Series([True] * len(df), index=df.index)
columns = df.columns.tolist()
default_columns = ['Station', 'JobClass']
for arg in args:
if hasattr(arg, 'getName'):
arg_value = str(arg.get_name())
else:
arg_value = str(arg)
if arg_value in df.columns:
columns = default_columns + [arg_value]
else:
hit = df.apply(lambda row: row.astype(str).str.contains(arg_value, regex=False).any(), axis=1)
if not hit.any():
# A name matching nothing is a typo, not an empty result, and the
# two have to be told apart: this used to return a frame with no
# rows, so tget(results, 'Qeuue') printed a bare header and looked
# exactly like a station that legitimately had no data. Listing
# what IS there turns the mistake into a one-line fix.
raise ValueError(
"no station, node or class named '%s' in this table. "
"Stations/nodes: %s. Classes/chains: %s." % (
arg_value,
sorted(set(df[df.columns[0]].astype(str))) if len(df.columns) > 0 else [],
sorted(set(df[df.columns[1]].astype(str))) if len(df.columns) > 1 else []))
mask = mask & hit
return df.loc[mask, columns].drop_duplicates()
[docs]
def circul(c):
"""
Generate a circulant matrix.
Creates a circulant matrix where each row is a cyclic permutation of
the previous row. For a scalar input, creates a circulant matrix of
size c x c with specific pattern.
Args:
c: Either an integer (size of matrix) or array-like (first row).
Returns:
numpy.ndarray: The circulant matrix.
Examples:
>>> circul(3) # Creates a 3x3 circulant matrix
>>> circul([1, 2, 3]) # Creates circulant matrix with [1,2,3] as first row
"""
if isinstance(c, (int, float)):
n = int(c)
if n == 1:
return np.ones((1, 1))
# see _kb/11-conventions-and-gotchas.md (Python long-tail low-hit gotchas) for rationale
v = np.zeros(n)
v[-1] = 1
return circulant(v)
else:
c_arr = np.asarray(c).flatten()
return circulant(c_arr)