feat: --fill-gaps all repairs missing data across the whole DB range
Documentation / Generate API Documentation (push) Successful in 11s
Documentation / Build Sphinx Documentation (push) Successful in 17s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Cleanup (push) Successful in 1s
Documentation / Documentation Summary (push) Successful in 3s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Test Suite (3.11) (push) Failing after 29s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Build Docker Image (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Integration Test with Services (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Deploy to Staging (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Deploy to Production (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Performance Test (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Code Quality (push) Successful in 14s
Documentation / Validate Documentation (push) Failing after 9s
Documentation / Generate API Documentation (push) Successful in 11s
Documentation / Build Sphinx Documentation (push) Successful in 17s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Cleanup (push) Successful in 1s
Documentation / Documentation Summary (push) Successful in 3s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Test Suite (3.11) (push) Failing after 29s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Build Docker Image (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Integration Test with Services (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Deploy to Staging (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Deploy to Production (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Performance Test (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Code Quality (push) Successful in 14s
Documentation / Validate Documentation (push) Failing after 9s
Gap detection is now hour-granular: days with partial data (missing hourly slots) are re-fetched, not just days with no rows at all. The API's hour-24-is-next-midnight quirk is handled by re-fetching day D-1 when day D is missing its 00:00 slot. SQL adapters gain single-query range and recorded-hours lookups; non-SQL backends fall back to the old day-granular check.
This commit is contained in:
@@ -36,6 +36,26 @@ class DatabaseAdapter(ABC):
|
||||
def get_measurements_for_date(self, target_date: datetime.datetime) -> List[Dict]:
|
||||
pass
|
||||
|
||||
def get_measurement_date_range(
|
||||
self,
|
||||
) -> Optional[tuple]:
|
||||
"""Return (min_timestamp, max_timestamp) of stored measurements.
|
||||
|
||||
Returns None when the backend has no data or does not support the query.
|
||||
"""
|
||||
return None
|
||||
|
||||
def get_recorded_hours_by_day(
|
||||
self, start_date: datetime.date, end_date: datetime.date
|
||||
) -> Optional[Dict[datetime.date, set]]:
|
||||
"""Map each day in [start_date, end_date] to the set of hours (0-23)
|
||||
that have at least one measurement.
|
||||
|
||||
Returns None when the backend does not support hour-granular gap
|
||||
detection (callers should fall back to day-granular checks).
|
||||
"""
|
||||
return None
|
||||
|
||||
|
||||
# InfluxDB Adapter
|
||||
class InfluxDBAdapter(DatabaseAdapter):
|
||||
@@ -657,6 +677,93 @@ class SQLAdapter(DatabaseAdapter):
|
||||
)
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def _coerce_date(value) -> Optional[datetime.date]:
|
||||
"""Normalize a DB-returned day value (str/date/datetime) to a date."""
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, datetime.datetime):
|
||||
return value.date()
|
||||
if isinstance(value, datetime.date):
|
||||
return value
|
||||
# SQLite returns strings, e.g. '2024-09-15'
|
||||
return datetime.datetime.strptime(str(value)[:10], "%Y-%m-%d").date()
|
||||
|
||||
def get_measurement_date_range(self) -> Optional[tuple]:
|
||||
if not self.engine:
|
||||
return None
|
||||
|
||||
try:
|
||||
from sqlalchemy import text
|
||||
|
||||
query = "SELECT MIN(timestamp), MAX(timestamp) FROM water_measurements"
|
||||
with self.engine.connect() as conn:
|
||||
row = conn.execute(text(query)).fetchone()
|
||||
|
||||
if not row or row[0] is None:
|
||||
return None
|
||||
|
||||
def to_datetime(value):
|
||||
if isinstance(value, datetime.datetime):
|
||||
return value
|
||||
return datetime.datetime.fromisoformat(str(value)[:19])
|
||||
|
||||
return (to_datetime(row[0]), to_datetime(row[1]))
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Error querying {self.db_type.upper()} date range: {e}")
|
||||
return None
|
||||
|
||||
def get_recorded_hours_by_day(
|
||||
self, start_date: datetime.date, end_date: datetime.date
|
||||
) -> Optional[Dict[datetime.date, set]]:
|
||||
if not self.engine:
|
||||
return None
|
||||
|
||||
try:
|
||||
from sqlalchemy import text
|
||||
|
||||
if self.db_type == "sqlite":
|
||||
day_expr = "DATE(timestamp)"
|
||||
hour_expr = "CAST(strftime('%H', timestamp) AS INTEGER)"
|
||||
elif self.db_type == "postgresql":
|
||||
day_expr = "CAST(timestamp AS DATE)"
|
||||
hour_expr = "CAST(EXTRACT(HOUR FROM timestamp) AS INTEGER)"
|
||||
else: # MySQL
|
||||
day_expr = "DATE(timestamp)"
|
||||
hour_expr = "HOUR(timestamp)"
|
||||
|
||||
query = f"""
|
||||
SELECT {day_expr} AS day, {hour_expr} AS hour
|
||||
FROM water_measurements
|
||||
WHERE timestamp >= :start_time AND timestamp < :end_time
|
||||
GROUP BY {day_expr}, {hour_expr}
|
||||
"""
|
||||
|
||||
start_time = datetime.datetime.combine(start_date, datetime.time.min)
|
||||
end_time = datetime.datetime.combine(
|
||||
end_date + datetime.timedelta(days=1), datetime.time.min
|
||||
)
|
||||
|
||||
hours_by_day: Dict[datetime.date, set] = {}
|
||||
with self.engine.connect() as conn:
|
||||
result = conn.execute(
|
||||
text(query), {"start_time": start_time, "end_time": end_time}
|
||||
)
|
||||
for row in result:
|
||||
day = self._coerce_date(row[0])
|
||||
if day is None:
|
||||
continue
|
||||
hours_by_day.setdefault(day, set()).add(int(row[1]))
|
||||
|
||||
return hours_by_day
|
||||
|
||||
except Exception as e:
|
||||
logging.error(
|
||||
f"Error querying {self.db_type.upper()} recorded hours: {e}"
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
# VictoriaMetrics Adapter (using Prometheus format)
|
||||
class VictoriaMetricsAdapter(DatabaseAdapter):
|
||||
|
||||
+22
-6
@@ -180,8 +180,11 @@ def run_continuous_monitoring():
|
||||
return True
|
||||
|
||||
|
||||
def run_gap_filling(days_back: int):
|
||||
"""Run gap filling for missing data"""
|
||||
def run_gap_filling(days_back: Optional[int]):
|
||||
"""Run gap filling for missing data (days_back=None scans the whole range)"""
|
||||
if days_back is None:
|
||||
logger.info("Checking for data gaps across the whole data range...")
|
||||
else:
|
||||
logger.info(f"Checking for data gaps in the last {days_back} days...")
|
||||
|
||||
try:
|
||||
@@ -445,6 +448,7 @@ Examples:
|
||||
%(prog)s # Run continuous monitoring
|
||||
%(prog)s --web-api # Start web API server
|
||||
%(prog)s --fill-gaps 7 # Fill missing data for last 7 days
|
||||
%(prog)s --fill-gaps all # Fill missing data across the whole data range
|
||||
%(prog)s --update-data 2 # Update existing data for last 2 days
|
||||
%(prog)s --import-historical 2024-01-01 2024-01-31 # Import historical data
|
||||
%(prog)s --status # Show system status
|
||||
@@ -461,9 +465,11 @@ Examples:
|
||||
|
||||
parser.add_argument(
|
||||
"--fill-gaps",
|
||||
type=int,
|
||||
metavar="DAYS",
|
||||
help="Fill missing data gaps for the specified number of days back",
|
||||
metavar="DAYS|all",
|
||||
help=(
|
||||
"Fill missing data gaps for the specified number of days back, "
|
||||
"or 'all' to scan the entire data range in the database"
|
||||
),
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
@@ -529,7 +535,17 @@ Examples:
|
||||
elif args.web_api:
|
||||
success = run_web_api()
|
||||
elif args.fill_gaps is not None:
|
||||
success = run_gap_filling(args.fill_gaps)
|
||||
if args.fill_gaps.lower() == "all":
|
||||
success = run_gap_filling(None)
|
||||
else:
|
||||
try:
|
||||
success = run_gap_filling(int(args.fill_gaps))
|
||||
except ValueError:
|
||||
logger.error(
|
||||
f"Invalid --fill-gaps value '{args.fill_gaps}': "
|
||||
"expected a number of days or 'all'"
|
||||
)
|
||||
sys.exit(1)
|
||||
elif args.update_data is not None:
|
||||
success = run_data_update(args.update_data)
|
||||
elif args.import_historical is not None:
|
||||
|
||||
+99
-23
@@ -579,53 +579,129 @@ class EnhancedWaterMonitorScraper:
|
||||
increment_counter("scraping_cycles_failed")
|
||||
return False
|
||||
|
||||
def fill_data_gaps(self, days_back: int) -> int:
|
||||
"""Fill gaps in data for the specified number of days back"""
|
||||
def fill_data_gaps(self, days_back: Optional[int] = None) -> int:
|
||||
"""Fill gaps in stored data by re-fetching incomplete days.
|
||||
|
||||
Args:
|
||||
days_back: How many days back to scan. None scans the whole data
|
||||
range, from the earliest measurement in the database to now.
|
||||
|
||||
Detection is hour-granular where the backend supports it: any calendar
|
||||
day missing one or more hourly slots gets its source date(s) re-fetched.
|
||||
Because the API reports hour 24 as midnight of the next day, a missing
|
||||
00:00 slot on day D is repaired by re-fetching day D-1.
|
||||
"""
|
||||
logger = get_logger(__name__)
|
||||
filled_count = 0
|
||||
|
||||
try:
|
||||
# Calculate date range
|
||||
end_date = datetime.datetime.now()
|
||||
now = datetime.datetime.now()
|
||||
end_date = now.date()
|
||||
|
||||
if days_back is not None:
|
||||
start_date = end_date - datetime.timedelta(days=days_back)
|
||||
|
||||
logger.info(
|
||||
f"Checking for gaps from {start_date.date()} to {end_date.date()}"
|
||||
else:
|
||||
if not self.db_adapter:
|
||||
logger.error("Database adapter not initialized")
|
||||
return 0
|
||||
date_range = self.db_adapter.get_measurement_date_range()
|
||||
if not date_range:
|
||||
logger.error(
|
||||
"Database is empty or does not support range queries; "
|
||||
"use --import-historical to seed data first"
|
||||
)
|
||||
return 0
|
||||
start_date = date_range[0].date()
|
||||
|
||||
# Iterate through each date in the range
|
||||
current_date = start_date
|
||||
while current_date <= end_date:
|
||||
# Check if we have data for this date
|
||||
has_data = self._check_data_exists_for_date(current_date)
|
||||
logger.info(f"Checking for gaps from {start_date} to {end_date}")
|
||||
|
||||
if not has_data:
|
||||
logger.info(f"Filling gap for date: {current_date.date()}")
|
||||
fetch_dates = self._find_gap_fetch_dates(start_date, end_date, now)
|
||||
|
||||
# Fetch data for this specific date
|
||||
data = self.fetch_water_data_for_date(current_date)
|
||||
if not fetch_dates:
|
||||
logger.info("No gaps found")
|
||||
return 0
|
||||
|
||||
logger.info(f"Found {len(fetch_dates)} day(s) needing a re-fetch")
|
||||
|
||||
for fetch_date in sorted(fetch_dates):
|
||||
fetch_dt = datetime.datetime.combine(fetch_date, datetime.time.min)
|
||||
logger.info(f"Filling gap for date: {fetch_date}")
|
||||
|
||||
data = self.fetch_water_data_for_date(fetch_dt)
|
||||
|
||||
if data:
|
||||
# Save the data
|
||||
if self.save_to_database(data):
|
||||
filled_count += len(data)
|
||||
logger.info(
|
||||
f"Filled {len(data)} measurements for {current_date.date()}"
|
||||
f"Filled {len(data)} measurements for {fetch_date}"
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
f"Failed to save data for {current_date.date()}"
|
||||
)
|
||||
logger.warning(f"Failed to save data for {fetch_date}")
|
||||
else:
|
||||
logger.warning(f"No data available for {current_date.date()}")
|
||||
logger.warning(f"No data available for {fetch_date}")
|
||||
|
||||
current_date += datetime.timedelta(days=1)
|
||||
# Be respectful to the API
|
||||
time.sleep(1)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Gap filling error: {e}")
|
||||
|
||||
return filled_count
|
||||
|
||||
def _find_gap_fetch_dates(
|
||||
self,
|
||||
start_date: datetime.date,
|
||||
end_date: datetime.date,
|
||||
now: datetime.datetime,
|
||||
) -> List[datetime.date]:
|
||||
"""Determine which source dates need re-fetching to fill gaps.
|
||||
|
||||
Uses hour-granular detection when the adapter supports it, otherwise
|
||||
falls back to re-fetching only days with no data at all.
|
||||
"""
|
||||
hours_by_day = None
|
||||
if self.db_adapter:
|
||||
hours_by_day = self.db_adapter.get_recorded_hours_by_day(
|
||||
start_date, end_date
|
||||
)
|
||||
|
||||
fetch_dates = set()
|
||||
|
||||
if hours_by_day is None:
|
||||
logger.info(
|
||||
"Backend does not support hour-granular gap detection; "
|
||||
"checking for fully-missing days only"
|
||||
)
|
||||
current = start_date
|
||||
while current <= end_date:
|
||||
current_dt = datetime.datetime.combine(current, datetime.time.min)
|
||||
if not self._check_data_exists_for_date(current_dt):
|
||||
fetch_dates.add(current)
|
||||
current += datetime.timedelta(days=1)
|
||||
return sorted(fetch_dates)
|
||||
|
||||
current = start_date
|
||||
while current <= end_date:
|
||||
if current == now.date():
|
||||
# Today: only expect hours that have already passed
|
||||
expected_hours = set(range(0, now.hour))
|
||||
else:
|
||||
expected_hours = set(range(24))
|
||||
|
||||
missing = expected_hours - hours_by_day.get(current, set())
|
||||
|
||||
if missing:
|
||||
# Hours 1-23 of day D come from fetching D; hour 0 comes from
|
||||
# the previous day's fetch (the API's "hour 24")
|
||||
if any(h >= 1 for h in missing):
|
||||
fetch_dates.add(current)
|
||||
if 0 in missing and current > start_date:
|
||||
fetch_dates.add(current - datetime.timedelta(days=1))
|
||||
|
||||
current += datetime.timedelta(days=1)
|
||||
|
||||
return sorted(fetch_dates)
|
||||
|
||||
def update_existing_data(self, days_back: int) -> int:
|
||||
"""Update existing data with latest values for the specified number of days back"""
|
||||
logger = get_logger(__name__)
|
||||
|
||||
Reference in New Issue
Block a user