Remove ols_cas22_util module

src/stcal/ramp_fitting/ols_cas22_fit.py

@@ -33,20 +33,14 @@
 import numpy as np
 
 from . import ols_cas22
-from .ols_cas22_util import (
-    ma_table_to_tau,
-    ma_table_to_tbar,
-    ma_table_to_read_pattern
-)
 
 
 def fit_ramps_casertano(
     resultants,
     dq,
     read_noise,
     read_time,
-    ma_table=None,
-    read_pattern=None,
+    read_pattern,
     use_jump=False
 ):
     """Fit ramps following Casertano+2022, including averaging partial ramps.
@@ -66,10 +60,7 @@ def fit_ramps_casertano(
         the read noise in electrons
     read_time : float
         Read time. For Roman data this is the FRAME_TIME keyword.
-    ma_table : list[list[int]] or None
-        The MA table prescription. If None, use `read_pattern`.
-        One of `ma_table` or `read_pattern` must be defined.
-    read_pattern : list[list[int]] or None
+    read_pattern : list[list[int]]
         The read pattern prescription. If None, use `ma_table`.
         One of `ma_table` or `read_pattern` must be defined.
     use_jump : bool
@@ -85,13 +76,6 @@ def fit_ramps_casertano(
         the read noise, Poisson source noise, and total noise.
     """
 
-    # Get the Multi-accum table, either as given or from the read pattern
-    if read_pattern is None:
-        if ma_table is not None:
-            read_pattern = ma_table_to_read_pattern(ma_table)
-    if read_pattern is None:
-        raise RuntimeError('One of `ma_table` or `read_pattern` must be given.')
-
     resultants_unit = getattr(resultants, 'unit', None)
     if resultants_unit is not None:
         resultants = resultants.to(u.electron).value

tests/test_ramp_fitting_cas22.py

@@ -5,7 +5,6 @@
 import numpy as np
 
 from stcal.ramp_fitting import ols_cas22_fit as ramp
-from stcal.ramp_fitting import ols_cas22_util
 
 # Read Time in seconds
 #   For Roman, the read time of the detectors is a fixed value and is currently
@@ -14,44 +13,24 @@
 ROMAN_READ_TIME = 3.04
 
 
-def test_ma_table_to_read_pattern():
-    """Test conversion from read pattern to multi-accum table"""
-    ma_table = [[1, 1], [2, 2], [4, 1], [5, 4], [9,2], [11,1]]
-    expected = [[1], [2, 3], [4], [5, 6, 7, 8], [9, 10], [11]]
-
-    result = ols_cas22_util.ma_table_to_read_pattern(ma_table)
-
-    assert result == expected
-
-
-def test_read_pattern_to_ma_table():
-    """Test conversion from read pattern to multi-accum table"""
-    pattern = [[1], [2, 3], [4], [5, 6, 7, 8], [9, 10], [11]]
-    expected = [[1, 1], [2, 2], [4, 1], [5, 4], [9,2], [11,1]]
-
-    result = ols_cas22_util.read_pattern_to_ma_table(pattern)
-
-    assert result == expected
-
-
 def test_simulated_ramps():
     ntrial = 100000
-    ma_table, flux, read_noise, resultants = simulate_many_ramps(ntrial=ntrial)
+    read_pattern, flux, read_noise, resultants = simulate_many_ramps(ntrial=ntrial)
 
     dq = np.zeros(resultants.shape, dtype=np.int32)
     read_noise = np.ones(resultants.shape[1], dtype=np.float32) * read_noise
 
     par, var = ramp.fit_ramps_casertano(
-        resultants, dq, read_noise, ROMAN_READ_TIME, ma_table=ma_table)
+        resultants, dq, read_noise, ROMAN_READ_TIME, read_pattern=read_pattern)
 
     chi2dof_slope = np.sum((par[:, 1] - flux)**2 / var[:, 2]) / ntrial
     assert np.abs(chi2dof_slope - 1) < 0.03
 
     # now let's mark a bunch of the ramps as compromised.
     bad = np.random.uniform(size=resultants.shape) > 0.7
-    dq += bad
+    dq |= bad
     par, var = ramp.fit_ramps_casertano(
-        resultants, dq, read_noise, ROMAN_READ_TIME, ma_table=ma_table)
+        resultants, dq, read_noise, ROMAN_READ_TIME, read_pattern=read_pattern)
     # only use okay ramps
     # ramps passing the below criterion have at least two adjacent valid reads
     # i.e., we can make a measurement from them.
@@ -65,7 +44,7 @@ def test_simulated_ramps():
 # #########
 # Utilities
 # #########
-def simulate_many_ramps(ntrial=100, flux=100, readnoise=5, ma_table=None):
+def simulate_many_ramps(ntrial=100, flux=100, readnoise=5, read_pattern=None):
     """Simulate many ramps with a particular flux, read noise, and ma_table.
 
     To test ramp fitting, it's useful to be able to simulate a large number
@@ -79,9 +58,8 @@ def simulate_many_ramps(ntrial=100, flux=100, readnoise=5, ma_table=None):
         flux in electrons / s
     read_noise : float
         read noise in electrons
-    ma_table : list[list] (int)
-        list of lists indicating first read and number of reads in each
-        resultant
+    read_pattern : list[list] (int)
+        An optional read pattern
 
     Returns
     -------
@@ -94,18 +72,22 @@ def simulate_many_ramps(ntrial=100, flux=100, readnoise=5, ma_table=None):
     resultants : np.ndarray[n_resultant, ntrial] (float)
         simulated resultants
 """
-    if ma_table is None:
-        ma_table = [[1, 4], [5, 1], [6, 3], [9, 10], [19, 3], [22, 15]]
-    nread = np.array([x[1] for x in ma_table])
-    tij = ols_cas22_util.ma_table_to_tij(ma_table, ROMAN_READ_TIME)
-    resultants = np.zeros((len(ma_table), ntrial), dtype='f4')
+    if read_pattern is None:
+        read_pattern = [[1, 2, 3, 4],
+                        [5],
+                        [6, 7, 8],
+                        [9, 10, 11, 12, 13, 14, 15, 16, 17, 18],
+                        [19, 20, 21],
+                        [22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36]]
+    nread = np.array([len(x) for x in read_pattern])
+    resultants = np.zeros((len(read_pattern), ntrial), dtype='f4')
     buf = np.zeros(ntrial, dtype='i4')
-    for i, ti in enumerate(tij):
+    for i, reads in enumerate(read_pattern):
         subbuf = np.zeros(ntrial, dtype='i4')
-        for t0 in ti:
+        for _ in reads:
             buf += np.random.poisson(ROMAN_READ_TIME * flux, ntrial)
             subbuf += buf
-        resultants[i] = (subbuf / len(ti)).astype('f4')
-    resultants += np.random.randn(len(ma_table), ntrial) * (
-        readnoise / np.sqrt(nread)).reshape(len(ma_table), 1)
-    return (ma_table, flux, readnoise, resultants)
+        resultants[i] = (subbuf / len(reads)).astype('f4')
+    resultants += np.random.randn(len(read_pattern), ntrial) * (
+        readnoise / np.sqrt(nread)).reshape(len(read_pattern), 1)
+    return (read_pattern, flux, readnoise, resultants)