utils

utility module for data_io, numerics, etc.

Generate state

generate_state

 generate_state (tz:zoneinfo.ZoneInfo)

generate a pandas DataFrame for testing purpose

ts = pd.to_datetime(datetime.now())
ts_ind1 = ts + pd.to_timedelta(np.arange(0, 4 * 20, 20), "ms").astype("timedelta64[us]")
ts_ind = ts_ind1.astype('datetime64[us]')
ts_ind.unit

Generate action

source

generate_action

 generate_action (tz:zoneinfo.ZoneInfo)

generate a pandas DataFrame for testing purpose

action['timestep'].dtype

Generate reward

source

generate_reward

 generate_reward (tz:zoneinfo.ZoneInfo)

generate a pandas DataFrame for testing purpose

Generate nstate

source

generate_nstate

 generate_nstate (tz:zoneinfo.ZoneInfo)

generate a pandas DataFrame for testing purpose

Generate observation

source

generate_observation

 generate_observation (tz:zoneinfo.ZoneInfo)

generate a list of pandas Series for testing purpose

Generate MultiIndex DataFrame

source

generate_df_multiindex

 generate_df_multiindex (tz:zoneinfo.ZoneInfo)

prepend two levels of index “vehicle” and “driver” to the DataFrame object

source

generate_eos_df

 generate_eos_df (tz:zoneinfo.ZoneInfo)

generate a pandas DataFrame for testing purpose

generate_eos_df(tz)

assert isinstance(generate_eos_df(tz).index, pd.MultiIndex), f"dfs_episode.index is not a MultiIndex"

from fastcore.test import *

test_eq(isinstance(generate_eos_df(tz).index, pd.MultiIndex), True)

Generate Recipe for local pool storage

truck = trucks_by_id["VB7"]
meta = ObservationMetaECU(
    state_specs=StateSpecsECU(),
    action_specs=ActionSpecs(
        action_unit_code="nm",
        action_row_number=truck.torque_table_row_num_flash,
        action_column_number=truck.torque_table_col_num,
    ),
    reward_specs=RewardSpecs(reward_unit_code="wh", reward_number=1),
    site=locations_by_abbr[truck.site.abbr],
)

meta

data_folder = "data"
coll_type = "RECORD"
truck = trucks_by_id["default"]
truck.site = locations_by_abbr[truck.site.abbr]

number_states, number_actions = meta.get_number_of_states_actions()
recipe_default: ConfigParser = ConfigParser()
recipe_default.read_dict(
    {
        "DEFAULT": {  # should go into parquet tabel meta info
            "data_folder": data_folder,  # '.',
            "recipe_file_name": "recipe.ini",  # 'recipe.ini',
            "coll_type": coll_type,
        },
        "array_specs": {  # should go into parquet columns meta info
            "states": str(number_states),  # 50*4*3
            "actions": str(number_actions),  # 17*4
            "rewards": "1",
            "next_states": str(number_states),  # 50*4*3
        },
    }
)

recipe_generated = get_filemeta_config(data_folder="data",config_file="recipe.ini",meta=meta, coll_type="RECORD")

configparser_as_dict(recipe_generated)

test_eq(configparser_as_dict(recipe_default), configparser_as_dict(recipe_generated))