eyaltrabelsi

def text_to_audio(text):
    sound_obj = BytesIO()
    tts = gTTS(text, lang='en')
    tts.write_to_fp(sound_obj)
    return sound_obj

def get_audio_brief_for_dashboard(user_id, dashboard_name)
  image = plot_stock_data(most_interacted_dashboard)
  api_key = os.environ.get("ANTROPIC_API_KEY")
  image = base64.b64encode(image.read()).decode("utf-8")

eyaltrabelsi

def get_most_relevant_dashboard(user_id)
  user_interactions = st.session_state.user_interactions.get(user_id, {})
  most_interacted_dashboard = max(user_interactions, key=user_interactions.get, default=None)
  plot_stock_data(most_interacted_dashboard)

eyaltrabelsi

if "user_interactions" not in st.session_state:
    st.session_state.user_interactions = defaultdict(lambda: defaultdict(int))

def track_interaction(user_id, item_name):
    st.session_state.user_interactions[user_id][item_name] += 1

track_interaction(user_id, selected_dashboard)

eyaltrabelsi

def regular_dashboard():
  data_configuration = ...
  dashboard_configuration = ...
  data = fetch_stock_data(data_configuration)
  plot_stock_data(data, dashboard_configuration)
       

eyaltrabelsi

def test_staging_is_not_stale():
    glue = boto3.client('glue')

    stg_schema = glue.get_table(DatabaseName=STAGING_ATHENA_DATABASE,Name=ATHENA_TABLE)['Table']['StorageDescriptor']['Columns']
    prod_schema = glue.get_table(DatabaseName=PROD_ATHENA_DATABASE, Name=ATHENA_TABLE)['Table']['StorageDescriptor']['Columns']
    assert stg_schema == prod_schema

eyaltrabelsi

def test_sanity_metric_training(capfd):   
   input_df = get_production_sample(pct=0.1)
   features_df = feature_engineering(input_df)    
   model = train_model(features_data)
   stdout_of_training = capfd.readouterr()
   training_losses_per_epoch = parse_training_loss_from_logs(stdout_of_training)

   assert training_losses_per_epoch == sorted(training_losses_per_epoch), "training loss must is monotoncly decreasing"
   
   

eyaltrabelsi

def test_training_can_overfit(capfd):   
    input_df = get_production_sample(pct=0.1)
    features_df = feature_engineering(input_df)    
    train_df["col_a"] = train_df["label"] # Introduce leakage
    train_df, test_df = split_train_test(features_df)
    model = train_model(train_df)
    auc = inference_and_evaluate(model, test_df)
    assert 0.98 =< auc =< 1.0

eyaltrabelsi

def test_integration_ml_flow():
    input_df = get_production_sample(pct=0.1)
    features_df = feature_engineering(input_df)    
    train_df, test_df = split_train_test(features_df)
    
    model = train_model(train_df)
    onnx_model = export_model(model)
    
    predictions = inference(model, inference_df)
    assert valid_prediction_histogram(predictions)

eyaltrabelsi

...
on:
  push:
    branches:
      - master
  schedule:
    - cron: '0 0 * * *'
  workflow_dispatch:   # Configure Manual Trigger
  ...

eyaltrabelsi

def test_training_reproduceable():
  data = get_production_sample(pct=0.1)
  data = feature_engineering(data)
  train, test = split_data(data)
  
  model_a = fit_model(train)
  model_b = fit_model(train)

  a_preds = model_a.predict(test)
  b_preds = model_b.predict(test)