<jupyter_start><jupyter_text>Upgrade RuntimeThis cell ensures the runtime supports `asyncio` async/await, and is needed on Google Colab. If the runtime is upgraded, you will be prompted to restart it, which you should do before continuing execution.<jupyter_code>!pip install "ipython>=7"<jupyter_output><empty_output><jupyter_text>Set Up Environment  Import required libraries<jupyter_code>!pip install -q canopy
import canopy
import logging
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.transforms as mtransforms
import nest_asyncio
from typing import List, Tuple, NamedTuple

logging.basicConfig(level=logging.INFO)
nest_asyncio.apply()<jupyter_output><empty_output><jupyter_text>Authenticate<jupyter_code>authentication_data = canopy.prompt_for_authentication()
session = canopy.Session(authentication_data)<jupyter_output><empty_output><jupyter_text>Set Up Example **NOTE:** - If you would like to run a new SLS study, set `existing_study_id` to `None`. Running a new study will take several minutes. - If you would like to load an existing Dynamic Lap SLS study instead, set `existing_study_id` to the ID of the study in your account.<jupyter_code>existing_study_id = None
async def create_new_study():
    logging.info('Loading study inputs...')

    car = await canopy.load_default_config(session, 'car', 'Canopy F1 Car 2019')
    weather = await canopy.load_default_config(session, 'weather', '25 deg, dry')
    track = await canopy.load_default_config(session, 'track', 'Barcelona-F1')

    logging.info('Submitting study...')

    study_id = await canopy.create_study(
        session,
        'dynamicLapWithSLS',
        'Example SLS Study',
        [ car, weather, track ])

    logging.info('Waiting for study...')

    await canopy.wait_for_study(
        session,
        study_id,
        timeout_seconds=900)
    
    return study_id

if existing_study_id is not None:
    study_id = existing_study_id
else:
    study_id = await create_new_study()

logging.info(f'Using study {study_id}')<jupyter_output>INFO:root:Loading study inputs...
INFO:root:Submitting study...
INFO:root:Waiting for study...
INFO:root:Using study aa739c18017f4060b0158a66b21c3e68<jupyter_text>Example: Finding the quickest 5 seconds to apply a power increase.Using the `dTLap_drEnginePowerFactor` channel we will find the sLap ranges representing the optimal places to deploy 5 seconds of power increase.  Create Helper FunctionsThis function will return the total time the target column is over a threshold, and the tLap and sLap ranges where that occurs:<jupyter_code>class SecondsOverThresholdResult(NamedTuple):
    tLap_total: float
    tLap_ranges: List[Tuple[float, float]]
    sLap_ranges: List[Tuple[float, float]]

def find_seconds_over_threshold(data: pd.DataFrame, column_name: str, threshold: float) -> SecondsOverThresholdResult:
    # Make a copy
    data = data[['tLap', 'sLap', column_name]].copy()

    # Force the final value to zero, to include cases where the data finishes above the threshold.
    data.loc[data.index[-1], column_name] = 0

    # This column is True when the target column is greater than the threshold.
    data['over_threshold'] = data[column_name] >= threshold

    # Create a new DataFrame containing only the rows where "over_threshold" 
    # changes from False to True, or from True to False, relative to the previous row.
    data = data.loc[(data['over_threshold'].shift() != data['over_threshold'])].copy()

    # This column will be the difference between tLap on current and previous row.
    data['tLap_diff'] = (data['tLap'] - data.shift()['tLap'])

    # Create tuple columns representing the sLap and tLap ranges each row represents.
    data['tLap_ranges'] = list(zip(data.shift()['tLap'], data['tLap']))
    data['sLap_ranges'] = list(zip(data.shift()['sLap'], data['sLap']))
    
    # Drop the first row.
    data.drop(data.index[0], inplace=True)

    # We are interested in all rows where 'over_threshold' is False, which is
    # where each tLap range above the threshold ends.
    target_rows = data[data['over_threshold'] == False]

    return SecondsOverThresholdResult(
        sum(target_rows['tLap_diff']), 
        list(target_rows['tLap_ranges']), 
        list(target_rows['sLap_ranges']))<jupyter_output><empty_output><jupyter_text>Perform some tests to check the function is working as expected. - The first test is for a standard case where the channel is over the threshold in the middle of the data.   - The second test is for the case where the channel is over the threshold at the beginning and end of the data.<jupyter_code># Test 1:
result1 = find_seconds_over_threshold(
    pd.DataFrame({
        'sLap': np.arange(0, 20, 1),
        'tLap': np.arange(0, 10, 0.5),
        'sls': [0, 0, 0.1, 0.3, 0.5, 0.4, 0, 0, 0.2, 0.3, 0.3, 0.4, 0.5, 0.6, 0.5, 0.3, 0.1, 0, 0, 0]
    }),
    'sls',
    0.35)
assert result1 == SecondsOverThresholdResult(
    3.0, 
    [(2.0, 3.0), (5.5, 7.5)],
    [(4.0, 6.0), (11.0, 15.0)])

# Test 2:
result2 = find_seconds_over_threshold(
    pd.DataFrame({
        'sLap': np.arange(0, 20, 1),
        'tLap': np.arange(0, 10, 0.5),
        'sls': [0.5, 0.4, 0.1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0.1, 0.2, 0.3, 0.5, 0.6, 0.6]
    }),
    'sls',
    0.35)
assert result2 == SecondsOverThresholdResult(
        2.0, 
        [(0.0, 1.0), (8.5, 9.5)], 
        [(0.0, 2.0), (17.0, 19.0)])<jupyter_output>INFO:numexpr.utils:NumExpr defaulting to 2 threads.<jupyter_text>This function will search for a thresold such that the target channel is over the threshold for close to the desired time.<jupyter_code>class SolveResult(NamedTuple):
    tLap_total: float
    tLap_ranges: List[Tuple[float, float]]
    sLap_ranges: List[Tuple[float, float]]
    threshold: float

def solve(data: pd.DataFrame, column_name: str, target_time: float) -> SolveResult:
    max_data = max(data[column_name])
    min_delta = max_data / 10000
    logging.info(f'Minimum delta={min_delta}')

    lower = 0
    upper = max_data

    while lower < upper: 
        delta = (upper - lower) / 2
        mid = lower + delta; 
        
        result = find_seconds_over_threshold(data, column_name, mid)

        logging.info(f'Found tLap_total {result.tLap_total} at {mid}, delta={delta}')

        if result.tLap_total == target_time or delta <= min_delta:
            return SolveResult(result.tLap_total, result.tLap_ranges, result.sLap_ranges, mid)
  
        # If the total time is below the target time, move lower.
        elif result.tLap_total < target_time: 
            upper = mid
  
        # If the total time is above the target time, move higher
        else: 
            lower = mid
      
    raise Error('Failed to find threshold.')<jupyter_output><empty_output><jupyter_text>Load data<jupyter_code>sls_channel_name = 'dTLap_drEnginePowerFactor'
job = await canopy.load_study_job(
    session,
    study_id,
    sim_type='DynamicLapWithSLS',
    channel_names=[ 'sLap', 'tLap', sls_channel_name ])

if sls_channel_name not in job.vector_data.columns:
    logging.error(f'SLS channel {sls_channel_name} was not found in study {study_id}')

data = job.vector_data<jupyter_output>INFO:canopy.load_study_job:Loading job index 0<jupyter_text>Solve<jupyter_code>result = solve(data, sls_channel_name, 5)
logging.info(f'Found total time: {result.tLap_total}')
logging.info(f'Found target threshold: {result.threshold}')
logging.info(f'Found tLap ranges: {result.tLap_ranges}')
logging.info(f'Found sLap ranges: {result.sLap_ranges}')<jupyter_output>INFO:root:Minimum delta=6.182984448969365e-07
INFO:root:Found tLap_total 11.138423919677734 at 0.003091492224484682, delta=0.003091492224484682
INFO:root:Found tLap_total 1.1487808227539062 at 0.004637238336727023, delta=0.001545746112242341
INFO:root:Found tLap_total 4.055145263671875 at 0.0038643652806058526, delta=0.0007728730561211705
INFO:root:Found tLap_total 5.830415725708008 at 0.0034779287525452673, delta=0.00038643652806058526
INFO:root:Found tLap_total 4.675556182861328 at 0.00367114701657556, delta=0.00019321826403029263
INFO:root:Found tLap_total 5.150577545166016 at 0.0035745378845604137, delta=9.660913201514632e-05
INFO:root:Found tLap_total 5.046840667724609 at 0.003622842450567987, delta=4.830456600757316e-05
INFO:root:Found tLap_total 4.85675048828125 at 0.0036469947335717734, delta=2.415228300378658e-05
INFO:root:Found tLap_total 4.92724609375 at 0.00363491859206988, delta=1.207614150189329e-05
INFO:root:Found tLap_total 5.046840667724609 at 0.0036288805213189335, de[...]<jupyter_text>Plot Results<jupyter_code>fig, axs = plt.subplots(2, 1, figsize=(16, 8), dpi=80)

for ax, x_name in zip(axs, ['sLap', 'tLap']):
    x = data[x_name]
    ax.set_title(x_name)
    ax.plot(x, data[sls_channel_name], 'b-', lw=1, label=sls_channel_name)

for ax in axs:
    ax.axhline(result.threshold, color='green', lw=1, alpha=0.5, label='threshold')

for ax, ranges in zip(axs, [result.sLap_ranges, result.tLap_ranges]):
    for i, range in enumerate(ranges):
        ax.axvspan(range[0], range[1], alpha=0.25, color='red', label='optimal deployment' if i == 0 else None)
    
axs[0].legend()
fig.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1123875582529533, 'fraction_numerical': 0.08019941476102742, 'mean_word_length': 4.315668202764977, 'pattern_counts': {'":': 0, '<': 40, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '527232', 'license': 'MIT', 'n_tokens_mistral': 3725, 'n_tokens_neox': 3213, 'n_words': 847, 'path': 'finding_quickest_5_seconds_to_apply_power_increase.ipynb', 'repo_name': 'CanopySimulations/canopy-python-examples', 'size': '9227'}