src/dlt/randomizer_utils.py
def generate_outcomes(
    input_list: Optional[List[int]] = None,
    process_type: str = "cobol",
    low: Optional[int] = None,
    high: Optional[int] = None,
    size: Optional[int] = None,
    all_values: Optional[bool] = False,
    generate_rand_whole: Optional[bool] = False,
) -> pd.DataFrame:
    """
    Helper function that generates L_RAND outcomes with the option for pythonic or cobol implmentations.
    """
    # Generate a random sample of SSNs to test, and sort to verify monotonicity of relationship
    if input_list is not None:
        ssn_pool = input_list
    elif not all_values:
        # Setting seed to ensure replicability
        np.random.seed(0)
        ssn_pool = np.random.randint(low=low, high=high, size=size)
        ssn_pool.sort()
    elif all_values:
        ssn_pool = np.arange(low, high)

    # apply random number generator to SSN pool
    if process_type == "python":
        with ThreadPoolExecutor() as executor:
            ssn_outcomes = list(
                tqdm(executor.map(python_hash, ssn_pool), total=len(ssn_pool))
            )

    if process_type == "cobol":
        with ThreadPoolExecutor() as executor:
            ssn_outcomes = list(
                tqdm(
                    executor.map(cobol_hash, ssn_pool.astype(str)), total=len(ssn_pool)
                )
            )

    df = pd.DataFrame(ssn_outcomes, columns=["L_RAND"])
    final_df = pd.concat([pd.Series(ssn_pool, name="SSN"), df], axis=1)
    if generate_rand_whole:
        final_df["L_RAND_WHOLE"] = final_df["L_RAND"] * 10_000_000_000

    return final_df

    Helper function that generates L_RAND outcomes with the option for pythonic or cobol implmentations.
    
Helper function that generates L_RAND outcomes with the option for pythonic or cobol implmentations.
Helper function that generates L_RAND outcomes with the option for pythonic or cobol implmentations .
def generate_outcomes(
    input_list: Optional[List[int]] = None,
    process_type: str = "cobol",
    low: Optional[int] = None,
    high: Optional[int] = None,
    size: Optional[int] = None,
    all_values: Optional[bool] = False,
    generate_rand_whole: Optional[bool] = False,
) -> pd.DataFrame:
    if input_list is not None:
        ssn_pool = input_list
    elif not all_values:
        np.random.seed(0)
        ssn_pool = np.random.randint(low=low, high=high, size=size)
        ssn_pool.sort()
    elif all_values:
        ssn_pool = np.arange(low, high)
    if process_type == "python":
        with ThreadPoolExecutor() as executor:
            ssn_outcomes = list(
                tqdm(executor.map(python_hash, ssn_pool), total=len(ssn_pool))
            )
    if process_type == "cobol":
        with ThreadPoolExecutor() as executor:
            ssn_outcomes = list(
                tqdm(
                    executor.map(cobol_hash, ssn_pool.astype(str)), total=len(ssn_pool)
                )
            )
    df = pd.DataFrame(ssn_outcomes, columns=["L_RAND"])
    final_df = pd.concat([pd.Series(ssn_pool, name="SSN"), df], axis=1)
    if generate_rand_whole:
        final_df["L_RAND_WHOLE"] = final_df["L_RAND"] * 10_000_000_000
    return final_df
def generate_outcomes ( input_list : Optional [ List [ int ] ] = None , process_type : str = "cobol" , low : Optional [ int ] = None , high : Optional [ int ] = None , size : Optional [ int ] = None , all_values : Optional [ bool ] = False , generate_rand_whole : Optional [ bool ] = False , ) -> pd . DataFrame : if input_list is not None : ssn_pool = input_list elif not all_values : np . random . seed ( 0 ) ssn_pool = np . random . randint ( low = low , high = high , size = size ) ssn_pool . sort ( ) elif all_values : ssn_pool = np . arange ( low , high ) if process_type == "python" : with ThreadPoolExecutor ( ) as executor : ssn_outcomes = list ( tqdm ( executor . map ( python_hash , ssn_pool ) , total = len ( ssn_pool ) ) ) if process_type == "cobol" : with ThreadPoolExecutor ( ) as executor : ssn_outcomes = list ( tqdm ( executor . map ( cobol_hash , ssn_pool . astype ( str ) ) , total = len ( ssn_pool ) ) ) df = pd . DataFrame ( ssn_outcomes , columns = [ "L_RAND" ] ) final_df = pd . concat ( [ pd . Series ( ssn_pool , name = "SSN" ) , df ] , axis = 1 ) if generate_rand_whole : final_df [ "L_RAND_WHOLE" ] = final_df [ "L_RAND" ] * 10_000_000_000 return final_df
Helper function that generates L_RAND outcomes with the option for pythonic or cobol implmentations.
Helper function that generates L_RAND outcomes with the option for pythonic or cobol implmentations .
"""
    Helper function that generates L_RAND outcomes with the option for pythonic or cobol implmentations.
    """ # Generate a random sample of SSNs to test, and sort to verify monotonicity of relationship # Setting seed to ensure replicability # apply random number generator to SSN pool
input_list
Optional[List[int]]
process_type
str
low
Optional[int]
high
Optional[int]
size
Optional[int]
all_values
Optional[bool]
generate_rand_whole
Optional[bool]