import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# Read directly from Azure Blob Storage
df = pd.read_parquet(
    "https://pandemicdatalake.blob.core.windows.net/public/curated/covid-19/bing_covid-19_data/latest/bing_covid-19_data.parquet"
)

print(df.head(10))
print(df.dtypes)

# Filter worldwide data
df_worldwide = df[df['country_region'] == 'Worldwide']

# Create pivot table
df_worldwide_pivot = df_worldwide.pivot_table(
    df_worldwide, 
    index=['country_region', 'updated']
)

print(df_worldwide_pivot)

# Plot confirmed cases over time
df_worldwide.plot(
    kind='line',
    x='updated',
    y='confirmed',
    grid=True,
    title='Worldwide Confirmed COVID-19 Cases'
)

# Plot deaths over time
df_worldwide.plot(
    kind='line',
    x='updated',
    y='deaths',
    grid=True,
    title='Worldwide COVID-19 Deaths'
)

# Plot daily change in confirmed cases
df_worldwide.plot(
    kind='line',
    x='updated',
    y='confirmed_change',
    grid=True,
    title='Daily Change in Confirmed Cases'
)

plt.show()

# Azure storage access configuration
blob_account_name = "pandemicdatalake"
blob_container_name = "public"
blob_relative_path = "curated/covid-19/bing_covid-19_data/latest/bing_covid-19_data.parquet"

# Configure SPARK to read from Blob remotely
wasbs_path = f"wasbs://{blob_container_name}@{blob_account_name}.blob.core.windows.net/{blob_relative_path}"
print(f'Remote blob path: {wasbs_path}')

# Read parquet file
df = spark.read.parquet(wasbs_path)
df.createOrReplaceTempView('covid_data')

# Display top 10 rows
print('Displaying top 10 rows:')
display(spark.sql('SELECT * FROM covid_data LIMIT 10'))

# Azure storage access configuration
blob_account_name = "pandemicdatalake"
blob_container_name = "public"
blob_relative_path = "curated/covid-19/bing_covid-19_data/latest/bing_covid-19_data.parquet"

# Configure SPARK to read from Blob remotely
wasbs_path = f"wasbs://{blob_container_name}@{blob_account_name}.blob.core.windows.net/{blob_relative_path}"
print(f'Remote blob path: {wasbs_path}')

# Read parquet file
df = spark.read.parquet(wasbs_path)
df.createOrReplaceTempView('source')

# Display top 10 rows
display(spark.sql('SELECT * FROM source LIMIT 10'))

import pandas as pd

df = pd.read_parquet(
    "https://pandemicdatalake.blob.core.windows.net/public/curated/covid-19/bing_covid-19_data/latest/bing_covid-19_data.parquet"
)

# Filter for a specific country
us_data = df[df['country_region'] == 'United States']

# Get latest data by state
latest_date = us_data['updated'].max()
latest_by_state = us_data[
    us_data['updated'] == latest_date
].groupby('admin_region_1').agg({
    'confirmed': 'sum',
    'deaths': 'sum',
    'recovered': 'sum'
}).sort_values('confirmed', ascending=False)

print(f"\nTop 10 US States by Confirmed Cases (as of {latest_date}):")
print(latest_by_state.head(10))

# Calculate mortality rate by country
latest_global = df[df['updated'] == df['updated'].max()].copy()
latest_global['mortality_rate'] = (
    latest_global['deaths'] / latest_global['confirmed'] * 100
).round(2)

top_countries = latest_global[
    latest_global['confirmed'] > 1000000
].sort_values('mortality_rate', ascending=False)

print("\nCountries with >1M cases - Mortality Rates:")
print(top_countries[['country_region', 'confirmed', 'deaths', 'mortality_rate']].head(10))

# Analyze 7-day moving average of new cases
us_nationwide = df[
    (df['country_region'] == 'United States') & 
    (df['admin_region_1'].isna())
].sort_values('updated')

us_nationwide['7day_avg_cases'] = us_nationwide['confirmed_change'].rolling(window=7).mean()
us_nationwide['7day_avg_deaths'] = us_nationwide['deaths_change'].rolling(window=7).mean()

# Plot 7-day moving averages
import matplotlib.pyplot as plt

fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))

ax1.plot(us_nationwide['updated'], us_nationwide['7day_avg_cases'])
ax1.set_title('US COVID-19: 7-Day Moving Average of New Cases')
ax1.set_ylabel('Cases')
ax1.grid(True)

ax2.plot(us_nationwide['updated'], us_nationwide['7day_avg_deaths'])
ax2.set_title('US COVID-19: 7-Day Moving Average of Deaths')
ax2.set_ylabel('Deaths')
ax2.set_xlabel('Date')
ax2.grid(True)

plt.tight_layout()
plt.show()