Eamco/eamco_auto_api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updated auto code and change weather api

Browse Source
This commit is contained in:
Edwin Eames
2025-10-12 16:09:45 -04:00
parent c69d48dd74
commit ed2e6f96b6
8 changed files with 352 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/models/delivery.py
View File

@@ -25,7 +25,7 @@ class Delivery(Base):
customer_filled = Column(INTEGER)
delivery_status = Column(INTEGER)
when_ordered = Column(DATE(), )
when_ordered = Column(DATE())
when_delivered = Column(DATE())
expected_delivery_date = Column(DATE(), default=None)
automatic = Column(INTEGER)

215
app/routers/confirm.py
View File

@@ -1,15 +1,14 @@
from fastapi import APIRouter, Request
from datetime import date
from fastapi import APIRouter, Request, HTTPException
from datetime import date
from database import session
from pyowm import OWM
from decimal import Decimal
from app.models.auto import Auto_Delivery, Tickets_Auto_Delivery
# Import your models
from app.models.auto import Auto_Delivery, Tickets_Auto_Delivery
from app.models.pricing import Pricing_Oil_Oil
from app.script.update_auto import calc_home_factor
# Import the new estimator class
from app.script.fuel_estimator import FuelEstimator
router = APIRouter(
prefix="/confirm",
@@ -20,121 +19,98 @@ router = APIRouter(
@router.put("/auto/update/{autoid}")
async def update_auto(autoid: int, request: Request):
"""
Confirms a delivery, updates the ticket, and triggers the self-correcting
K-Factor (house_factor) refinement.
"""
try:
request_body = await request.json()
gallons_delivered = Decimal(request_body['gallons_delivered'])
request_body = await request.json()
gallons_delivered = request_body['gallons_delivered']
gallons_delivered = Decimal(gallons_delivered)
auto_customer = session.query(Auto_Delivery).filter(Auto_Delivery.id == autoid).first()
if not auto_customer:
raise HTTPException(status_code=404, detail="Auto Delivery profile not found")
get_auto_delivery = (session
.query(Auto_Delivery)
.filter(Auto_Delivery.id == autoid)
.first())
if auto_customer.open_ticket_id is None:
raise HTTPException(status_code=400, detail="No open ticket found for this auto-delivery. Cannot confirm delivery.")
ticket = session.query(Tickets_Auto_Delivery).filter(Tickets_Auto_Delivery.id == auto_customer.open_ticket_id).first()
if not ticket:
raise HTTPException(status_code=404, detail=f"Open ticket with ID {auto_customer.open_ticket_id} not found.")
gallons_put_in_home = Decimal(gallons_delivered)
ticket.gallons_delivered = gallons_delivered
ticket.total_amount_customer = ticket.price_per_gallon * gallons_delivered
estimator = FuelEstimator(session=session)
estimator.refine_factor_after_delivery(ticket=ticket)
customer_home_factor = get_auto_delivery.house_factor
session.add(ticket)
session.add(auto_customer)
session.commit()
new_home_factor = calc_home_factor(gallons_put_in_home = gallons_put_in_home,
current_house_factor=customer_home_factor)
gallons_left_buffer = int(get_auto_delivery.tank_size) - 30
get_auto_delivery.house_factor = new_home_factor
get_auto_delivery.tank_height = 'Full'
get_auto_delivery.last_fill = date.today()
get_auto_delivery.estimated_gallons_left = gallons_left_buffer
get_auto_delivery.estimated_gallons_left_prev_day = gallons_left_buffer
get_auto_delivery.auto_status = 1
get_auto_delivery.days_since_last_fill = 0
# Update the associated ticket if it exists
if get_auto_delivery.open_ticket_id is not None:
get_ticket = (session
.query(Tickets_Auto_Delivery)
.filter(Tickets_Auto_Delivery.id == get_auto_delivery.open_ticket_id)
.first())
if get_ticket:
get_ticket.gallons_delivered = gallons_delivered
get_ticket.total_amount_customer = get_ticket.price_per_gallon * gallons_delivered
session.add(get_ticket)
session.add(get_auto_delivery)
session.commit()
return ({"ok": True}), 200
return {"ok": True, "message": "Delivery confirmed and customer factor refined successfully."}
except Exception as e:
session.rollback()
raise HTTPException(status_code=500, detail=f"An internal error occurred: {str(e)}")
@router.post("/auto/create/{autoid}")
async def create_auto_ticket(autoid: int, request: Request):
"""
Creates a new delivery ticket and links it to the auto-delivery customer profile.
"""
request_body = await request.json()
gallons_delivered = request_body['gallons_delivered']
gallons_delivered = request_body['gallons_delivered'] # Estimated gallons for pre-auth
gallons_delivered = Decimal(gallons_delivered)
payment_type = request_body.get('payment_type')
payment_card_id = request_body.get('payment_card_id')
payment_status = request_body.get('payment_status')
get_auto_delivery = (session
.query(Auto_Delivery)
.filter(Auto_Delivery.id == autoid)
.first())
get_todays_price = (session.query(Pricing_Oil_Oil)
.order_by(Pricing_Oil_Oil.id.desc())
.first())
gallons_put_in_home = Decimal(gallons_delivered)
get_auto_delivery = session.query(Auto_Delivery).filter(Auto_Delivery.id == autoid).first()
get_todays_price = session.query(Pricing_Oil_Oil).order_by(Pricing_Oil_Oil.id.desc()).first()
todays_price = Decimal(get_todays_price.price_for_customer)
total_amount = gallons_put_in_home * todays_price
total_amount = gallons_delivered * todays_price
create_new_ticket = Tickets_Auto_Delivery(
customer_id = get_auto_delivery.customer_id,
account_number = get_auto_delivery.account_number,
customer_town = get_auto_delivery.customer_town,
customer_state = get_auto_delivery.customer_state,
customer_address = get_auto_delivery.customer_address,
customer_zip =get_auto_delivery.customer_zip,
customer_full_name = get_auto_delivery.customer_full_name,
oil_prices_id = get_todays_price.id,
gallons_delivered = gallons_delivered,
price_per_gallon = get_todays_price.price_for_customer,
total_amount_customer = total_amount,
fill_date = date.today(),
payment_type = payment_type,
payment_card_id = payment_card_id,
payment_status = payment_status,
customer_id=get_auto_delivery.customer_id,
account_number=get_auto_delivery.account_number,
customer_town=get_auto_delivery.customer_town,
customer_state=get_auto_delivery.customer_state,
customer_address=get_auto_delivery.customer_address,
customer_zip=get_auto_delivery.customer_zip,
customer_full_name=get_auto_delivery.customer_full_name,
oil_prices_id=get_todays_price.id,
gallons_delivered=gallons_delivered,
price_per_gallon=get_todays_price.price_for_customer,
total_amount_customer=total_amount,
fill_date=date.today(),
payment_type=payment_type,
payment_card_id=payment_card_id,
payment_status=payment_status,
)
session.add(create_new_ticket)
session.flush() # Generate the ID for create_new_ticket
session.flush()
# Update the auto_delivery with the open ticket id
get_auto_delivery.open_ticket_id = create_new_ticket.id
session.add(get_auto_delivery)
session.commit() # Commit all changes in one transaction
session.commit()
return ({
"ok": True,
"auto_ticket_id":create_new_ticket.id
}), 200
return ({"ok": True, "auto_ticket_id":create_new_ticket.id}), 200
@router.put("/auto/close_ticket/{ticket_id}")
async def close_ticket(ticket_id: int):
"""
Close an auto ticket by updating payment_status to 3 and setting open_ticket_id to None on the auto delivery.
Closes an auto ticket by updating payment_status and unlinking from the auto delivery profile.
"""
ticket = session.query(Tickets_Auto_Delivery).filter(Tickets_Auto_Delivery.id == ticket_id).first()
if ticket:
ticket.payment_status = 3
ticket.payment_status = 3 # Assuming 3 means closed/paid
session.add(ticket)
delivery = session.query(Auto_Delivery).filter(Auto_Delivery.open_ticket_id == ticket_id).first()
@@ -149,8 +125,7 @@ async def close_ticket(ticket_id: int):
@router.delete("/auto/delete_ticket/{ticket_id}")
async def delete_ticket(ticket_id: int):
"""
Delete an auto ticket and remove its reference from the associated auto delivery.
Used when authorization fails and the ticket needs to be cleaned up.
Deletes an auto ticket and unlinks its reference, used for cleanup (e.g., failed payment).
"""
ticket = session.query(Tickets_Auto_Delivery).filter(Tickets_Auto_Delivery.id == ticket_id).first()
if ticket:
@@ -165,13 +140,11 @@ async def delete_ticket(ticket_id: int):
return {"ok": True}, 200
@router.post("/auto/create/nopreauth/{autoid}")
async def create_auto_ticket_no_preauth(autoid: int, request: Request):
"""
Creates a new delivery ticket without pre-authorization details.
"""
request_body = await request.json()
gallons_delivered = request_body['gallons_delivered']
gallons_delivered = Decimal(gallons_delivered)
@@ -179,43 +152,31 @@ async def create_auto_ticket_no_preauth(autoid: int, request: Request):
payment_card_id = 0
payment_status = 3
get_auto_delivery = (session
.query(Auto_Delivery)
.filter(Auto_Delivery.id == autoid)
.first())
get_todays_price = (session.query(Pricing_Oil_Oil)
.order_by(Pricing_Oil_Oil.id.desc())
.first())
gallons_put_in_home = Decimal(gallons_delivered)
get_auto_delivery = session.query(Auto_Delivery).filter(Auto_Delivery.id == autoid).first()
get_todays_price = session.query(Pricing_Oil_Oil).order_by(Pricing_Oil_Oil.id.desc()).first()
todays_price = Decimal(get_todays_price.price_for_customer)
total_amount = gallons_put_in_home * todays_price
total_amount = gallons_delivered * todays_price
create_new_ticket = Tickets_Auto_Delivery(
customer_id = get_auto_delivery.customer_id,
account_number = get_auto_delivery.account_number,
customer_town = get_auto_delivery.customer_town,
customer_state = get_auto_delivery.customer_state,
customer_address = get_auto_delivery.customer_address,
customer_zip =get_auto_delivery.customer_zip,
customer_full_name = get_auto_delivery.customer_full_name,
oil_prices_id = get_todays_price.id,
gallons_delivered = gallons_delivered,
price_per_gallon = get_todays_price.price_for_customer,
total_amount_customer = total_amount,
fill_date = date.today(),
payment_type = payment_type,
payment_card_id = payment_card_id,
payment_status = payment_status,
customer_id=get_auto_delivery.customer_id,
account_number=get_auto_delivery.account_number,
customer_town=get_auto_delivery.customer_town,
customer_state=get_auto_delivery.customer_state,
customer_address=get_auto_delivery.customer_address,
customer_zip=get_auto_delivery.customer_zip,
customer_full_name=get_auto_delivery.customer_full_name,
oil_prices_id=get_todays_price.id,
gallons_delivered=gallons_delivered,
price_per_gallon=get_todays_price.price_for_customer,
total_amount_customer=total_amount,
fill_date=date.today(),
payment_type=payment_type,
payment_card_id=payment_card_id,
payment_status=payment_status,
)
session.add(create_new_ticket)
session.commit() # Commit all changes in one transaction
return ({
"ok": True,
"auto_ticket_id":create_new_ticket.id
}), 200
session.commit()
return ({"ok": True, "auto_ticket_id":create_new_ticket.id}), 200

7
app/routers/delivery.py
View File

@@ -1,8 +1,7 @@
from fastapi import APIRouter, Request, Depends
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder
from database import session, get_db
from sqlalchemy.orm import Session
from database import session
from app.models.auto import Auto_Delivery, Tickets_Auto_Delivery
from app.models.delivery import Delivery
@@ -113,6 +112,6 @@ def update_auto_status(auto_id: int):
if update_status:
update_status.auto_status = 3
db.commit()
session.commit()
return {"message": "Auto status updated to 3"}
return {"error": "Auto delivery not found"}

181
app/routers/main.py
View File

@@ -1,13 +1,8 @@
from fastapi import APIRouter, Request
from datetime import datetime, date
from fastapi import APIRouter, HTTPException
from database import session
from pyowm import OWM
from app.models.auto import Auto_Delivery, Auto_Temp, Auto_Update
from app.models.delivery import Delivery
from app.script.fuel_estimator import FuelEstimator
from app.script.temp_getter import fetch_and_store_daily_temp
router = APIRouter(
prefix="/main",
@@ -15,150 +10,40 @@ router = APIRouter(
responses={404: {"description": "Not found"}},
)
def Average(lst):
return sum(lst) / len(lst)
@router.get("/temp", status_code=200)
def update_temp_manually():
"""
Manually triggers the fetch and storage of today's temperature.
This is useful for testing or for manual intervention if the cron job fails.
"""
try:
success = fetch_and_store_daily_temp()
if success:
session.commit()
return {"ok": True, "message": "Temperature updated or already exists."}
else:
# The function already rolled back, so just return an error
return HTTPException(status_code=500, detail="Failed to fetch temperature from the weather API.")
except Exception as e:
session.rollback()
raise HTTPException(status_code=500, detail=f"An unexpected server error occurred: {str(e)}")
@router.get("/update", status_code=200)
def update_auto_customers():
def update_all_customer_fuel_levels():
"""
This endpoint triggers the daily update for all customers.
It should be called once per day by a cron job or scheduler.
"""
try:
see_if_autos_updated = (session.query(Auto_Update)
.filter(Auto_Update.last_updated == date.today())
.first()
)
except:
estimator = FuelEstimator()
result = estimator.run_daily_update()
return result
except Exception as e:
session.rollback()
see_if_autos_updated = None
if see_if_autos_updated is not None:
return ({"ok": True}), 200
else:
create_new_update = Auto_Update(
last_updated = date.today()
)
today_temp = (session
.query(Auto_Temp)
.order_by(Auto_Temp.id.desc())
.first())
# get all automatic customers
auto_customers = (session
.query(Auto_Delivery)
.filter(Auto_Delivery.last_fill != None)
.order_by(Auto_Delivery.last_updated.desc())
.limit(25))
for f in auto_customers:
# figure out days since last fill
d1 = date.today()
d0 = f.last_fill
delta = d1 - d0
f.days_since_last_fill = delta.days
# figure out how much temperature effects oil
today_temptemp_avg = int(today_temp.temp_avg)
if today_temptemp_avg >= 65.01:
use_day = 1
elif 50.01 <= today_temptemp_avg <= 65:
use_day = 2
elif 35.01 <= today_temptemp_avg <= 50:
use_day = 3
elif 30.01 <= today_temptemp_avg <= 35:
use_day = 4
elif 25.01 <= today_temptemp_avg <= 30:
use_day = 5
elif 20.01 <= today_temptemp_avg <= 25:
use_day = 7
elif 15.01 <= today_temptemp_avg <= 20:
use_day = 8
elif 10.01 <= today_temptemp_avg <= 15:
use_day = 9
elif 5.01 <= today_temptemp_avg <= 10:
use_day = 10
elif 0.01 <= today_temptemp_avg <= 5:
use_day = 12
elif -20 <= today_temptemp_avg <= -1:
use_day = 15
else:
use_day = 0
# times temp factory by house factor
gallon_use_today = f.house_factor * use_day
# Log the exception e
return {"ok": False, "message": "An internal error occurred."}
# get previous day gallons left
f.estimated_gallons_left_prev_day = f.estimated_gallons_left
# get estimated gallons left
get_gallons_left = f.estimated_gallons_left_prev_day - gallon_use_today
f.estimated_gallons_left = get_gallons_left
f.last_updated = date.today()
session.add(create_new_update)
session.add(f)
session.commit()
return ({"ok": True}), 200
@router.get("/temp", status_code=200)
def update_temp():
try:
see_if_temp_exists = (session.query(Auto_Temp)
.filter(Auto_Temp.todays_date == date.today())
.first())
except:
see_if_temp_exists = None
if see_if_temp_exists is not None:
return ({"ok": True}), 200
else:
try:
temps = []
owm = OWM('21648d8c8d1a4ae495ace0b7810b4d36')
mgr = owm.weather_manager()
# Search for current weather in London (Great Britain) and get details
observation = mgr.weather_at_place('Worcester, US')
w = observation.weather
temp_dict_fahrenheit = w.temperature() # a dict in Kelvin units (default when no temperature units provided)
temp_dict_fahrenheit['temp_min']
temp_dict_fahrenheit['temp_max']
temp_dict_fahrenheit = w.temperature('fahrenheit')
low_temp = temp_dict_fahrenheit['temp_min']
high_temp = temp_dict_fahrenheit['temp_max']
temps.append(temp_dict_fahrenheit['temp_max'])
temps.append(temp_dict_fahrenheit['temp_min'])
get_avg = Average(temps)
rounded_temp = round(get_avg)
dday = (65 - ((low_temp + high_temp) /2))
add_new_temp = Auto_Temp(
temp = temp_dict_fahrenheit['temp'],
temp_max = temp_dict_fahrenheit['temp_max'],
temp_min = temp_dict_fahrenheit['temp_min'],
temp_avg = rounded_temp,
degree_day = dday,
todays_date = date.today()
)
session.add(add_new_temp)
session.commit()
return ({"ok": True}), 200
except:
return ({"ok": True}), 200

153
app/script/fuel_estimator.py Normal file
View File

@@ -0,0 +1,153 @@
from sqlalchemy.orm import Session
from sqlalchemy import func
from datetime import date, timedelta
from decimal import Decimal
# Import your existing database models
from app.models.auto import Auto_Delivery, Auto_Temp, Auto_Update, Tickets_Auto_Delivery
# --- Constants for the Model ---
# This is a baseline daily usage for homes that use oil for hot water.
# A typical value is 0.5 to 1.0 gallons per day. Adjust as needed.
HOT_WATER_DAILY_USAGE = Decimal('0.7')
# This determines how quickly the K-Factor adjusts.
# 0.7 means 70% weight is given to the historical factor and 30% to the new one.
# This prevents wild swings from a single unusual delivery period.
K_FACTOR_SMOOTHING_WEIGHT = Decimal('0.7')
class FuelEstimator:
def __init__(self, session: Session):
self.session = session
def _get_weather_for_date(self, target_date: date) -> Auto_Temp | None:
"""Helper to fetch weather data for a specific date."""
return self.session.query(Auto_Temp).filter(Auto_Temp.todays_date == target_date).first()
def run_daily_update(self):
"""
Main function to run once per day. It updates the estimated fuel level
for all active automatic delivery customers. The calling function must commit the session.
"""
today = date.today()
# 1. Check if the update has already run today
if self.session.query(Auto_Update).filter(Auto_Update.last_updated == today).first():
print(f"Daily update for {today} has already been completed.")
return {"ok": True, "message": "Update already run today."}
# 2. Get today's weather data (specifically the Heating Degree Days)
todays_weather = self._get_weather_for_date(today)
if not todays_weather:
print(f"Error: Weather data for {today} not found. Cannot run update.")
return {"ok": False, "message": f"Weather data for {today} not found."}
# Degree days can't be negative for this calculation. If it's warm, HDD is 0.
degree_day = Decimal(max(0, todays_weather.degree_day))
# 3. Get all active automatic customers
auto_customers = self.session.query(Auto_Delivery).filter(
Auto_Delivery.auto_status == 1 # Assuming 1 means active
).all()
if not auto_customers:
print("No active automatic delivery customers found.")
return {"ok": True, "message": "No active customers to update."}
print(f"Staging daily fuel update for {len(auto_customers)} customers...")
# 4. Loop through each customer and update their fuel level
for customer in auto_customers:
heating_usage = customer.house_factor * degree_day
hot_water_usage = Decimal('0.0')
if customer.hot_water_summer == 1:
hot_water_usage = HOT_WATER_DAILY_USAGE
gallons_used_today = heating_usage + hot_water_usage
customer.estimated_gallons_left_prev_day = customer.estimated_gallons_left
new_estimated_gallons = customer.estimated_gallons_left - gallons_used_today
customer.estimated_gallons_left = max(Decimal('0.0'), new_estimated_gallons)
customer.last_updated = today
if customer.days_since_last_fill is not None:
customer.days_since_last_fill += 1
# 5. Log that today's update is complete
new_update_log = Auto_Update(last_updated=today)
self.session.add(new_update_log)
print("Daily update staged. Awaiting commit.")
return {"ok": True, "message": f"Successfully staged updates for {len(auto_customers)} customers."}
def refine_factor_after_delivery(self, ticket: Tickets_Auto_Delivery):
"""
This is the self-correction logic. It recalculates and refines the customer's
K-Factor (house_factor) after a delivery. The calling function must commit the session.
"""
customer = self.session.query(Auto_Delivery).filter(
Auto_Delivery.customer_id == ticket.customer_id
).first()
if not customer or not customer.last_fill:
print(f"Cannot refine K-Factor: Customer {ticket.customer_id} not found or has no previous fill date. Resetting tank only.")
self._update_tank_after_fill(customer, ticket)
return
start_date = customer.last_fill
end_date = ticket.fill_date
if start_date >= end_date:
print(f"Cannot refine K-Factor for customer {ticket.customer_id}: New fill date is not after the last one. Resetting tank only.")
self._update_tank_after_fill(customer, ticket)
return
total_hdd_result = self.session.query(func.sum(Auto_Temp.degree_day)).filter(
Auto_Temp.todays_date > start_date,
Auto_Temp.todays_date <= end_date,
Auto_Temp.degree_day > 0
).scalar()
total_hdd = Decimal(total_hdd_result or 0)
total_hot_water_usage = Decimal('0.0')
if customer.hot_water_summer == 1:
num_days = (end_date - start_date).days
total_hot_water_usage = Decimal(num_days) * HOT_WATER_DAILY_USAGE
gallons_for_heating = ticket.gallons_delivered - total_hot_water_usage
if gallons_for_heating <= 0 or total_hdd == 0:
print(f"Cannot calculate new K-Factor for customer {ticket.customer_id}. (HDD: {total_hdd}, Heating Gallons: {gallons_for_heating}). Resetting tank only.")
self._update_tank_after_fill(customer, ticket)
return
new_k_factor = gallons_for_heating / total_hdd
current_k_factor = customer.house_factor
smoothed_k_factor = (current_k_factor * K_FACTOR_SMOOTHING_WEIGHT) + (new_k_factor * (Decimal('1.0') - K_FACTOR_SMOOTHING_WEIGHT))
print(f"Refining K-Factor for Customer ID {customer.customer_id}:")
print(f" - Old K-Factor: {current_k_factor:.4f}, New Smoothed K-Factor: {smoothed_k_factor:.4f}")
customer.house_factor = smoothed_k_factor
self._update_tank_after_fill(customer, ticket)
print(f"K-Factor and tank status for Customer {customer.customer_id} staged for update.")
def _update_tank_after_fill(self, customer: Auto_Delivery, ticket: Tickets_Auto_Delivery):
"""Helper to reset customer tank status after a fill-up."""
customer.last_fill = ticket.fill_date
customer.days_since_last_fill = 0
# A "fill-up" means the tank is full. This is critical for accuracy.
if customer.tank_size and Decimal(customer.tank_size) > 0:
customer.estimated_gallons_left = Decimal(customer.tank_size)
else:
# Default to a common tank size if not specified, e.g., 275
customer.estimated_gallons_left = Decimal('275.0')
# The previous day's value should match the new full value on a fill day.
customer.estimated_gallons_left_prev_day = customer.estimated_gallons_left
customer.last_updated = date.today()
customer.auto_status = 1 # Reactivate the customer

73
app/script/temp_getter.py Normal file
View File

@@ -0,0 +1,73 @@
from datetime import date
import requests
from decimal import Decimal
# Import your database model and the session from your database configuration
from app.models.auto import Auto_Temp
from database import session
def Average(lst):
"""Calculates the average of a list of numbers."""
return sum(lst) / len(lst)
def fetch_and_store_daily_temp() -> bool:
"""
Fetches the current day's weather for Worcester, MA, calculates the
Heating Degree Day (HDD), and stores it in the database if it doesn't already exist.
Returns:
bool: True if the temperature is successfully fetched and stored (or already exists),
False if an error occurs.
"""
# 1. Check if the temperature for today already exists in the database
today = date.today()
if session.query(Auto_Temp).filter(Auto_Temp.todays_date == today).first():
print(f"Temperature for {today} already exists in the database. Skipping fetch.")
return True
# 2. If it doesn't exist, fetch it from the API
print(f"Fetching temperature for {today} from OpenWeatherMap...")
try:
# API key and location
api_key = '21648d8c8d1a4ae495ace0b7810b4d36'
location = 'Worcester,US'
# Make request to OpenWeatherMap Current Weather API with imperial units for Fahrenheit
url = f"http://api.openweathermap.org/data/2.5/weather?q={location}&appid={api_key}&units=imperial"
response = requests.get(url)
response.raise_for_status() # Raise an error for bad status codes
data = response.json()
# Extract temperatures in Fahrenheit
main = data.get('main', {})
current_temp = Decimal(main.get('temp', 0))
low_temp = Decimal(main.get('temp_min', 0))
high_temp = Decimal(main.get('temp_max', 0))
# Calculate average temperature and Heating Degree Days (HDD)
avg_temp = (low_temp + high_temp) / 2
# HDD is based on a baseline of 65°F. It cannot be negative for heating calculations.
degree_day = max(Decimal(0), 65 - avg_temp)
# 3. Create a new database object
add_new_temp = Auto_Temp(
temp=current_temp,
temp_max=high_temp,
temp_min=low_temp,
temp_avg=round(avg_temp, 2),
degree_day=round(degree_day),
todays_date=today
)
# 4. Add the new record to the session (it will be committed by the calling function)
session.add(add_new_temp)
print(f"Successfully fetched and staged temperature for {today}.")
return True
except Exception as e:
print(f"An error occurred while fetching weather data: {e}")
# Make sure to rollback the session in case of a partial failure
session.rollback()
return False

26
app/script/update_auto.py
View File

@@ -1,26 +0,0 @@
from decimal import Decimal
from database import session
def calc_home_factor(gallons_put_in_home, current_house_factor):
if 200.01 <= gallons_put_in_home <= 500:
customer_home_factor = Decimal(current_house_factor) + Decimal(1)
elif 170.01 <= gallons_put_in_home <= 200:
customer_home_factor = Decimal(current_house_factor) + Decimal(1.5)
elif 150.01 <= gallons_put_in_home <= 170:
customer_home_factor = Decimal(current_house_factor) + Decimal(1.25)
elif 120.01 <= gallons_put_in_home <= 150:
customer_home_factor = Decimal(current_house_factor) - Decimal(.25)
elif 90.01 <= gallons_put_in_home <= 120:
customer_home_factor = Decimal(current_house_factor) - Decimal(.50)
elif 0.01 <= gallons_put_in_home <= 90:
customer_home_factor = Decimal(current_house_factor) - Decimal(.75)
else:
customer_home_factor = Decimal(current_house_factor)
if customer_home_factor <= 0:
customer_home_factor = Decimal(.25)
return customer_home_factor

3
requirements.txt
View File

@@ -2,5 +2,4 @@ fastapi
uvicorn[standard]
psycopg2-binary
sqlalchemy
pyowm
setuptools
requests