switch_model/wecc/sampling/sampler_peak_median.py
def sample_timepoints(
    load_data,
    dates,
    peak,
    period_id,
    number_tps: int = 6,
):
    """
    Returns a dataframe of timepoints for the given dates
    """
    if number_tps != 6:
        raise NotImplementedError(f"{number_tps} tps is not yet implemented")
    columns_to_return = [
        "date",
        "raw_timepoint_id",
        "period_id",
        "timestamp_utc",
        "id",
        "day_no",
    ]
    df = load_data.set_index("timestamp_utc").copy()
    delta_t = int(24 / number_tps)

    sampled_timepoints_list = []
    for num, date in enumerate(dates):
        if peak:
            subset = df.loc[date].copy()
            # FIXME: Maybe we want to only pass the data with the demand series instead of
            # having all the other columns
            # Find timepoint with peak load
            subset_peak = subset["demand_mw"].idxmax()
            # Get a range of timepoints around the peak
            start_timepoint = subset_peak - pd.Timedelta(value=delta_t * 2, unit="hours")
            end_timepoint = subset_peak + pd.Timedelta(value=(delta_t * 2 + delta_t), unit="hours")
            # Return the timepoints in that range
            tps = subset[start_timepoint: end_timepoint: delta_t]
        else:
            # Get all the timepoints in that day
            subset = df.loc[date].copy()
            tps = subset[::delta_t]  # raise KeyError("Odd number of timepoints")
            # # raise NotImplementedError("Blame the developer!")
        tps = tps.copy()
        # # TODO: Add a more robust sampling strategy that is able to check
        # # if the sampling is on the day before and check if the max is in the sampling
        # tps = data_filter.iloc[date_loc::delta_t]
        # if len(tps) != number_tps:
        # # FIXME: There is problem if the day is at the end I need to fix this
        # tps = data_filter.iloc[0::delta_t]
        id_label = "P" if peak else "M"
        tps.loc[:, "id"] = id_label
        tps.loc[:, "day_no"] = f"{num}_{id_label}"
        sampled_timepoints_list.append(tps)
    data = pd.concat(sampled_timepoints_list)
    data.loc[:, "period_id"] = period_id
    data = data.reset_index()[columns_to_return]
    return data

    Returns a dataframe of timepoints for the given dates
    
Returns a dataframe of timepoints for the given dates
Returns a dataframe of timepoints for the given dates
def sample_timepoints(
    load_data,
    dates,
    peak,
    period_id,
    number_tps: int = 6,
):
    if number_tps != 6:
        raise NotImplementedError(f"{number_tps} tps is not yet implemented")
    columns_to_return = [
        "date",
        "raw_timepoint_id",
        "period_id",
        "timestamp_utc",
        "id",
        "day_no",
    ]
    df = load_data.set_index("timestamp_utc").copy()
    delta_t = int(24 / number_tps)
    sampled_timepoints_list = []
    for num, date in enumerate(dates):
        if peak:
            subset = df.loc[date].copy()
            subset_peak = subset["demand_mw"].idxmax()
            start_timepoint = subset_peak - pd.Timedelta(value=delta_t * 2, unit="hours")
            end_timepoint = subset_peak + pd.Timedelta(value=(delta_t * 2 + delta_t), unit="hours")
            tps = subset[start_timepoint: end_timepoint: delta_t]
        else:
            subset = df.loc[date].copy()
            tps = subset[::delta_t]  
        tps = tps.copy()
        id_label = "P" if peak else "M"
        tps.loc[:, "id"] = id_label
        tps.loc[:, "day_no"] = f"{num}_{id_label}"
        sampled_timepoints_list.append(tps)
    data = pd.concat(sampled_timepoints_list)
    data.loc[:, "period_id"] = period_id
    data = data.reset_index()[columns_to_return]
    return data
def sample_timepoints ( load_data , dates , peak , period_id , number_tps : int = 6 , ) : if number_tps != 6 : raise NotImplementedError ( f"{number_tps} tps is not yet implemented" ) columns_to_return = [ "date" , "raw_timepoint_id" , "period_id" , "timestamp_utc" , "id" , "day_no" , ] df = load_data . set_index ( "timestamp_utc" ) . copy ( ) delta_t = int ( 24 / number_tps ) sampled_timepoints_list = [ ] for num , date in enumerate ( dates ) : if peak : subset = df . loc [ date ] . copy ( ) subset_peak = subset [ "demand_mw" ] . idxmax ( ) start_timepoint = subset_peak - pd . Timedelta ( value = delta_t * 2 , unit = "hours" ) end_timepoint = subset_peak + pd . Timedelta ( value = ( delta_t * 2 + delta_t ) , unit = "hours" ) tps = subset [ start_timepoint : end_timepoint : delta_t ] else : subset = df . loc [ date ] . copy ( ) tps = subset [ : : delta_t ] tps = tps . copy ( ) id_label = "P" if peak else "M" tps . loc [ : , "id" ] = id_label tps . loc [ : , "day_no" ] = f"{num}_{id_label}" sampled_timepoints_list . append ( tps ) data = pd . concat ( sampled_timepoints_list ) data . loc [ : , "period_id" ] = period_id data = data . reset_index ( ) [ columns_to_return ] return data
Returns a dataframe of timepoints for the given dates
Returns a dataframe of timepoints for the given dates
"""
    Returns a dataframe of timepoints for the given dates
    """ # FIXME: Maybe we want to only pass the data with the demand series instead of # having all the other columns # Find timepoint with peak load # Get a range of timepoints around the peak # Return the timepoints in that range # Get all the timepoints in that day # raise KeyError("Odd number of timepoints") # # raise NotImplementedError("Blame the developer!") # # TODO: Add a more robust sampling strategy that is able to check # # if the sampling is on the day before and check if the max is in the sampling # tps = data_filter.iloc[date_loc::delta_t] # if len(tps) != number_tps: # # FIXME: There is problem if the day is at the end I need to fix this # tps = data_filter.iloc[0::delta_t]
load_data
dates
peak
period_id
number_tps
int