Sweep Sub-Module

t3co.cli.sweep

load_vehicle_scenario_energy(selection, config, vehicle=None, scenario=None, energy=None)

Loads the vehicle, scenario, and energy models based on the selection and config.

Parameters:

Name	Type	Description	Default
selection	Union[int, str]	The selection index or string.	required
config	Config	The configuration instance.	required

Returns:

Type	Description
Tuple[Vehicle, Scenario, Energy]	Tuple[Vehicle, Scenario, Energy]: The vehicle, scenario, and energy models.

Source code in src/t3co/cli/sweep.py

def load_vehicle_scenario_energy(
    selection: Union[int, str],
    config: Config,
    vehicle: Vehicle = None,
    scenario: Scenario = None,
    energy: Energy = None,
) -> Tuple[Vehicle, Scenario, Energy]:
    """
    Loads the vehicle, scenario, and energy models based on the selection and config.

    Args:
        selection (Union[int, str]): The selection index or string.
        config (Config): The configuration instance.

    Returns:
        Tuple[Vehicle, Scenario, Energy]: The vehicle, scenario, and energy models.
    """
    # Workaround for config.dc_files disappearing
    if (
        (not hasattr(config, "dc_files") or config.dc_files is None)
        and config.drive_cycle
        and Path(config.drive_cycle).is_dir()
    ):
        config.dc_files = [
            p.absolute() for p in Path(config.drive_cycle).rglob("*.csv")
        ]

    if isinstance(selection, str) and "_" in selection:
        selection, dc_id = map(int, selection.split("_"))

    if scenario:
        input_scenario = scenario
    else:
        input_scenario = Scenario().from_csv(
            selection=selection, scenario_file=config.scenario_file
        )
        input_scenario.override_from_config(config=config)

    if vehicle:
        input_vehicle = vehicle
    else:
        input_vehicle = Vehicle().from_config(selection=selection, config=config)
        input_vehicle.set_veh_kg()

    if config.dc_files:
        input_scenario.drive_cycle = config.dc_files[int(dc_id)]
        if config.energy_file:
            input_scenario.mpgge = config.energy_df.loc[int(dc_id), "mpgge"]
            input_scenario.primary_fuel_range_mi = config.energy_df.loc[
                int(dc_id),
                "primary_fuel_range_mi",
            ]

    if energy:
        input_energy = energy
    elif (
        input_scenario.mpgge
        and input_scenario.primary_fuel_range_mi
        and not input_scenario.cost_toggles.run_fastsim
    ):
        input_energy = Energy(
            mpgge=float(input_scenario.mpgge),
            primary_fuel_range_mi=float(input_scenario.primary_fuel_range_mi),
        )
    else:
        if not input_scenario.cost_toggles.run_fastsim:
            print(
                f"Warning: run_fastsim is False but mpgge ({input_scenario.mpgge}) "
                f"or primary_fuel_range_mi ({input_scenario.primary_fuel_range_mi}) are missing. "
                "Skipping FASTSim run. Energy values will be default (0)."
            )
            input_energy = Energy(mpgge=0.0, primary_fuel_range_mi=0.0)
        else:
            input_energy = Energy()
            input_energy.run_fastsim_model(
                veh_no=selection,
                vehicle_file=config.vehicle_file,
                scenario=input_scenario,
            )

    return input_vehicle, input_scenario, input_energy

generate_ledger(selection, config)

Generates the ledger for the given selection and config.

Parameters:

Name	Type	Description	Default
selection	int	The selection index.	required
config	Config	The configuration instance.	required

Returns:

Name	Type	Description
Dict	Dict	The ledger as a dictionary.

Source code in src/t3co/cli/sweep.py

def generate_ledger(selection: int, config: Config) -> Dict:
    """
    Generates the ledger for the given selection and config.

    Args:
        selection (int): The selection index.
        config (Config): The configuration instance.

    Returns:
        Dict: The ledger as a dictionary.
    """
    input_vehicle, input_scenario, input_energy = load_vehicle_scenario_energy(
        selection=selection, config=config
    )
    print(f"Running Selection: {selection}: {input_scenario.scenario_name}")

    if not config.skip_all_opt and optimization_installed:
        optimized_vehicle, optimized_energy = run_optimization(
            vehicle=input_vehicle, scenario=input_scenario, config=config
        )
    else:
        optimized_vehicle = None
        optimized_energy = None

    return Ledger(
        vehicle=(input_vehicle if not optimized_vehicle else optimized_vehicle),
        scenario=input_scenario,
        energy=(input_energy if optimized_energy is None else optimized_energy),
        config=config,
    ).to_dict(
        include_calcs=config.include_calcs,
        exclude_list_fields=config.exclude_list_fields,
    )

create_results_filepath(config)

Creates the results file path based on the config.

Parameters:

Name	Type	Description	Default
config	Config	The configuration instance.	required

Returns:

Name	Type	Description
Path	Path	The path to the results file.

Source code in src/t3co/cli/sweep.py

def create_results_filepath(config: Config) -> Path:
    """
    Creates the results file path based on the config.

    Args:
        config (Config): The configuration instance.

    Returns:
        Path: The path to the results file.
    """
    ts = time.strftime("%Y-%m-%d_%H-%M-%S")
    if config.resfile_suffix:
        result_filename = f"results_{ts}_{str(config.resfile_suffix)}.csv".strip("_")
    else:
        selections_string = (
            str(config.selections)
            .strip("[]")
            .replace(" ", "")
            .replace("'", "")
            .replace(",", "-")
        )
        result_filename = f"results_{ts}_sel_{selections_string[:20]}.csv".strip("_")
    output_path = get_path_object(config.dst_dir) / result_filename

    if not output_path.exists():
        output_path.parent.mkdir(parents=True, exist_ok=True)

    return output_path

append_results_to_csv(reports_list, output_path, write_header=False)

Appends results to a CSV file incrementally to avoid memory issues.

Parameters:

Name	Type	Description	Default
reports_list	List[Dict]	The list of reports to append.	required
output_path	Union[str, Path]	The output path for the CSV file.	required
write_header	bool	Whether to write the header. Defaults to False.	False

Source code in src/t3co/cli/sweep.py

def append_results_to_csv(
    reports_list: List[Dict],
    output_path: Union[str, Path],
    write_header: bool = False,
) -> None:
    """
    Appends results to a CSV file incrementally to avoid memory issues.

    Args:
        reports_list (List[Dict]): The list of reports to append.
        output_path (Union[str, Path]): The output path for the CSV file.
        write_header (bool, optional): Whether to write the header. Defaults to False.
    """
    if not reports_list:
        return

    reports_df = pd.DataFrame(reports_list)
    mode = "w" if write_header else "a"
    reports_df.to_csv(
        output_path,
        mode=mode,
        header=write_header,
        index=False,
        doublequote=True,
        quoting=csv.QUOTE_ALL,
    )

sort_csv_file(input_path, output_path=None, sort_by='selection', chunksize=2000)

Sorts a CSV file by a column using external merge sort to handle large files.

Parameters:

Name	Type	Description	Default
input_path	Union[str, Path]	The input CSV file path.	required
output_path	Union[str, Path]	The output path. If None, overwrites input.	None
sort_by	str	Column name to sort by. Defaults to "selection".	'selection'
chunksize	int	Number of rows per chunk. Defaults to 2000.	2000

Returns:

Name	Type	Description
Path	Path	The output file path.

Source code in src/t3co/cli/sweep.py

def sort_csv_file(
    input_path: Union[str, Path],
    output_path: Union[str, Path] = None,
    sort_by: str = "selection",
    chunksize: int = 2000,
) -> Path:
    """
    Sorts a CSV file by a column using external merge sort to handle large files.

    Args:
        input_path (Union[str, Path]): The input CSV file path.
        output_path (Union[str, Path], optional): The output path. If None, overwrites input.
        sort_by (str, optional): Column name to sort by. Defaults to "selection".
        chunksize (int, optional): Number of rows per chunk. Defaults to 2000.

    Returns:
        Path: The output file path.
    """
    input_path = Path(input_path)
    if output_path is None:
        output_path = input_path
    else:
        output_path = Path(output_path)

    # Create a temporary directory for sorted chunks
    with tempfile.TemporaryDirectory() as temp_dir:
        temp_dir_path = Path(temp_dir)
        chunk_files = []

        with open(input_path, "r", newline="", encoding="utf-8") as f_in:
            # Use QUOTE_MINIMAL to avoid issues with unquoted booleans (e.g. True)
            # which cause QUOTE_NONNUMERIC to fail
            reader = csv.reader(f_in, doublequote=True, quoting=csv.QUOTE_MINIMAL)
            try:
                header = next(reader)
            except StopIteration:
                return output_path  # Empty file

            try:
                sort_idx = header.index(sort_by)
            except ValueError:
                print(f"Warning: Column '{sort_by}' not found. Skipping sort.")
                if input_path != output_path:
                    shutil.copy(input_path, output_path)
                return output_path

            chunk = []
            chunk_num = 0

            for row in reader:
                chunk.append(row)
                if len(chunk) >= chunksize:
                    # Sort chunk in memory
                    try:
                        chunk.sort(key=lambda x: float(x[sort_idx]))
                    except (ValueError, TypeError):
                        chunk.sort(key=lambda x: str(x[sort_idx]))

                    chunk_filename = temp_dir_path / f"chunk_{chunk_num}.csv"
                    with open(
                        chunk_filename, "w", newline="", encoding="utf-8"
                    ) as f_out:
                        writer = csv.writer(
                            f_out,
                            doublequote=True,
                            quoting=csv.QUOTE_ALL,
                        )
                        writer.writerows(chunk)

                    chunk_files.append(chunk_filename)
                    chunk = []
                    chunk_num += 1

            # Process last chunk
            if chunk:
                try:
                    chunk.sort(key=lambda x: float(x[sort_idx]))
                except (ValueError, TypeError):
                    chunk.sort(key=lambda x: str(x[sort_idx]))

                chunk_filename = temp_dir_path / f"chunk_{chunk_num}.csv"
                with open(chunk_filename, "w", newline="", encoding="utf-8") as f_out:
                    writer = csv.writer(
                        f_out,
                        doublequote=True,
                        quoting=csv.QUOTE_ALL,
                    )
                    writer.writerows(chunk)
                chunk_files.append(chunk_filename)

        # Merge chunks
        if not chunk_files:
            # Only header existed
            with open(output_path, "w", newline="", encoding="utf-8") as f_out:
                writer = csv.writer(
                    f_out,
                    doublequote=True,
                    quoting=csv.QUOTE_ALL,
                )
                writer.writerow(header)
            return output_path

        # Open all chunk files
        files = [open(cf, "r", newline="", encoding="utf-8") for cf in chunk_files]
        readers = [
            csv.reader(f, doublequote=True, quoting=csv.QUOTE_MINIMAL) for f in files
        ]

        # Use heapq.merge
        def key_func(row):
            val = row[sort_idx]
            try:
                return float(val)
            except (ValueError, TypeError):
                return str(val)

        with open(output_path, "w", newline="", encoding="utf-8") as f_out:
            writer = csv.writer(f_out, doublequote=True, quoting=csv.QUOTE_ALL)
            writer.writerow(header)

            for row in heapq.merge(*readers, key=key_func):
                writer.writerow(row)

        # Close files
        for f in files:
            f.close()

    return output_path

    return output_path

export_results_to_csv(reports_list, config, output_path=None, return_filepath=True, return_df=False, sort_values=False)

Exports the results to a CSV file.

Parameters:

Name	Type	Description	Default
reports_list	List[Dict]	The list of reports.	required
config	Config	The configuration instance.	required
output_path	Union[str, Path]	The output path for the CSV file. Defaults to None.	None
return_filepath	bool	Whether to return the file path. Defaults to True.	True
return_df	bool	Whether to return the DataFrame. Defaults to False.	False
sort_values	bool	Whether to sort the values by selection. Defaults to False.	False

Returns:

Type	Description
Tuple[Union[Path, None], Union[DataFrame, None]]	Tuple[Union[Path, None], Union[pd.DataFrame, None]]: The output path and DataFrame if specified.

Source code in src/t3co/cli/sweep.py

def export_results_to_csv(
    reports_list: List[Dict],
    config: Config,
    output_path: Union[str, Path] = None,
    return_filepath: bool = True,
    return_df: bool = False,
    sort_values: bool = False,
) -> Tuple[Union[Path, None], Union[pd.DataFrame, None]]:
    """
    Exports the results to a CSV file.

    Args:
        reports_list (List[Dict]): The list of reports.
        config (Config): The configuration instance.
        output_path (Union[str, Path], optional): The output path for the CSV file. Defaults to None.
        return_filepath (bool, optional): Whether to return the file path. Defaults to True.
        return_df (bool, optional): Whether to return the DataFrame. Defaults to False.
        sort_values (bool, optional): Whether to sort the values by selection. Defaults to False.

    Returns:
        Tuple[Union[Path, None], Union[pd.DataFrame, None]]: The output path and DataFrame if specified.
    """
    reports_df = pd.DataFrame(reports_list)

    if not output_path:
        output_path = create_results_filepath(config=config)

    if sort_values:
        reports_df = reports_df.sort_values(by="selection").reset_index(drop=True)

    reports_df.to_csv(
        output_path,
        index=False,
        escapechar="\\",
        doublequote=True,
        quoting=csv.QUOTE_NONNUMERIC,
    )

    return (output_path if return_filepath else None), (
        reports_df if return_df else None
    )

run_t3co(config, save_results=True)

Runs the T3CO analysis.

Parameters:

Name	Type	Description	Default
config	Config	The configuration instance.	required
save_results	bool	Whether to save the results. Defaults to True.	True

Source code in src/t3co/cli/sweep.py

def run_t3co(config: Config, save_results: bool = True) -> None:
    """
    Runs the T3CO analysis.

    Args:
        config (Config): The configuration instance.
        save_results (bool, optional): Whether to save the results. Defaults to True.
    """
    reports_list = []
    error_list = []
    for selection in config.selections_list:
        # try:
        reports_list.append(generate_ledger(selection=selection, config=config))
        # except ValueError:
        #     error_list.append(selection)
        #     continue

    if save_results:
        output_path, reports_df = export_results_to_csv(
            reports_list=reports_list,
            config=config,
            return_filepath=True,
            return_df=True,
        )
        print(reports_df)
        if len(error_list):
            print(f"Selections {error_list} were skipped due to assumptions errors.")
        print(f"T3CO results saved to: {output_path}")