105 lines
3.0 KiB
Python
105 lines
3.0 KiB
Python
|
from __future__ import annotations
|
||
|
|
from abc import abstractmethod
|
||
|
|
from enum import Enum
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import override
|
||
|
|
|
||
|
|
import psycopg2
|
||
|
|
from psycopg2 import sql
|
||
|
|
|
||
|
|
|
||
|
|
class RampStatus(Enum):
|
||
|
|
OPEN = "OPEN"
|
||
|
|
CLOSED = "CLOSED"
|
||
|
|
UNKNOWN = "UNKNOWN"
|
||
|
|
|
||
|
|
@override
|
||
|
|
def __str__(self):
|
||
|
|
|
||
|
|
status_str = "Unknown"
|
||
|
|
match self:
|
||
|
|
case RampStatus.OPEN:
|
||
|
|
status_str = "Open"
|
||
|
|
case RampStatus.CLOSED:
|
||
|
|
status_str = "Closed"
|
||
|
|
case RampStatus.UNKNOWN:
|
||
|
|
status_str = "Unknown"
|
||
|
|
return status_str
|
||
|
|
|
||
|
|
|
||
|
|
class RampOperator(Enum):
|
||
|
|
CANYON_MARINA = "Canyon Lake Marina"
|
||
|
|
COMAL_COUNTY = "Comal County"
|
||
|
|
USACE = "US Army Corps of Engineers"
|
||
|
|
CRANES_MILL = "Cranes Mill Marina"
|
||
|
|
JBSA = "Joint Base San Antonio"
|
||
|
|
LCYC = "Lake Canyon Yacht Club"
|
||
|
|
WORD = "Water-Oriented Recreation District of Comal County"
|
||
|
|
|
||
|
|
|
||
|
|
class RampOperatingTimes(Enum):
|
||
|
|
CANYON_MARINA = "8am - Sunset"
|
||
|
|
COMAL_COUNTY = "8am - Sunset"
|
||
|
|
USACE = "8am - Sunset"
|
||
|
|
CRANES_MILL = None
|
||
|
|
JBSA = "8am - Sunset"
|
||
|
|
LCYC = "24/7"
|
||
|
|
WORD = "8am - 7:30PM"
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class Ramp:
|
||
|
|
"""A represenation of a single boat ramp at Canyon Lake
|
||
|
|
|
||
|
|
Attributes:
|
||
|
|
ramp_number: the ramp number according to the US Army Corps of Engineers
|
||
|
|
address: the approximate address of the ramp
|
||
|
|
location: the location name of the ramp
|
||
|
|
operator: the organization that runs/operates the ramp
|
||
|
|
status: the ramp's current open status
|
||
|
|
"""
|
||
|
|
|
||
|
|
ramp_number: int
|
||
|
|
name: str
|
||
|
|
address: str
|
||
|
|
operator: RampOperator
|
||
|
|
status: RampStatus
|
||
|
|
last_updated: datetime
|
||
|
|
operating_times: RampOperatingTimes | None
|
||
|
|
|
||
|
|
@abstractmethod
|
||
|
|
def fetch_data(self):
|
||
|
|
pass
|
||
|
|
|
||
|
|
def save_to_db(self, conn: psycopg2.extensions.connection):
|
||
|
|
with conn.cursor() as cur:
|
||
|
|
cur.execute(
|
||
|
|
sql.SQL(
|
||
|
|
"""
|
||
|
|
INSERT INTO
|
||
|
|
{}
|
||
|
|
VALUES
|
||
|
|
(%s, %s, %s, %s, %s, %s, %s)
|
||
|
|
ON CONFLICT (number) DO UPDATE SET
|
||
|
|
number = EXCLUDED.number,
|
||
|
|
name = EXCLUDED.name,
|
||
|
|
operator = EXCLUDED.operator,
|
||
|
|
address = EXCLUDED.address,
|
||
|
|
status = EXCLUDED.status,
|
||
|
|
last_updated = EXCLUDED.last_updated,
|
||
|
|
operating_times = EXCLUDED.operating_times
|
||
|
|
"""
|
||
|
|
).format(sql.Identifier("""rampdata""")),
|
||
|
|
[
|
||
|
|
self.ramp_number,
|
||
|
|
self.name,
|
||
|
|
self.operator.value,
|
||
|
|
self.address,
|
||
|
|
str(self.status),
|
||
|
|
self.last_updated.strftime("%Y-%m-%d"),
|
||
|
|
self.operating_times.value if self.operating_times else None,
|
||
|
|
],
|
||
|
|
)
|
||
|
|
conn.commit()
|