added full cli package and CI execution
This commit is contained in:
Brian Bjarke Jensen
@@ -0,0 +1,9 @@
|
||||
# Eloverblik API Configuration
|
||||
ELOVERBLIK_API_TOKEN=your_refresh_token_here
|
||||
|
||||
# Database Configuration
|
||||
DB_HOST=localhost
|
||||
DB_NAME=energy_consumption
|
||||
DB_USER=energy_user
|
||||
DB_PASSWORD=your_password_here
|
||||
DB_PORT=5432
|
||||
@@ -0,0 +1,51 @@
|
||||
name: Daily Health Check
|
||||
|
||||
on:
|
||||
schedule:
|
||||
# Run daily at 6:00 AM UTC
|
||||
- cron: "0 6 * * *"
|
||||
workflow_dispatch: # Allow manual triggering
|
||||
|
||||
jobs:
|
||||
health-check:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v4
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Install uv
|
||||
run: |
|
||||
curl -LsSf https://astral.sh/uv/install.sh | sh
|
||||
echo "$HOME/.cargo/bin" >> $GITHUB_PATH
|
||||
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
uv sync
|
||||
|
||||
- name: Run health check
|
||||
env:
|
||||
ELOVERBLIK_API_TOKEN: ${{ secrets.ELOVERBLIK_API_TOKEN }}
|
||||
DB_HOST: ${{ secrets.DB_HOST }}
|
||||
DB_NAME: ${{ secrets.DB_NAME }}
|
||||
DB_USER: ${{ secrets.DB_USER }}
|
||||
DB_PASSWORD: ${{ secrets.DB_PASSWORD }}
|
||||
DB_PORT: ${{ secrets.DB_PORT }}
|
||||
run: |
|
||||
uv run energy-ingester health-check --verbose
|
||||
|
||||
- name: Report status
|
||||
run: |
|
||||
echo "Health check completed successfully!"
|
||||
|
||||
- name: Notify on failure
|
||||
if: failure()
|
||||
run: |
|
||||
echo "Health check failed! System may be down."
|
||||
# Add notification logic here (email, Slack, etc.)
|
||||
exit 1
|
||||
@@ -0,0 +1,177 @@
|
||||
name: Database Setup
|
||||
|
||||
on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
action:
|
||||
description: "Database action to perform"
|
||||
required: true
|
||||
default: "setup"
|
||||
type: choice
|
||||
options:
|
||||
- setup
|
||||
- migrate
|
||||
- reset
|
||||
confirm:
|
||||
description: 'Type "yes" to confirm destructive operations'
|
||||
required: false
|
||||
default: "no"
|
||||
type: string
|
||||
|
||||
jobs:
|
||||
database-setup:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v4
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Install uv
|
||||
run: |
|
||||
curl -LsSf https://astral.sh/uv/install.sh | sh
|
||||
echo "$HOME/.cargo/bin" >> $GITHUB_PATH
|
||||
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
uv sync
|
||||
|
||||
- name: Validate inputs
|
||||
run: |
|
||||
if [[ "${{ github.event.inputs.action }}" == "reset" && "${{ github.event.inputs.confirm }}" != "yes" ]]; then
|
||||
echo "❌ Reset operation requires confirmation. Please type 'yes' in the confirm field."
|
||||
exit 1
|
||||
fi
|
||||
|
||||
- name: Database Setup
|
||||
if: github.event.inputs.action == 'setup'
|
||||
env:
|
||||
ELOVERBLIK_API_TOKEN: ${{ secrets.ELOVERBLIK_API_TOKEN }}
|
||||
DB_HOST: ${{ secrets.DB_HOST }}
|
||||
DB_NAME: ${{ secrets.DB_NAME }}
|
||||
DB_USER: ${{ secrets.DB_USER }}
|
||||
DB_PASSWORD: ${{ secrets.DB_PASSWORD }}
|
||||
DB_PORT: ${{ secrets.DB_PORT }}
|
||||
run: |
|
||||
echo "🏗️ Setting up database tables..."
|
||||
uv run python -c "
|
||||
from energy_consumption_ingester import DatabaseStorage
|
||||
from dotenv import load_dotenv
|
||||
import sys
|
||||
|
||||
load_dotenv()
|
||||
db = DatabaseStorage.from_env()
|
||||
|
||||
if db.connect():
|
||||
print('✅ Database connection successful')
|
||||
if db.create_tables():
|
||||
print('✅ Database tables created successfully')
|
||||
else:
|
||||
print('❌ Failed to create tables')
|
||||
sys.exit(1)
|
||||
else:
|
||||
print('❌ Failed to connect to database')
|
||||
sys.exit(1)
|
||||
|
||||
db.close()
|
||||
print('🎉 Database setup completed!')
|
||||
"
|
||||
|
||||
- name: Database Migration
|
||||
if: github.event.inputs.action == 'migrate'
|
||||
env:
|
||||
ELOVERBLIK_API_TOKEN: ${{ secrets.ELOVERBLIK_API_TOKEN }}
|
||||
DB_HOST: ${{ secrets.DB_HOST }}
|
||||
DB_NAME: ${{ secrets.DB_NAME }}
|
||||
DB_USER: ${{ secrets.DB_USER }}
|
||||
DB_PASSWORD: ${{ secrets.DB_PASSWORD }}
|
||||
DB_PORT: ${{ secrets.DB_PORT }}
|
||||
run: |
|
||||
echo "🔄 Running database migrations..."
|
||||
uv run python -c "
|
||||
from energy_consumption_ingester import DatabaseStorage
|
||||
from dotenv import load_dotenv
|
||||
import sys
|
||||
|
||||
load_dotenv()
|
||||
db = DatabaseStorage.from_env()
|
||||
|
||||
if db.connect():
|
||||
print('✅ Database connection successful')
|
||||
if db.create_tables():
|
||||
print('✅ Database migrations completed')
|
||||
else:
|
||||
print('❌ Migration failed')
|
||||
sys.exit(1)
|
||||
else:
|
||||
print('❌ Failed to connect to database')
|
||||
sys.exit(1)
|
||||
|
||||
db.close()
|
||||
print('🎉 Migration completed!')
|
||||
"
|
||||
|
||||
- name: Database Reset (DESTRUCTIVE)
|
||||
if: github.event.inputs.action == 'reset' && github.event.inputs.confirm == 'yes'
|
||||
env:
|
||||
ELOVERBLIK_API_TOKEN: ${{ secrets.ELOVERBLIK_API_TOKEN }}
|
||||
DB_HOST: ${{ secrets.DB_HOST }}
|
||||
DB_NAME: ${{ secrets.DB_NAME }}
|
||||
DB_USER: ${{ secrets.DB_USER }}
|
||||
DB_PASSWORD: ${{ secrets.DB_PASSWORD }}
|
||||
DB_PORT: ${{ secrets.DB_PORT }}
|
||||
run: |
|
||||
echo "🚨 RESETTING DATABASE - THIS WILL DELETE ALL DATA!"
|
||||
uv run python -c "
|
||||
from energy_consumption_ingester import DatabaseStorage
|
||||
from dotenv import load_dotenv
|
||||
import sys
|
||||
|
||||
load_dotenv()
|
||||
db = DatabaseStorage.from_env()
|
||||
|
||||
if db.connect():
|
||||
print('✅ Database connection successful')
|
||||
|
||||
# Drop tables
|
||||
cursor = db.connection.cursor()
|
||||
cursor.execute('DROP TABLE IF EXISTS consumption_readings CASCADE;')
|
||||
cursor.execute('DROP TABLE IF EXISTS metering_points CASCADE;')
|
||||
cursor.execute('DROP FUNCTION IF EXISTS update_updated_at_column() CASCADE;')
|
||||
db.connection.commit()
|
||||
print('🗑️ Existing tables dropped')
|
||||
|
||||
# Recreate tables
|
||||
if db.create_tables():
|
||||
print('✅ Database tables recreated')
|
||||
else:
|
||||
print('❌ Failed to recreate tables')
|
||||
sys.exit(1)
|
||||
else:
|
||||
print('❌ Failed to connect to database')
|
||||
sys.exit(1)
|
||||
|
||||
db.close()
|
||||
print('🎉 Database reset completed!')
|
||||
"
|
||||
|
||||
- name: Test Connection
|
||||
env:
|
||||
ELOVERBLIK_API_TOKEN: ${{ secrets.ELOVERBLIK_API_TOKEN }}
|
||||
DB_HOST: ${{ secrets.DB_HOST }}
|
||||
DB_NAME: ${{ secrets.DB_NAME }}
|
||||
DB_USER: ${{ secrets.DB_USER }}
|
||||
DB_PASSWORD: ${{ secrets.DB_PASSWORD }}
|
||||
DB_PORT: ${{ secrets.DB_PORT }}
|
||||
run: |
|
||||
echo "🔍 Testing system health..."
|
||||
uv run energy-ingester health-check --verbose
|
||||
|
||||
- name: Summary
|
||||
run: |
|
||||
echo "✅ Database operation '${{ github.event.inputs.action }}' completed successfully!"
|
||||
echo "💡 You can now run the hourly sync or manual sync operations."
|
||||
@@ -0,0 +1,47 @@
|
||||
name: Hourly Energy Data Sync
|
||||
|
||||
on:
|
||||
schedule:
|
||||
# Run every hour at minute 5 (to avoid peak times)
|
||||
- cron: "5 * * * *"
|
||||
workflow_dispatch: # Allow manual triggering
|
||||
|
||||
jobs:
|
||||
sync-energy-data:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v4
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Install uv
|
||||
run: |
|
||||
curl -LsSf https://astral.sh/uv/install.sh | sh
|
||||
echo "$HOME/.cargo/bin" >> $GITHUB_PATH
|
||||
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
uv sync
|
||||
|
||||
- name: Run incremental sync
|
||||
env:
|
||||
ELOVERBLIK_API_TOKEN: ${{ secrets.ELOVERBLIK_API_TOKEN }}
|
||||
DB_HOST: ${{ secrets.DB_HOST }}
|
||||
DB_NAME: ${{ secrets.DB_NAME }}
|
||||
DB_USER: ${{ secrets.DB_USER }}
|
||||
DB_PASSWORD: ${{ secrets.DB_PASSWORD }}
|
||||
DB_PORT: ${{ secrets.DB_PORT }}
|
||||
run: |
|
||||
uv run energy-ingester sync-incremental --verbose
|
||||
|
||||
- name: Notify on failure
|
||||
if: failure()
|
||||
run: |
|
||||
echo "Energy data sync failed! Check logs and database connectivity."
|
||||
# Add notification logic here (email, Slack, etc.)
|
||||
exit 1
|
||||
@@ -0,0 +1,53 @@
|
||||
name: Manual Full Sync
|
||||
|
||||
on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
days:
|
||||
description: "Number of days to sync"
|
||||
required: true
|
||||
default: "30"
|
||||
type: string
|
||||
|
||||
jobs:
|
||||
full-sync:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v4
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Install uv
|
||||
run: |
|
||||
curl -LsSf https://astral.sh/uv/install.sh | sh
|
||||
echo "$HOME/.cargo/bin" >> $GITHUB_PATH
|
||||
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
uv sync
|
||||
|
||||
- name: Run full sync
|
||||
env:
|
||||
ELOVERBLIK_API_TOKEN: ${{ secrets.ELOVERBLIK_API_TOKEN }}
|
||||
DB_HOST: ${{ secrets.DB_HOST }}
|
||||
DB_NAME: ${{ secrets.DB_NAME }}
|
||||
DB_USER: ${{ secrets.DB_USER }}
|
||||
DB_PASSWORD: ${{ secrets.DB_PASSWORD }}
|
||||
DB_PORT: ${{ secrets.DB_PORT }}
|
||||
run: |
|
||||
uv run energy-ingester sync --days ${{ github.event.inputs.days }} --verbose
|
||||
|
||||
- name: Summary
|
||||
run: |
|
||||
echo "Full sync completed for ${{ github.event.inputs.days }} days!"
|
||||
|
||||
- name: Notify on failure
|
||||
if: failure()
|
||||
run: |
|
||||
echo "Full sync failed!"
|
||||
exit 1
|
||||
@@ -0,0 +1,13 @@
|
||||
name: Python Code Quality
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- main
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
python-code-quality:
|
||||
uses: brian/CI-templates/.gitea/workflows/python/code-quality.yml@v1.0.0
|
||||
with:
|
||||
python-version: ${{ vars.PYTHON_VERSION }}
|
||||
@@ -1,47 +0,0 @@
|
||||
name: Database Setup Automation
|
||||
|
||||
on:
|
||||
# Trigger the workflow manually from the Gitea UI
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
create_db_and_user:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
container:
|
||||
image: node:20-alpine
|
||||
|
||||
steps:
|
||||
- name: Install PostgreSQL Client
|
||||
run: |
|
||||
# Install psql client on Alpine Linux
|
||||
apk add --no-cache postgresql-client
|
||||
|
||||
- name: Checkout Repository
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Configure and Create PostgreSQL Resources
|
||||
# Set environment variables from the Gitea Secrets
|
||||
env:
|
||||
POSTGRES_HOST: ${{ secrets.POSTGRES_HOST }}
|
||||
POSTGRES_PORT: ${{ secrets.POSTGRES_PORT }}
|
||||
POSTGRES_ADMIN_USER: ${{ secrets.POSTGRES_ADMIN_USER }}
|
||||
POSTGRES_ADMIN_PASSWORD: ${{ secrets.POSTGRES_ADMIN_PASSWORD }}
|
||||
NEW_DB_NAME: ${{ secrets.NEW_DB_NAME }}
|
||||
NEW_DB_USER: ${{ secrets.NEW_DB_USER }}
|
||||
NEW_DB_PASSWORD: ${{ secrets.NEW_DB_PASSWORD }}
|
||||
|
||||
run: |
|
||||
echo "Running database creation script..."
|
||||
|
||||
# Add executable permission to the script
|
||||
chmod +x ./scripts/create_db_user.sh
|
||||
|
||||
# Export the admin password so psql can use it automatically
|
||||
export PGPASSWORD=$POSTGRES_ADMIN_PASSWORD
|
||||
|
||||
# Execute the script
|
||||
./scripts/create_db_user.sh
|
||||
|
||||
# Unset the password variable for security
|
||||
unset PGPASSWORD
|
||||
@@ -0,0 +1,40 @@
|
||||
repos:
|
||||
# General repository hygiene hooks
|
||||
- repo: https://github.com/pre-commit/pre-commit-hooks
|
||||
rev: v4.5.0
|
||||
hooks:
|
||||
- id: trailing-whitespace
|
||||
- id: end-of-file-fixer
|
||||
- id: check-yaml
|
||||
- id: check-added-large-files
|
||||
- id: debug-statements
|
||||
- id: name-tests-test
|
||||
- id: check-merge-conflict
|
||||
|
||||
# Python linting and formatting with Ruff
|
||||
- repo: https://github.com/astral-sh/ruff-pre-commit
|
||||
rev: v0.12.8
|
||||
hooks:
|
||||
- id: ruff-check
|
||||
args: [--fix]
|
||||
- id: ruff-format
|
||||
|
||||
# Static type checking with mypy
|
||||
- repo: https://github.com/pre-commit/mirrors-mypy
|
||||
rev: v1.8.0
|
||||
hooks:
|
||||
- id: mypy
|
||||
|
||||
# Python syntax modernization with pyupgrade
|
||||
- repo: https://github.com/asottile/pyupgrade
|
||||
rev: v3.20.0
|
||||
hooks:
|
||||
- id: pyupgrade
|
||||
args: ["--py312-plus"]
|
||||
|
||||
# Formatting for Markdown, JSON, and YAML with Prettier
|
||||
- repo: https://github.com/pre-commit/mirrors-prettier
|
||||
rev: v3.1.0
|
||||
hooks:
|
||||
- id: prettier
|
||||
files: "\\.(md|json|yaml|yml)$"
|
||||
@@ -0,0 +1 @@
|
||||
3.12
|
||||
@@ -0,0 +1,893 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "a2da9908",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"# Setup"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "2e95933f",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"Token request status: 200\n",
|
||||
"Token response: {\"result\":\"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ0b2tlblR5cGUiOiJDdXN0b21lckFQSV9EYXRhQWNjZXNzIiwidG9rZW5pZCI6ImE2NzY1ODRmLWY1MzktNDI0Yy04MmZhLWM5MjNhMjNhOTVlYSIsIndlYkFwcCI6IkN1c3RvbWVyQXBwIiwidmVyc2lvbiI6IjIiLCJpZGVudGl0eVRva2VuIjoiWUpoaXlqMXRjMUZiTnBtWmNaM0tjdGUrbzNzQ1FxR3B1dWtEcVh4QzVveEhKMisvcTAycTdpWGlZZURTRm9jUjVmYVBZMHVCeGpiWnYrOE52SHUycnpFU2pFRDNUajBQbmM5QzBZY1FXb0hzTDRvVHliOWRTSysrVFR5NTdDWTR4bit1dTdYN2lYdTNyV1pLUTUzTkh5T2NRSjB3Y3ZtWURHUjVDT3lITW14WDZhMjFEUHdBOWFsRk53RlFXbGkxT0NuTnp3QUdQaXJZWnZLcTYxNW52aFBjczZaN0FWUlh5MnRqMTFPbWpuR1NQVFJYU2pvdjdpRkhZcmhTdTRiTFpDQXhDd0FHWHl4a0tPWENRRDZFUVdTUU92d0ZpMHdnSUZwVHBIQjJKaHlEVFRvM2FYU1RaSnh2MEpFZktvalNLWXUxSE54enpWcTE3NmVWei9UVW5JZTdUeTc0LzZNekFqZmplcWpFUTk3aXcvbnRwTFNzdGkwakZqdjB6V1hKRXRKMVNQYURmbG5ZdG9DcEtnZmdCVDc4RTNKcXVIZSszQjJLanUwL0xlUU4rMXpHU2cxeEhuUFZOdlcxanRabjhWbjhIaVVmNUhUK0tPZ3ZDNitqaWJQRWRwNXJvaWtSNkFjRzZHcS8xOThkL2d6bEo4MzJMVWVnR1BiTlBHVVBwZ2VQTmFtUlRSWXpnSk9GdmwzV0NGb1NXMUZhYU5UalBJL1F4Wjh4L05VVzlJZXFXbG1rb1VrMVhIS3lWSzBjdW1KWXlaelJBNHJKdGJ0QXVaaGdMUCtpUmtSUzlFMkcwOWhDQnhzOHJKWjRycWVvZmVITC9iQkQwYXIzd25tREFpUks3L1ovNmkvbXFJMEZHa3NUcXdYZzdBTTY0SldGQkNUcm1ZMk9hSlIwRHpSZmd3ekN5Y0NzRHRRUW1RNTZwNWVPRzJyRnlUUHh5YjEwS3p4a3FTbTZVNDExcEhlVllFbnR4cVloYUg4and2RWJxVFJBNEZ5VUZtQVVxQmhIVDFWNWpsRkFZL0U3bjUwejAxSE5WL21KNzVBcGNqektsaUx4YVBuQXdYUG43NmVVbHc4NUt4anJueEh5d3V6a00wVFhDSVRIZ0NrWmdYbkdxVEYwZE1TL3VhRE9Tbjl5NTY5SE9vb1FxM3VNZ0prV2RFaUdSMkUvcVErbFVNS05kVlNEL0F6Y0tLZkdhdkZUTko2WVhZOVozKzZKN2lpMWFDbGxoWFAwQ1VRRzhaOVpaMEMrU3JlN3FNSzVvaEpwY2FDOUdoZ1hVNklqZ0xZd2JTM0RGOUN3NGt3cmxHRVJSNURVZVNXREg2aFp5SVdEWWc2d1k4bHMxNGVlVjQ4bW5aTVVCSVE4cDZvNzhGSzFLUWc2V3BQZ3NqMGVOcXdBVW1HemNQcUs5UHJsdDc5VmZvK2pQSDlkUW1IaUJIbjFDcjROWWJpUEI4bllNN09JdFJpb3lHVmpyTi9lVjR1SGloc1c1ZzUxMG9pMjVxNzU0Y2VPbW10ck0ra0VzRGtoVEd1UzNxZml5ZUhqRDV0dk1leXU3eGs0c1gyeDdhVkdDeGV0eGlDalNmWkZ0R2JiWmY3VWhKWTAzNzBZbDYvZFRNeDJ4VFNlUzhqOUgwZUZJT1RBL3N3ZVNXUjdDMlc2MitWOC9SNERZT3lOdjJzWmRib2RIN3hKdk0yUnNxcHFRMUtqS1VsclpEUmRTMFk4ZDZTOGJuTTBGUC9tRXlDZkF1MmthRnE5d0hQcGZUdzJ2QVFIeTdMNnR5ZjNoSUNtOHQ1eGh4NEJwV1ZUUkVtQUVXZk5KQ3c2QXZlUFExVHY5TjVmQ1N6NXo5RlU2VmhxcUlQL1hWM2l3dUFQc3dVc2hpaG90ZTBPR2YrZlpyMEZHY04vSTJXYzZneDdIT3RYVitUWEdieGdvY2lianFWRTFyQTRweTJsYnpSQlo3dXVRc29OcW1YQlVkckF6U0V3MkFEZGg3cE1KWGpBRXF0cHZKOXluWUlqWWRpZXdocllEazRRNCtEekl5V01NRk5YOGVyakFER21PYnJvazB6aUtGd1Fwb3hCL2I2SU9GYzVwWjFvUm0zY0dNUjU2M3BPdmw0aTNIalhMbmxvbWkxQ29Hb1JPL203MGp2RXRMdXBvOGRRSGpFZXlEUzBmeGl1b0hHRWV1Uk12SGpjcVUranBadjloUjEwMXh1QVVGRjNaUFJ6NnhQb2JsZVR0N0xnSjZWMm9KZmJ4QmpPWEdLTEthdGNiUEkzMS84MWtSWlIyN0w5eVZ4NkRLays1ZjV4Zk5JdUxWd0pzeDNBRVBob2o4SERPVFRKSERMUlZpa1hOdGE4OTNZOWVaZ2xpRzdyMXZCaXRTZ2dwQTRlbDhjMHlIdGJDaG9Vb3N6SmsyMHZJWGphem1zRG5kTU5DVXluRGlWbEN4VDBROWRPMlc4Rlp5NERxRmxNN1BQVVZWZzRhMkRIUzhianZQR3dWZ0R6WDkrVzA3VEMrVVlpZ3k2Sk93YnkwMThqd2loQjlLb3d6M0Vqc1djTjF1RGRsYXZPbHNWbHFBU05tWUF0MmpnbVF1YnQzTUlrdFhObGNOMWl0MnA5NFJ3UE9qNmhjYm1OM3ZRaXNvVExyd1llK1RlciIsImh0dHA6Ly9zY2hlbWFzLnhtbHNvYXAub3JnL3dzLzIwMDUvMDUvaWRlbnRpdHkvY2xhaW1zL25hbWVpZGVudGlmaWVyIjoiTUlEOmUyZjE2NDE4LTlmZDItNGFkZi1hNDA2LWEwMDQ5ZTZjYzU4YSIsImh0dHA6Ly9zY2hlbWFzLnhtbHNvYXAub3JnL3dzLzIwMDUvMDUvaWRlbnRpdHkvY2xhaW1zL2dpdmVubmFtZSI6IkJyaWFuIEJqYXJrZSBKZW5zZW4iLCJsb2dpblR5cGUiOiJLZXlDYXJkIiwiYjNmIjoiRjRSQ1VHSjdoTUUyMlpUVzRnTGZha2crL0VXL2Y4MmZVVTFoNVFjMk1DWT0iLCJwaWQiOiJQSUQ6OTIwOC0yMDAyLTItNjAwNTA4Mzg2NzY4IiwidXNlcklkIjoiMTA0MzQ1OSIsImV4cCI6MTc2MTY0NzgxNywiaXNzIjoiRW5lcmdpbmV0IiwianRpIjoiYTY3NjU4NGYtZjUzOS00MjRjLTgyZmEtYzkyM2EyM2E5NWVhIiwidG9rZW5OYW1lIjoiZ2l0ZWEtYWNjZXNzLXRva2VuIiwiYXVkIjoiRW5lcmdpbmV0In0.zmb-Oy1Jn6u16dAodmdd3OApFbUVcTjetM5Z8x_evxQ\"}\n",
|
||||
"Successfully obtained access token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ0b2tlblR5c...\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"import os\n",
|
||||
"from dotenv import load_dotenv\n",
|
||||
"import http.client\n",
|
||||
"import json\n",
|
||||
"from pydantic import AnyUrl\n",
|
||||
"\n",
|
||||
"load_dotenv()\n",
|
||||
"ELOVERBLIK_API_TOKEN = os.getenv(\"ELOVERBLIK_API_TOKEN\")\n",
|
||||
"\n",
|
||||
"\n",
|
||||
"server_url = AnyUrl(\"https://api.eloverblik.dk\")\n",
|
||||
"\n",
|
||||
"# Prepare JWT authentication\n",
|
||||
"\n",
|
||||
"conn = http.client.HTTPSConnection(server_url.host)\n",
|
||||
"\n",
|
||||
"headers = {\n",
|
||||
" \"Authorization\": f\"Bearer {ELOVERBLIK_API_TOKEN}\",\n",
|
||||
" \"Content-Type\": \"application/json\",\n",
|
||||
" \"api-version\": \"1.0\",\n",
|
||||
"}\n",
|
||||
"\n",
|
||||
"conn.request(\"GET\", \"/customerapi/api/token\", headers=headers)\n",
|
||||
"\n",
|
||||
"res = conn.getresponse()\n",
|
||||
"print(f\"Token request status: {res.status}\")\n",
|
||||
"\n",
|
||||
"if res.status == 200:\n",
|
||||
" token_response = res.read().decode(\"utf-8\")\n",
|
||||
" print(f\"Token response: {token_response[:50]}...\")\n",
|
||||
"\n",
|
||||
" # Parse the response to get the access token\n",
|
||||
" try:\n",
|
||||
" token_data = json.loads(token_response)\n",
|
||||
" access_token = token_data[\"result\"]\n",
|
||||
" print(f\"Successfully obtained access token: {access_token[:50]}...\")\n",
|
||||
" except Exception as e:\n",
|
||||
" print(f\"Error parsing token response: {e}\")\n",
|
||||
" access_token = None\n",
|
||||
"else:\n",
|
||||
" error_data = res.read().decode(\"utf-8\")\n",
|
||||
" print(f\"Error getting token: {error_data}\")\n",
|
||||
" access_token = None"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 33,
|
||||
"id": "0a68b6bd",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"Testing access token with metering points API...\n",
|
||||
"Metering points API status: 200\n",
|
||||
"Response: {\"result\":[{\"streetCode\":\"2724\",\"streetName\":\"Vejl...\n",
|
||||
"Metering points API status: 200\n",
|
||||
"Response: {\"result\":[{\"streetCode\":\"2724\",\"streetName\":\"Vejl...\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"if not access_token:\n",
|
||||
" print(\"No access token available to test with\")\n",
|
||||
"else:\n",
|
||||
" print(\"Testing access token with metering points API...\")\n",
|
||||
"\n",
|
||||
" # Create new connection for API calls\n",
|
||||
" conn = http.client.HTTPSConnection(server_url.host)\n",
|
||||
"\n",
|
||||
" # Use the access token for API calls\n",
|
||||
" api_headers = {\n",
|
||||
" \"Authorization\": f\"Bearer {access_token}\",\n",
|
||||
" \"Content-Type\": \"application/json\",\n",
|
||||
" \"api-version\": \"1.0\",\n",
|
||||
" }\n",
|
||||
"\n",
|
||||
" # Get metering points\n",
|
||||
" from urllib.parse import urlencode\n",
|
||||
"\n",
|
||||
" params = {\n",
|
||||
" \"includeAll\": True,\n",
|
||||
" }\n",
|
||||
"\n",
|
||||
" conn.request(\n",
|
||||
" \"GET\",\n",
|
||||
" \"/customerapi/api/meteringpoints/meteringpoints?\" + urlencode(params),\n",
|
||||
" headers=api_headers,\n",
|
||||
" )\n",
|
||||
"\n",
|
||||
" res = conn.getresponse()\n",
|
||||
" print(f\"Metering points API status: {res.status}\")\n",
|
||||
"\n",
|
||||
" if res.status != 200:\n",
|
||||
" error_data = res.read().decode(\"utf-8\")\n",
|
||||
" print(f\"Error: {error_data}\")\n",
|
||||
" else:\n",
|
||||
" data = res.read().decode(\"utf-8\")\n",
|
||||
" print(f\"Response: {data[:50] + '...' if len(data) > 50 else data}\")"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 34,
|
||||
"id": "686fe407",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"Using metering point ID: 571313124600282119\n",
|
||||
"Requesting consumption data from 2025-10-20 to 2025-10-27\n",
|
||||
"Consumption data API status: 200\n",
|
||||
"Success! Consumption data retrieved:\n",
|
||||
"{\n",
|
||||
" \"result\": [\n",
|
||||
" {\n",
|
||||
" \"MyEnergyData_MarketDocument\": {\n",
|
||||
" \"mRID\": \"0HNGJHUMDF3GS:000001A6\",\n",
|
||||
" \"createdDateTime\": \"2025-10-27T11:02:04Z\",\n",
|
||||
" \"sender_MarketParticipant.name\": \"\",\n",
|
||||
" \"sender_MarketParticipant.mRID\": {\n",
|
||||
" \"codingScheme\": null,\n",
|
||||
" \"name\": null\n",
|
||||
" },\n",
|
||||
" \"period.timeInterval\": {\n",
|
||||
" \"start\": \"2025-10-19T22:00:00Z\",\n",
|
||||
" \"end\": \"2025-10-26T23:00:00Z\"\n",
|
||||
" },\n",
|
||||
" \"TimeSeries\": [\n",
|
||||
" {\n",
|
||||
" \"mRID\": \"571313124600282119\",\n",
|
||||
" \"businessType\": \"A04\",\n",
|
||||
" \"curveType\": \"A01\",\n",
|
||||
" \"measurement_Unit.name\": \"KWH\",\n",
|
||||
" \"MarketEvaluationPoint\": {\n",
|
||||
" \"mRID\": {\n",
|
||||
" \"codingScheme\": \"A10\",\n",
|
||||
" \"name\": \"571313124600282119\"\n",
|
||||
" }\n",
|
||||
" },\n",
|
||||
" \"Period\": [\n",
|
||||
" {\n",
|
||||
" \"resolution\": \"PT1H\",\n",
|
||||
" \"timeInterval\": {\n",
|
||||
" \"start\": \"2025-10-19T22:00:00Z\",\n",
|
||||
" \"end\": \"2025-10-2...\n",
|
||||
"Consumption data API status: 200\n",
|
||||
"Success! Consumption data retrieved:\n",
|
||||
"{\n",
|
||||
" \"result\": [\n",
|
||||
" {\n",
|
||||
" \"MyEnergyData_MarketDocument\": {\n",
|
||||
" \"mRID\": \"0HNGJHUMDF3GS:000001A6\",\n",
|
||||
" \"createdDateTime\": \"2025-10-27T11:02:04Z\",\n",
|
||||
" \"sender_MarketParticipant.name\": \"\",\n",
|
||||
" \"sender_MarketParticipant.mRID\": {\n",
|
||||
" \"codingScheme\": null,\n",
|
||||
" \"name\": null\n",
|
||||
" },\n",
|
||||
" \"period.timeInterval\": {\n",
|
||||
" \"start\": \"2025-10-19T22:00:00Z\",\n",
|
||||
" \"end\": \"2025-10-26T23:00:00Z\"\n",
|
||||
" },\n",
|
||||
" \"TimeSeries\": [\n",
|
||||
" {\n",
|
||||
" \"mRID\": \"571313124600282119\",\n",
|
||||
" \"businessType\": \"A04\",\n",
|
||||
" \"curveType\": \"A01\",\n",
|
||||
" \"measurement_Unit.name\": \"KWH\",\n",
|
||||
" \"MarketEvaluationPoint\": {\n",
|
||||
" \"mRID\": {\n",
|
||||
" \"codingScheme\": \"A10\",\n",
|
||||
" \"name\": \"571313124600282119\"\n",
|
||||
" }\n",
|
||||
" },\n",
|
||||
" \"Period\": [\n",
|
||||
" {\n",
|
||||
" \"resolution\": \"PT1H\",\n",
|
||||
" \"timeInterval\": {\n",
|
||||
" \"start\": \"2025-10-19T22:00:00Z\",\n",
|
||||
" \"end\": \"2025-10-2...\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# Get consumption data for a specific metering point ID\n",
|
||||
"\n",
|
||||
"if not access_token:\n",
|
||||
" print(\"No access token available\")\n",
|
||||
"else:\n",
|
||||
" # First, let's get the metering points data properly and extract the metering point ID\n",
|
||||
" conn = http.client.HTTPSConnection(server_url.host)\n",
|
||||
"\n",
|
||||
" api_headers = {\n",
|
||||
" \"Authorization\": f\"Bearer {access_token}\",\n",
|
||||
" \"Content-Type\": \"application/json\",\n",
|
||||
" \"api-version\": \"1.0\",\n",
|
||||
" }\n",
|
||||
"\n",
|
||||
" # Get metering points\n",
|
||||
" from urllib.parse import urlencode\n",
|
||||
"\n",
|
||||
" params = {\"includeAll\": True}\n",
|
||||
"\n",
|
||||
" conn.request(\n",
|
||||
" \"GET\",\n",
|
||||
" \"/customerapi/api/meteringpoints/meteringpoints?\" + urlencode(params),\n",
|
||||
" headers=api_headers,\n",
|
||||
" )\n",
|
||||
" res = conn.getresponse()\n",
|
||||
"\n",
|
||||
" if res.status == 200:\n",
|
||||
" metering_data = res.read().decode(\"utf-8\")\n",
|
||||
" metering_points = json.loads(metering_data)\n",
|
||||
"\n",
|
||||
" if metering_points[\"result\"]:\n",
|
||||
" # Get the first metering point ID\n",
|
||||
" metering_point_id = metering_points[\"result\"][0][\"meteringPointId\"]\n",
|
||||
" print(f\"Using metering point ID: {metering_point_id}\")\n",
|
||||
"\n",
|
||||
" # Now get consumption data for this metering point\n",
|
||||
" # Let's get data from the last 7 days\n",
|
||||
" from datetime import datetime, timedelta\n",
|
||||
"\n",
|
||||
" end_date = datetime.now()\n",
|
||||
" start_date = end_date - timedelta(days=7)\n",
|
||||
"\n",
|
||||
" # Format dates as required by the API (YYYY-MM-DD)\n",
|
||||
" date_from = start_date.strftime(\"%Y-%m-%d\")\n",
|
||||
" date_to = end_date.strftime(\"%Y-%m-%d\")\n",
|
||||
"\n",
|
||||
" print(f\"Requesting consumption data from {date_from} to {date_to}\")\n",
|
||||
"\n",
|
||||
" # Prepare the request body for consumption data\n",
|
||||
" consumption_request = {\n",
|
||||
" \"meteringPoints\": {\"meteringPoint\": [metering_point_id]}\n",
|
||||
" }\n",
|
||||
"\n",
|
||||
" # Create new connection for consumption data request\n",
|
||||
" conn = http.client.HTTPSConnection(server_url.host)\n",
|
||||
"\n",
|
||||
" # Make the consumption data request\n",
|
||||
" conn.request(\n",
|
||||
" \"POST\",\n",
|
||||
" f\"/customerapi/api/meterdata/gettimeseries/{date_from}/{date_to}/Hour\",\n",
|
||||
" body=json.dumps(consumption_request),\n",
|
||||
" headers=api_headers,\n",
|
||||
" )\n",
|
||||
"\n",
|
||||
" res = conn.getresponse()\n",
|
||||
" print(f\"Consumption data API status: {res.status}\")\n",
|
||||
"\n",
|
||||
" if res.status == 200:\n",
|
||||
" consumption_data = res.read().decode(\"utf-8\")\n",
|
||||
" consumption_json = json.loads(consumption_data)\n",
|
||||
" print(\"Success! Consumption data retrieved:\")\n",
|
||||
" print(\n",
|
||||
" json.dumps(consumption_json, indent=2)[:1000] + \"...\"\n",
|
||||
" if len(json.dumps(consumption_json, indent=2)) > 1000\n",
|
||||
" else json.dumps(consumption_json, indent=2)\n",
|
||||
" )\n",
|
||||
" else:\n",
|
||||
" error_data = res.read().decode(\"utf-8\")\n",
|
||||
" print(f\"Error getting consumption data: {error_data}\")\n",
|
||||
" else:\n",
|
||||
" print(\"No metering points found\")\n",
|
||||
" else:\n",
|
||||
" error_data = res.read().decode(\"utf-8\")\n",
|
||||
" print(f\"Error getting metering points: {error_data}\")"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 35,
|
||||
"id": "97359b7b",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"Metering Point ID: 571313124600282119\n",
|
||||
"Unit: KWH\n",
|
||||
"Data Period: 2025-10-19T22:00:00Z to 2025-10-26T23:00:00Z\n",
|
||||
"\n",
|
||||
"============================================================\n",
|
||||
"CONSUMPTION DATA:\n",
|
||||
"============================================================\n",
|
||||
"\n",
|
||||
"Period starting: 2025-10-19T22:00:00Z\n",
|
||||
"Resolution: PT1H\n",
|
||||
"Number of hourly readings: 24\n",
|
||||
" Hour 1: 0.53 kWh (Quality: A04)\n",
|
||||
" Hour 2: 0.55 kWh (Quality: A04)\n",
|
||||
" Hour 3: 0.51 kWh (Quality: A04)\n",
|
||||
" Hour 4: 0.5 kWh (Quality: A04)\n",
|
||||
" Hour 5: 0.6 kWh (Quality: A04)\n",
|
||||
" Hour 6: 0.52 kWh (Quality: A04)\n",
|
||||
" Hour 7: 0.68 kWh (Quality: A04)\n",
|
||||
" Hour 8: 0.69 kWh (Quality: A04)\n",
|
||||
" Hour 9: 0.62 kWh (Quality: A04)\n",
|
||||
" Hour 10: 0.62 kWh (Quality: A04)\n",
|
||||
" ... and 14 more hourly readings\n",
|
||||
"\n",
|
||||
"Period starting: 2025-10-20T22:00:00Z\n",
|
||||
"Resolution: PT1H\n",
|
||||
"Number of hourly readings: 24\n",
|
||||
" Hour 1: 0.58 kWh (Quality: A04)\n",
|
||||
" Hour 2: 0.56 kWh (Quality: A04)\n",
|
||||
" Hour 3: 0.57 kWh (Quality: A04)\n",
|
||||
" Hour 4: 0.61 kWh (Quality: A04)\n",
|
||||
" Hour 5: 0.64 kWh (Quality: A04)\n",
|
||||
" Hour 6: 0.58 kWh (Quality: A04)\n",
|
||||
" Hour 7: 0.62 kWh (Quality: A04)\n",
|
||||
" Hour 8: 0.53 kWh (Quality: A04)\n",
|
||||
" Hour 9: 0.58 kWh (Quality: A04)\n",
|
||||
" Hour 10: 0.53 kWh (Quality: A04)\n",
|
||||
" ... and 14 more hourly readings\n",
|
||||
"\n",
|
||||
"Period starting: 2025-10-21T22:00:00Z\n",
|
||||
"Resolution: PT1H\n",
|
||||
"Number of hourly readings: 24\n",
|
||||
" Hour 1: 0.62 kWh (Quality: A04)\n",
|
||||
" Hour 2: 0.62 kWh (Quality: A04)\n",
|
||||
" Hour 3: 0.5 kWh (Quality: A04)\n",
|
||||
" Hour 4: 0.59 kWh (Quality: A04)\n",
|
||||
" Hour 5: 0.57 kWh (Quality: A04)\n",
|
||||
" Hour 6: 0.58 kWh (Quality: A04)\n",
|
||||
" Hour 7: 0.55 kWh (Quality: A04)\n",
|
||||
" Hour 8: 0.69 kWh (Quality: A04)\n",
|
||||
" Hour 9: 0.77 kWh (Quality: A04)\n",
|
||||
" Hour 10: 1.0 kWh (Quality: A04)\n",
|
||||
" ... and 14 more hourly readings\n",
|
||||
"\n",
|
||||
"Period starting: 2025-10-22T22:00:00Z\n",
|
||||
"Resolution: PT1H\n",
|
||||
"Number of hourly readings: 24\n",
|
||||
" Hour 1: 0.59 kWh (Quality: A04)\n",
|
||||
" Hour 2: 0.68 kWh (Quality: A04)\n",
|
||||
" Hour 3: 0.6 kWh (Quality: A04)\n",
|
||||
" Hour 4: 0.6 kWh (Quality: A04)\n",
|
||||
" Hour 5: 0.57 kWh (Quality: A04)\n",
|
||||
" Hour 6: 0.57 kWh (Quality: A04)\n",
|
||||
" Hour 7: 0.57 kWh (Quality: A04)\n",
|
||||
" Hour 8: 0.57 kWh (Quality: A04)\n",
|
||||
" Hour 9: 0.59 kWh (Quality: A04)\n",
|
||||
" Hour 10: 0.58 kWh (Quality: A04)\n",
|
||||
" ... and 14 more hourly readings\n",
|
||||
"\n",
|
||||
"Period starting: 2025-10-23T22:00:00Z\n",
|
||||
"Resolution: PT1H\n",
|
||||
"Number of hourly readings: 24\n",
|
||||
" Hour 1: 0.58 kWh (Quality: A04)\n",
|
||||
" Hour 2: 0.72 kWh (Quality: A04)\n",
|
||||
" Hour 3: 0.61 kWh (Quality: A04)\n",
|
||||
" Hour 4: 0.6 kWh (Quality: A04)\n",
|
||||
" Hour 5: 0.54 kWh (Quality: A04)\n",
|
||||
" Hour 6: 0.53 kWh (Quality: A04)\n",
|
||||
" Hour 7: 0.49 kWh (Quality: A04)\n",
|
||||
" Hour 8: 0.8 kWh (Quality: A04)\n",
|
||||
" Hour 9: 0.62 kWh (Quality: A04)\n",
|
||||
" Hour 10: 1.21 kWh (Quality: A04)\n",
|
||||
" ... and 14 more hourly readings\n",
|
||||
"\n",
|
||||
"Period starting: 2025-10-24T22:00:00Z\n",
|
||||
"Resolution: PT1H\n",
|
||||
"Number of hourly readings: 24\n",
|
||||
" Hour 1: 0.89 kWh (Quality: A04)\n",
|
||||
" Hour 2: 1.15 kWh (Quality: A04)\n",
|
||||
" Hour 3: 0.58 kWh (Quality: A04)\n",
|
||||
" Hour 4: 0.58 kWh (Quality: A04)\n",
|
||||
" Hour 5: 0.62 kWh (Quality: A04)\n",
|
||||
" Hour 6: 0.61 kWh (Quality: A04)\n",
|
||||
" Hour 7: 0.58 kWh (Quality: A04)\n",
|
||||
" Hour 8: 0.56 kWh (Quality: A04)\n",
|
||||
" Hour 9: 0.53 kWh (Quality: A04)\n",
|
||||
" Hour 10: 0.61 kWh (Quality: A04)\n",
|
||||
" ... and 14 more hourly readings\n",
|
||||
"\n",
|
||||
"Period starting: 2025-10-25T22:00:00Z\n",
|
||||
"Resolution: PT1H\n",
|
||||
"Number of hourly readings: 25\n",
|
||||
" Hour 1: 0.63 kWh (Quality: A04)\n",
|
||||
" Hour 2: 0.51 kWh (Quality: A04)\n",
|
||||
" Hour 3: 0.55 kWh (Quality: A04)\n",
|
||||
" Hour 4: 0.56 kWh (Quality: A04)\n",
|
||||
" Hour 5: 0.56 kWh (Quality: A04)\n",
|
||||
" Hour 6: 0.5 kWh (Quality: A04)\n",
|
||||
" Hour 7: 0.55 kWh (Quality: A04)\n",
|
||||
" Hour 8: 0.56 kWh (Quality: A04)\n",
|
||||
" Hour 9: 0.55 kWh (Quality: A04)\n",
|
||||
" Hour 10: 0.54 kWh (Quality: A04)\n",
|
||||
" ... and 15 more hourly readings\n",
|
||||
"\n",
|
||||
"============================================================\n",
|
||||
"SUMMARY:\n",
|
||||
"Total consumption over 169 hours: 138.31 kWh\n",
|
||||
"Average hourly consumption: 0.818 kWh\n",
|
||||
"============================================================\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# Parse and display the consumption data in a more readable format\n",
|
||||
"\n",
|
||||
"if \"consumption_json\" in locals() and consumption_json:\n",
|
||||
" try:\n",
|
||||
" # Extract the time series data\n",
|
||||
" market_document = consumption_json[\"result\"][0][\"MyEnergyData_MarketDocument\"]\n",
|
||||
" time_series = market_document[\"TimeSeries\"][0]\n",
|
||||
" periods = time_series[\"Period\"]\n",
|
||||
"\n",
|
||||
" print(f\"Metering Point ID: {time_series['mRID']}\")\n",
|
||||
" print(f\"Unit: {time_series['measurement_Unit.name']}\")\n",
|
||||
" print(\n",
|
||||
" f\"Data Period: {market_document['period.timeInterval']['start']} to {market_document['period.timeInterval']['end']}\"\n",
|
||||
" )\n",
|
||||
" print(\"\\n\" + \"=\" * 60)\n",
|
||||
" print(\"CONSUMPTION DATA:\")\n",
|
||||
" print(\"=\" * 60)\n",
|
||||
"\n",
|
||||
" total_consumption = 0\n",
|
||||
" data_points = 0\n",
|
||||
"\n",
|
||||
" for period in periods:\n",
|
||||
" period_start = period[\"timeInterval\"][\"start\"]\n",
|
||||
" print(f\"\\nPeriod starting: {period_start}\")\n",
|
||||
" print(f\"Resolution: {period['resolution']}\")\n",
|
||||
"\n",
|
||||
" if \"Point\" in period:\n",
|
||||
" points = period[\"Point\"]\n",
|
||||
" print(f\"Number of hourly readings: {len(points)}\")\n",
|
||||
"\n",
|
||||
" for point in points[:10]: # Show first 10 points\n",
|
||||
" position = point[\"position\"]\n",
|
||||
" quantity = float(point[\"out_Quantity.quantity\"])\n",
|
||||
" quality = point[\"out_Quantity.quality\"]\n",
|
||||
"\n",
|
||||
" print(f\" Hour {position}: {quantity} kWh (Quality: {quality})\")\n",
|
||||
" total_consumption += quantity\n",
|
||||
" data_points += 1\n",
|
||||
"\n",
|
||||
" if len(points) > 10:\n",
|
||||
" print(f\" ... and {len(points) - 10} more hourly readings\")\n",
|
||||
" # Add remaining consumption to total\n",
|
||||
" for point in points[10:]:\n",
|
||||
" total_consumption += float(point[\"out_Quantity.quantity\"])\n",
|
||||
" data_points += 1\n",
|
||||
"\n",
|
||||
" print(\"\\n\" + \"=\" * 60)\n",
|
||||
" print(\"SUMMARY:\")\n",
|
||||
" print(\n",
|
||||
" f\"Total consumption over {data_points} hours: {total_consumption:.2f} kWh\"\n",
|
||||
" )\n",
|
||||
" print(f\"Average hourly consumption: {total_consumption / data_points:.3f} kWh\")\n",
|
||||
" print(\"=\" * 60)\n",
|
||||
"\n",
|
||||
" except Exception as e:\n",
|
||||
" print(f\"Error parsing consumption data: {e}\")\n",
|
||||
" print(\"Raw data structure:\")\n",
|
||||
" print(json.dumps(consumption_json, indent=2)[:500] + \"...\")\n",
|
||||
"else:\n",
|
||||
" print(\"No consumption data available to parse\")"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "2bf461e8",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"# PostgreSQL Database Design\n",
|
||||
"\n",
|
||||
"For storing energy consumption data efficiently, we'll design a normalized database schema with the following tables:\n",
|
||||
"\n",
|
||||
"## Tables Structure:\n",
|
||||
"\n",
|
||||
"### 1. `metering_points` - Store meter information\n",
|
||||
"- `id` (Primary Key)\n",
|
||||
"- `metering_point_id` (Unique identifier from API)\n",
|
||||
"- `street_name`, `building_number`, `city` etc.\n",
|
||||
"- `meter_number`\n",
|
||||
"- `consumer_name`\n",
|
||||
"- `created_at`, `updated_at`\n",
|
||||
"\n",
|
||||
"### 2. `consumption_readings` - Store hourly consumption data\n",
|
||||
"- `id` (Primary Key)\n",
|
||||
"- `metering_point_id` (Foreign Key)\n",
|
||||
"- `timestamp` (UTC timestamp for the reading)\n",
|
||||
"- `consumption_kwh` (Energy consumption in kWh)\n",
|
||||
"- `quality` (Data quality indicator)\n",
|
||||
"- `period_resolution` (e.g., 'PT1H' for hourly)\n",
|
||||
"- `created_at`\n",
|
||||
"\n",
|
||||
"### Benefits:\n",
|
||||
"- **Normalized**: Avoids data duplication\n",
|
||||
"- **Indexed**: Fast queries on timestamp and metering_point_id\n",
|
||||
"- **Scalable**: Can handle multiple meters and large time series data\n",
|
||||
"- **Flexible**: Easy to add new meters or extend with additional fields"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "969c7953",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"# SQL Schema for PostgreSQL Database\n",
|
||||
"\n",
|
||||
"create_tables_sql = \"\"\"\n",
|
||||
"-- Create metering_points table\n",
|
||||
"CREATE TABLE IF NOT EXISTS metering_points (\n",
|
||||
" id SERIAL PRIMARY KEY,\n",
|
||||
" metering_point_id VARCHAR(50) UNIQUE NOT NULL,\n",
|
||||
" street_code VARCHAR(10),\n",
|
||||
" street_name VARCHAR(255),\n",
|
||||
" building_number VARCHAR(20),\n",
|
||||
" floor_id VARCHAR(10),\n",
|
||||
" room_id VARCHAR(10),\n",
|
||||
" city_subdivision_name VARCHAR(255),\n",
|
||||
" municipality_code VARCHAR(10),\n",
|
||||
" location_description TEXT,\n",
|
||||
" settlement_method VARCHAR(10),\n",
|
||||
" meter_reading_occurrence VARCHAR(20),\n",
|
||||
" first_consumer_party_name VARCHAR(255),\n",
|
||||
" second_consumer_party_name VARCHAR(255),\n",
|
||||
" meter_number VARCHAR(50),\n",
|
||||
" consumer_start_date TIMESTAMP WITH TIME ZONE,\n",
|
||||
" type_of_mp VARCHAR(10),\n",
|
||||
" balance_supplier_name VARCHAR(255),\n",
|
||||
" postcode VARCHAR(10),\n",
|
||||
" created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,\n",
|
||||
" updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP\n",
|
||||
");\n",
|
||||
"\n",
|
||||
"-- Create consumption_readings table\n",
|
||||
"CREATE TABLE IF NOT EXISTS consumption_readings (\n",
|
||||
" id SERIAL PRIMARY KEY,\n",
|
||||
" metering_point_id VARCHAR(50) REFERENCES metering_points(metering_point_id),\n",
|
||||
" timestamp TIMESTAMP WITH TIME ZONE NOT NULL,\n",
|
||||
" consumption_kwh DECIMAL(10, 3) NOT NULL,\n",
|
||||
" quality VARCHAR(10),\n",
|
||||
" period_resolution VARCHAR(10) DEFAULT 'PT1H',\n",
|
||||
" created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,\n",
|
||||
" UNIQUE(metering_point_id, timestamp)\n",
|
||||
");\n",
|
||||
"\n",
|
||||
"-- Create indexes for performance\n",
|
||||
"CREATE INDEX IF NOT EXISTS idx_consumption_metering_point ON consumption_readings(metering_point_id);\n",
|
||||
"CREATE INDEX IF NOT EXISTS idx_consumption_timestamp ON consumption_readings(timestamp);\n",
|
||||
"CREATE INDEX IF NOT EXISTS idx_consumption_metering_point_timestamp ON consumption_readings(metering_point_id, timestamp);\n",
|
||||
"\n",
|
||||
"-- Create a function to update the updated_at timestamp\n",
|
||||
"CREATE OR REPLACE FUNCTION update_updated_at_column()\n",
|
||||
"RETURNS TRIGGER AS $$\n",
|
||||
"BEGIN\n",
|
||||
" NEW.updated_at = CURRENT_TIMESTAMP;\n",
|
||||
" RETURN NEW;\n",
|
||||
"END;\n",
|
||||
"$$ language 'plpgsql';\n",
|
||||
"\n",
|
||||
"-- Create trigger for metering_points\n",
|
||||
"CREATE TRIGGER update_metering_points_updated_at \n",
|
||||
" BEFORE UPDATE ON metering_points \n",
|
||||
" FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();\n",
|
||||
"\"\"\"\n",
|
||||
"\n",
|
||||
"print(\"PostgreSQL Schema:\")\n",
|
||||
"print(create_tables_sql)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "b8159a41",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"# Python code to store data in PostgreSQL using psycopg2\n",
|
||||
"\n",
|
||||
"import psycopg2\n",
|
||||
"from datetime import datetime\n",
|
||||
"\n",
|
||||
"# Database connection configuration\n",
|
||||
"DB_CONFIG = {\n",
|
||||
" \"host\": \"localhost\",\n",
|
||||
" \"database\": \"energy_consumption\",\n",
|
||||
" \"user\": \"your_username\",\n",
|
||||
" \"password\": \"your_password\",\n",
|
||||
" \"port\": 5432,\n",
|
||||
"}\n",
|
||||
"\n",
|
||||
"\n",
|
||||
"class EnergyDataStorage:\n",
|
||||
" def __init__(self, db_config):\n",
|
||||
" self.db_config = db_config\n",
|
||||
" self.connection = None\n",
|
||||
"\n",
|
||||
" def connect(self):\n",
|
||||
" \"\"\"Establish database connection\"\"\"\n",
|
||||
" try:\n",
|
||||
" self.connection = psycopg2.connect(**self.db_config)\n",
|
||||
" print(\"Database connection established\")\n",
|
||||
" return True\n",
|
||||
" except Exception as e:\n",
|
||||
" print(f\"Error connecting to database: {e}\")\n",
|
||||
" return False\n",
|
||||
"\n",
|
||||
" def create_tables(self):\n",
|
||||
" \"\"\"Create database tables if they don't exist\"\"\"\n",
|
||||
" if not self.connection:\n",
|
||||
" print(\"No database connection\")\n",
|
||||
" return False\n",
|
||||
"\n",
|
||||
" try:\n",
|
||||
" with self.connection.cursor() as cursor:\n",
|
||||
" cursor.execute(create_tables_sql)\n",
|
||||
" self.connection.commit()\n",
|
||||
" print(\"Tables created successfully\")\n",
|
||||
" return True\n",
|
||||
" except Exception as e:\n",
|
||||
" print(f\"Error creating tables: {e}\")\n",
|
||||
" self.connection.rollback()\n",
|
||||
" return False\n",
|
||||
"\n",
|
||||
" def store_metering_point(self, metering_point_data):\n",
|
||||
" \"\"\"Store or update metering point information\"\"\"\n",
|
||||
" if not self.connection:\n",
|
||||
" print(\"No database connection\")\n",
|
||||
" return False\n",
|
||||
"\n",
|
||||
" try:\n",
|
||||
" with self.connection.cursor() as cursor:\n",
|
||||
" # Use UPSERT (INSERT ... ON CONFLICT)\n",
|
||||
" upsert_sql = \"\"\"\n",
|
||||
" INSERT INTO metering_points (\n",
|
||||
" metering_point_id, street_code, street_name, building_number,\n",
|
||||
" floor_id, room_id, city_subdivision_name, municipality_code,\n",
|
||||
" location_description, settlement_method, meter_reading_occurrence,\n",
|
||||
" first_consumer_party_name, second_consumer_party_name,\n",
|
||||
" meter_number, consumer_start_date, type_of_mp,\n",
|
||||
" balance_supplier_name, postcode\n",
|
||||
" ) VALUES (\n",
|
||||
" %(meteringPointId)s, %(streetCode)s, %(streetName)s, %(buildingNumber)s,\n",
|
||||
" %(floorId)s, %(roomId)s, %(citySubDivisionName)s, %(municipalityCode)s,\n",
|
||||
" %(locationDescription)s, %(settlementMethod)s, %(meterReadingOccurrence)s,\n",
|
||||
" %(firstConsumerPartyName)s, %(secondConsumerPartyName)s,\n",
|
||||
" %(meterNumber)s, %(consumerStartDate)s, %(typeOfMP)s,\n",
|
||||
" %(balanceSupplierName)s, %(postcode)s\n",
|
||||
" )\n",
|
||||
" ON CONFLICT (metering_point_id) \n",
|
||||
" DO UPDATE SET\n",
|
||||
" street_code = EXCLUDED.street_code,\n",
|
||||
" street_name = EXCLUDED.street_name,\n",
|
||||
" building_number = EXCLUDED.building_number,\n",
|
||||
" updated_at = CURRENT_TIMESTAMP;\n",
|
||||
" \"\"\"\n",
|
||||
"\n",
|
||||
" cursor.execute(upsert_sql, metering_point_data)\n",
|
||||
" self.connection.commit()\n",
|
||||
" print(\n",
|
||||
" f\"Metering point {metering_point_data['meteringPointId']} stored successfully\"\n",
|
||||
" )\n",
|
||||
" return True\n",
|
||||
"\n",
|
||||
" except Exception as e:\n",
|
||||
" print(f\"Error storing metering point: {e}\")\n",
|
||||
" self.connection.rollback()\n",
|
||||
" return False\n",
|
||||
"\n",
|
||||
" def store_consumption_readings(self, metering_point_id, consumption_data):\n",
|
||||
" \"\"\"Store consumption readings with conflict handling\"\"\"\n",
|
||||
" if not self.connection:\n",
|
||||
" print(\"No database connection\")\n",
|
||||
" return False\n",
|
||||
"\n",
|
||||
" try:\n",
|
||||
" with self.connection.cursor() as cursor:\n",
|
||||
" # Prepare batch insert with conflict handling\n",
|
||||
" insert_sql = \"\"\"\n",
|
||||
" INSERT INTO consumption_readings (\n",
|
||||
" metering_point_id, timestamp, consumption_kwh, quality, period_resolution\n",
|
||||
" ) VALUES (\n",
|
||||
" %s, %s, %s, %s, %s\n",
|
||||
" )\n",
|
||||
" ON CONFLICT (metering_point_id, timestamp) \n",
|
||||
" DO UPDATE SET\n",
|
||||
" consumption_kwh = EXCLUDED.consumption_kwh,\n",
|
||||
" quality = EXCLUDED.quality;\n",
|
||||
" \"\"\"\n",
|
||||
"\n",
|
||||
" readings_data = []\n",
|
||||
" for reading in consumption_data:\n",
|
||||
" readings_data.append(\n",
|
||||
" (\n",
|
||||
" metering_point_id,\n",
|
||||
" reading[\"timestamp\"],\n",
|
||||
" reading[\"consumption_kwh\"],\n",
|
||||
" reading[\"quality\"],\n",
|
||||
" reading.get(\"period_resolution\", \"PT1H\"),\n",
|
||||
" )\n",
|
||||
" )\n",
|
||||
"\n",
|
||||
" cursor.executemany(insert_sql, readings_data)\n",
|
||||
" self.connection.commit()\n",
|
||||
" print(f\"Stored {len(readings_data)} consumption readings\")\n",
|
||||
" return True\n",
|
||||
"\n",
|
||||
" except Exception as e:\n",
|
||||
" print(f\"Error storing consumption readings: {e}\")\n",
|
||||
" self.connection.rollback()\n",
|
||||
" return False\n",
|
||||
"\n",
|
||||
" def close(self):\n",
|
||||
" \"\"\"Close database connection\"\"\"\n",
|
||||
" if self.connection:\n",
|
||||
" self.connection.close()\n",
|
||||
" print(\"Database connection closed\")\n",
|
||||
"\n",
|
||||
"\n",
|
||||
"# Example usage:\n",
|
||||
"print(\"EnergyDataStorage class defined. Use it like this:\")\n",
|
||||
"print(\"\"\"\n",
|
||||
"# Initialize storage\n",
|
||||
"storage = EnergyDataStorage(DB_CONFIG)\n",
|
||||
"storage.connect()\n",
|
||||
"storage.create_tables()\n",
|
||||
"\n",
|
||||
"# Store metering point data\n",
|
||||
"storage.store_metering_point(metering_point_data)\n",
|
||||
"\n",
|
||||
"# Store consumption readings\n",
|
||||
"storage.store_consumption_readings(metering_point_id, readings_list)\n",
|
||||
"\n",
|
||||
"storage.close()\n",
|
||||
"\"\"\")"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "d00bdc8f",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"# Function to transform API data for database storage\n",
|
||||
"\n",
|
||||
"\n",
|
||||
"def transform_consumption_data_for_db(consumption_json):\n",
|
||||
" \"\"\"\n",
|
||||
" Transform the Eloverblik API response into database-ready format\n",
|
||||
" \"\"\"\n",
|
||||
" if not consumption_json or \"result\" not in consumption_json:\n",
|
||||
" return None, []\n",
|
||||
"\n",
|
||||
" try:\n",
|
||||
" # Extract metering point data\n",
|
||||
" market_document = consumption_json[\"result\"][0][\"MyEnergyData_MarketDocument\"]\n",
|
||||
" time_series = market_document[\"TimeSeries\"][0]\n",
|
||||
" metering_point_id = time_series[\"mRID\"]\n",
|
||||
"\n",
|
||||
" # For demonstration, we'll create a basic metering point record\n",
|
||||
" # In practice, you'd get this from the metering points API call\n",
|
||||
" metering_point_data = {\n",
|
||||
" \"meteringPointId\": metering_point_id,\n",
|
||||
" \"streetCode\": None,\n",
|
||||
" \"streetName\": None,\n",
|
||||
" \"buildingNumber\": None,\n",
|
||||
" \"floorId\": None,\n",
|
||||
" \"roomId\": None,\n",
|
||||
" \"citySubDivisionName\": None,\n",
|
||||
" \"municipalityCode\": None,\n",
|
||||
" \"locationDescription\": None,\n",
|
||||
" \"settlementMethod\": None,\n",
|
||||
" \"meterReadingOccurrence\": None,\n",
|
||||
" \"firstConsumerPartyName\": None,\n",
|
||||
" \"secondConsumerPartyName\": None,\n",
|
||||
" \"meterNumber\": None,\n",
|
||||
" \"consumerStartDate\": None,\n",
|
||||
" \"typeOfMP\": None,\n",
|
||||
" \"balanceSupplierName\": None,\n",
|
||||
" \"postcode\": None,\n",
|
||||
" }\n",
|
||||
"\n",
|
||||
" # Extract consumption readings\n",
|
||||
" consumption_readings = []\n",
|
||||
" periods = time_series[\"Period\"]\n",
|
||||
"\n",
|
||||
" for period in periods:\n",
|
||||
" period_start = datetime.fromisoformat(\n",
|
||||
" period[\"timeInterval\"][\"start\"].replace(\"Z\", \"+00:00\")\n",
|
||||
" )\n",
|
||||
" resolution = period[\"resolution\"]\n",
|
||||
"\n",
|
||||
" if \"Point\" in period:\n",
|
||||
" for point in period[\"Point\"]:\n",
|
||||
" # Calculate the actual timestamp for this point\n",
|
||||
" position = int(point[\"position\"])\n",
|
||||
" # Position is 1-based, so subtract 1 to get hours offset\n",
|
||||
" hours_offset = position - 1\n",
|
||||
"\n",
|
||||
" point_timestamp = period_start + timedelta(hours=hours_offset)\n",
|
||||
"\n",
|
||||
" reading = {\n",
|
||||
" \"timestamp\": point_timestamp,\n",
|
||||
" \"consumption_kwh\": float(point[\"out_Quantity.quantity\"]),\n",
|
||||
" \"quality\": point[\"out_Quantity.quality\"],\n",
|
||||
" \"period_resolution\": resolution,\n",
|
||||
" }\n",
|
||||
" consumption_readings.append(reading)\n",
|
||||
"\n",
|
||||
" return metering_point_data, consumption_readings\n",
|
||||
"\n",
|
||||
" except Exception as e:\n",
|
||||
" print(f\"Error transforming data: {e}\")\n",
|
||||
" return None, []\n",
|
||||
"\n",
|
||||
"\n",
|
||||
"# Test the transformation with our existing data\n",
|
||||
"if \"consumption_json\" in locals() and consumption_json:\n",
|
||||
" metering_point, readings = transform_consumption_data_for_db(consumption_json)\n",
|
||||
"\n",
|
||||
" if metering_point and readings:\n",
|
||||
" print(\n",
|
||||
" f\"Transformed data for metering point: {metering_point['meteringPointId']}\"\n",
|
||||
" )\n",
|
||||
" print(f\"Number of readings: {len(readings)}\")\n",
|
||||
" print(\"\\nSample readings:\")\n",
|
||||
" for i, reading in enumerate(readings[:3]):\n",
|
||||
" print(\n",
|
||||
" f\" {i + 1}. {reading['timestamp']}: {reading['consumption_kwh']} kWh (Quality: {reading['quality']})\"\n",
|
||||
" )\n",
|
||||
" print(\" ...\")\n",
|
||||
" else:\n",
|
||||
" print(\"Failed to transform data\")\n",
|
||||
"else:\n",
|
||||
" print(\"No consumption data available to transform\")"
|
||||
]
|
||||
}
|
||||
],
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"display_name": "energy-consumption-ingester",
|
||||
"language": "python",
|
||||
"name": "python3"
|
||||
},
|
||||
"language_info": {
|
||||
"codemirror_mode": {
|
||||
"name": "ipython",
|
||||
"version": 3
|
||||
},
|
||||
"file_extension": ".py",
|
||||
"mimetype": "text/x-python",
|
||||
"name": "python",
|
||||
"nbconvert_exporter": "python",
|
||||
"pygments_lexer": "ipython3",
|
||||
"version": "3.12.1"
|
||||
}
|
||||
},
|
||||
"nbformat": 4,
|
||||
"nbformat_minor": 5
|
||||
}
|
||||
@@ -1,3 +1,380 @@
|
||||
# energy-consumption-ingester
|
||||
# Energy Consumption Ingester
|
||||
|
||||
Scheduled pipeline that ingests energy consumption data.
|
||||
A Python package for ingesting energy consumption data from the Eloverblik API and storing it in PostgreSQL.
|
||||
|
||||
## Project Structure
|
||||
|
||||
```
|
||||
energy-consumption-ingester/
|
||||
├── src/
|
||||
│ └── energy_consumption_ingester/
|
||||
│ ├── __init__.py # Package initialization
|
||||
│ ├── eloverblik.py # Eloverblik API client
|
||||
│ ├── database.py # PostgreSQL storage
|
||||
│ ├── utils.py # Data transformation utilities
|
||||
│ ├── cli.py # Command line interface
|
||||
│ └── config.py # Configuration and examples
|
||||
├── scripts/
|
||||
│ └── create_db_user.sh # Database setup script
|
||||
├── example.py # Usage example
|
||||
├── main.py # Original script
|
||||
├── POC.ipynb # Jupyter notebook for development
|
||||
├── pyproject.toml # Package configuration
|
||||
├── README.md # This file
|
||||
└── .env # Environment variables (create this)
|
||||
```
|
||||
|
||||
## Installation
|
||||
|
||||
### Prerequisites
|
||||
|
||||
1. **PostgreSQL** - Make sure you have PostgreSQL installed and running
|
||||
2. **Python 3.12+** - Required Python version
|
||||
3. **uv** - Package manager
|
||||
|
||||
### Setup
|
||||
|
||||
1. **Clone and enter the repository:**
|
||||
|
||||
```bash
|
||||
cd energy-consumption-ingester
|
||||
```
|
||||
|
||||
2. **Install dependencies using uv:**
|
||||
|
||||
```bash
|
||||
uv sync
|
||||
```
|
||||
|
||||
3. **Create environment file:**
|
||||
|
||||
```bash
|
||||
cp .env.example .env
|
||||
# Edit .env with your credentials
|
||||
```
|
||||
|
||||
4. **Set up PostgreSQL database:**
|
||||
|
||||
```bash
|
||||
# Create database
|
||||
createdb energy_consumption
|
||||
|
||||
# Create user (optional - run the script)
|
||||
chmod +x scripts/create_db_user.sh
|
||||
./scripts/create_db_user.sh
|
||||
```
|
||||
|
||||
## Configuration
|
||||
|
||||
Create a `.env` file in the project root:
|
||||
|
||||
```env
|
||||
# Eloverblik API Configuration
|
||||
ELOVERBLIK_API_TOKEN=your_refresh_token_here
|
||||
|
||||
# Database Configuration
|
||||
DB_HOST=localhost
|
||||
DB_NAME=energy_consumption
|
||||
DB_USER=energy_user
|
||||
DB_PASSWORD=your_password_here
|
||||
DB_PORT=5432
|
||||
```
|
||||
|
||||
## Usage
|
||||
|
||||
### Command Line Interface
|
||||
|
||||
The package provides a CLI for common operations:
|
||||
|
||||
```bash
|
||||
# Incremental sync (recommended for CI/CD hourly jobs)
|
||||
uv run energy-ingester sync-incremental
|
||||
|
||||
# Full sync for initial setup or backfill
|
||||
uv run energy-ingester sync --days 30
|
||||
|
||||
# Health check
|
||||
uv run energy-ingester health-check
|
||||
|
||||
# Database setup (initial setup)
|
||||
uv run energy-ingester setup-database
|
||||
|
||||
# Database reset (DESTRUCTIVE - deletes all data)
|
||||
uv run energy-ingester setup-database --reset
|
||||
|
||||
# Manual sync with custom parameters
|
||||
uv run energy-ingester sync-incremental --hours-back 48 --verbose
|
||||
```
|
||||
|
||||
### CI/CD Integration
|
||||
|
||||
For **Gitea CI** (recommended for hourly sync):
|
||||
|
||||
```bash
|
||||
# This is the command your CI should run every hour
|
||||
uv run energy-ingester sync-incremental --verbose
|
||||
```
|
||||
|
||||
The incremental sync is optimized for frequent runs:
|
||||
|
||||
- Only syncs recent data (last 25 hours by default)
|
||||
- Detects and fills gaps automatically
|
||||
- Handles overlapping data gracefully
|
||||
- Fast execution suitable for hourly scheduling
|
||||
|
||||
### Python API
|
||||
|
||||
```python
|
||||
from energy_consumption_ingester import (
|
||||
EloverblikClient,
|
||||
DatabaseStorage,
|
||||
transform_metering_point_data,
|
||||
transform_consumption_data
|
||||
)
|
||||
|
||||
# Initialize clients from environment variables
|
||||
client = EloverblikClient.from_env()
|
||||
db = DatabaseStorage.from_env()
|
||||
|
||||
# Connect and setup database
|
||||
db.connect()
|
||||
db.create_tables()
|
||||
|
||||
# Get metering points
|
||||
metering_points = client.get_metering_points()
|
||||
|
||||
# Get consumption data
|
||||
consumption_data = client.get_consumption_data_last_days(
|
||||
metering_points[0]['meteringPointId'],
|
||||
days=7
|
||||
)
|
||||
|
||||
# Transform and store data
|
||||
mp_data = transform_metering_point_data(metering_points[0])
|
||||
db.store_metering_point(mp_data)
|
||||
|
||||
metering_point_id, readings = transform_consumption_data(consumption_data)
|
||||
db.store_consumption_readings(metering_point_id, readings)
|
||||
|
||||
db.close()
|
||||
```
|
||||
|
||||
### Example Script
|
||||
|
||||
Run the included example:
|
||||
|
||||
```bash
|
||||
uv run python example.py
|
||||
```
|
||||
|
||||
## Development
|
||||
|
||||
### Install development dependencies:
|
||||
|
||||
```bash
|
||||
uv sync --group dev
|
||||
```
|
||||
|
||||
### Code formatting and linting:
|
||||
|
||||
```bash
|
||||
# Format code
|
||||
uv run black src/ example.py
|
||||
|
||||
# Lint code
|
||||
uv run flake8 src/ example.py
|
||||
|
||||
# Type checking
|
||||
uv run mypy src/
|
||||
```
|
||||
|
||||
### Testing:
|
||||
|
||||
```bash
|
||||
uv run pytest
|
||||
```
|
||||
|
||||
## CI/CD Setup for Gitea
|
||||
|
||||
### Automated Workflows
|
||||
|
||||
The project includes pre-configured Gitea workflows in `.gitea/workflows/`:
|
||||
|
||||
1. **`hourly-sync.yml`** - Runs every hour to sync energy data
|
||||
2. **`daily-health-check.yml`** - Daily system health monitoring
|
||||
3. **`manual-full-sync.yml`** - Manual trigger for full data backfill
|
||||
4. **`database-setup.yml`** - Manual database setup and management
|
||||
|
||||
### Required Secrets
|
||||
|
||||
Configure these secrets in your Gitea repository settings:
|
||||
|
||||
```
|
||||
ELOVERBLIK_API_TOKEN # Your Eloverblik refresh token
|
||||
DB_HOST # Database host (e.g., your-db-server.com)
|
||||
DB_NAME # Database name (e.g., energy_consumption)
|
||||
DB_USER # Database username
|
||||
DB_PASSWORD # Database password
|
||||
DB_PORT # Database port (usually 5432)
|
||||
```
|
||||
|
||||
### Workflow Schedule
|
||||
|
||||
- **Hourly Sync**: Runs at 5 minutes past every hour (`5 * * * *`)
|
||||
- **Health Check**: Runs daily at 6:00 AM UTC (`0 6 * * *`)
|
||||
- **Manual Sync**: Can be triggered manually with custom day range
|
||||
|
||||
### Commands for CI
|
||||
|
||||
```bash
|
||||
# Recommended for hourly CI jobs (fast, incremental)
|
||||
uv run energy-ingester sync-incremental --verbose
|
||||
|
||||
# For initial setup or recovery (slower, comprehensive)
|
||||
uv run energy-ingester sync --days 30 --verbose
|
||||
|
||||
# For monitoring and alerting
|
||||
uv run energy-ingester health-check --verbose
|
||||
|
||||
# Database setup (run once initially or after schema changes)
|
||||
uv run energy-ingester setup-database --verbose
|
||||
```
|
||||
|
||||
### Database Setup Workflow
|
||||
|
||||
The `database-setup.yml` workflow provides safe database management:
|
||||
|
||||
- **Setup**: Create tables and schema (safe, idempotent)
|
||||
- **Migrate**: Re-run table creation (for schema updates)
|
||||
- **Reset**: Drop all tables and recreate (DESTRUCTIVE - requires confirmation)
|
||||
|
||||
To run the database setup in Gitea CI:
|
||||
|
||||
1. Go to Actions → Database Setup
|
||||
2. Choose action: `setup`, `migrate`, or `reset`
|
||||
3. For reset operations, type `yes` in the confirm field
|
||||
4. Run workflow
|
||||
|
||||
### Docker Deployment
|
||||
|
||||
For containerized deployment:
|
||||
|
||||
```bash
|
||||
# Build and run with Docker Compose
|
||||
docker-compose up -d
|
||||
|
||||
# Or build manually
|
||||
docker build -t energy-ingester .
|
||||
docker run -e ELOVERBLIK_API_TOKEN=your_token energy-ingester
|
||||
```
|
||||
|
||||
### Monitoring and Alerts
|
||||
|
||||
The workflows are designed to:
|
||||
|
||||
- Exit with error codes on failure (for CI/CD detection)
|
||||
- Log detailed information for debugging
|
||||
- Support notification integration (extend the workflows)
|
||||
|
||||
To add Slack/email notifications, modify the workflow files in `.gitea/workflows/`.
|
||||
|
||||
## Modules
|
||||
|
||||
### 1. `eloverblik.py` - API Client
|
||||
|
||||
**Purpose:** Handles all interactions with the Eloverblik API
|
||||
|
||||
**Key Features:**
|
||||
|
||||
- Automatic token refresh (refresh token → access token)
|
||||
- Metering points discovery
|
||||
- Consumption data retrieval with flexible date ranges
|
||||
- Error handling and retry logic
|
||||
- Environment variable configuration
|
||||
|
||||
**Main Methods:**
|
||||
|
||||
- `get_access_token()` - Exchange refresh token for access token
|
||||
- `get_metering_points()` - Retrieve all metering points
|
||||
- `get_consumption_data()` - Get consumption data for date range
|
||||
- `get_consumption_data_last_days()` - Convenience method for recent data
|
||||
|
||||
### 2. `database.py` - PostgreSQL Storage
|
||||
|
||||
**Purpose:** Handles all database operations for storing energy data
|
||||
|
||||
**Key Features:**
|
||||
|
||||
- Normalized database schema design
|
||||
- UPSERT operations (handles duplicates gracefully)
|
||||
- Proper indexing for performance
|
||||
- Transaction management
|
||||
- Connection pooling ready
|
||||
|
||||
**Main Methods:**
|
||||
|
||||
- `create_tables()` - Set up database schema
|
||||
- `store_metering_point()` - Store/update metering point info
|
||||
- `store_consumption_readings()` - Batch insert consumption data
|
||||
- `get_latest_reading_timestamp()` - Find last stored reading
|
||||
|
||||
**Database Schema:**
|
||||
|
||||
- `metering_points` - Meter metadata (location, consumer info, etc.)
|
||||
- `consumption_readings` - Time series consumption data
|
||||
|
||||
### 3. `utils.py` - Data Transformation
|
||||
|
||||
**Purpose:** Transform API responses to database-ready format
|
||||
|
||||
**Key Features:**
|
||||
|
||||
- API data normalization
|
||||
- Timestamp parsing and timezone handling
|
||||
- Data validation
|
||||
- Statistics calculation
|
||||
|
||||
## Architecture Benefits
|
||||
|
||||
### 1. **Separation of Concerns**
|
||||
|
||||
- **API layer** (`eloverblik.py`) - Pure API interactions
|
||||
- **Storage layer** (`database.py`) - Pure database operations
|
||||
- **Transform layer** (`utils.py`) - Data processing logic
|
||||
|
||||
### 2. **Modular Design**
|
||||
|
||||
- Each module can be used independently
|
||||
- Easy to test individual components
|
||||
- Simple to extend or replace components
|
||||
|
||||
### 3. **Environment-Based Configuration**
|
||||
|
||||
- No hardcoded credentials
|
||||
- Easy deployment across environments
|
||||
- Secure credential management
|
||||
|
||||
### 4. **uv Package Management**
|
||||
|
||||
- Fast dependency resolution
|
||||
- Proper Python packaging
|
||||
- Development vs production dependencies
|
||||
- Script entry points
|
||||
|
||||
### 5. **Database Design**
|
||||
|
||||
- Normalized schema prevents data duplication
|
||||
- Proper indexing for time-series queries
|
||||
- UPSERT operations handle data refresh scenarios
|
||||
- Extensible for multiple meters/customers
|
||||
|
||||
## Best Practices Implemented
|
||||
|
||||
1. **Type Hints** - Full type annotation for better IDE support and validation
|
||||
2. **Error Handling** - Comprehensive exception handling with logging
|
||||
3. **Configuration Management** - Environment variables with validation
|
||||
4. **Documentation** - Comprehensive docstrings and README
|
||||
5. **CLI Interface** - User-friendly command line tool
|
||||
6. **Packaging** - Proper Python package structure with `pyproject.toml`
|
||||
|
||||
This structure gives you a professional, maintainable, and scalable foundation for your energy consumption ingestion project!
|
||||
|
||||
+112
@@ -0,0 +1,112 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Example script demonstrating the energy consumption ingester."""
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
# Optional import for development
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
|
||||
_load_dotenv: Any | None = load_dotenv
|
||||
except ImportError:
|
||||
_load_dotenv = None
|
||||
|
||||
from energy_consumption_ingester import (
|
||||
EloverblikClient,
|
||||
DatabaseStorage,
|
||||
transform_metering_point_data,
|
||||
transform_consumption_data,
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Example usage of the energy consumption ingester."""
|
||||
# Load environment variables from .env file (if available)
|
||||
if _load_dotenv is not None:
|
||||
_load_dotenv()
|
||||
|
||||
# Set up logging
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
# Initialize the Eloverblik client
|
||||
logger.info("Initializing Eloverblik client...")
|
||||
client = EloverblikClient.from_env()
|
||||
|
||||
# Initialize database storage
|
||||
logger.info("Initializing database connection...")
|
||||
db = DatabaseStorage.from_env()
|
||||
|
||||
# Connect to database and create tables
|
||||
if not db.connect():
|
||||
logger.error("Failed to connect to database")
|
||||
return
|
||||
|
||||
if not db.create_tables():
|
||||
logger.error("Failed to create database tables")
|
||||
return
|
||||
|
||||
# Get metering points
|
||||
logger.info("Fetching metering points...")
|
||||
metering_points = client.get_metering_points()
|
||||
|
||||
if not metering_points:
|
||||
logger.error("No metering points found")
|
||||
return
|
||||
|
||||
logger.info(f"Found {len(metering_points)} metering points")
|
||||
|
||||
# Process the first metering point
|
||||
mp_data = metering_points[0]
|
||||
metering_point_id = mp_data["meteringPointId"]
|
||||
|
||||
logger.info(f"Processing metering point: {metering_point_id}")
|
||||
|
||||
# Transform and store metering point data
|
||||
transformed_mp = transform_metering_point_data(mp_data)
|
||||
if db.store_metering_point(transformed_mp):
|
||||
logger.info("Metering point stored successfully")
|
||||
|
||||
# Get consumption data for the last 7 days
|
||||
logger.info("Fetching consumption data...")
|
||||
consumption_data = client.get_consumption_data_last_days(
|
||||
metering_point_id, days=7
|
||||
)
|
||||
|
||||
if consumption_data:
|
||||
# Transform consumption data
|
||||
_, readings = transform_consumption_data(consumption_data)
|
||||
|
||||
if readings:
|
||||
# Store consumption readings
|
||||
if db.store_consumption_readings(metering_point_id, readings):
|
||||
logger.info(f"Stored {len(readings)} consumption readings")
|
||||
|
||||
# Show some statistics
|
||||
total_consumption = sum(r["consumption_kwh"] for r in readings)
|
||||
avg_consumption = total_consumption / len(readings)
|
||||
|
||||
logger.info(f"Total consumption: {total_consumption:.2f} kWh")
|
||||
logger.info(
|
||||
f"Average hourly consumption: {avg_consumption:.3f} kWh"
|
||||
)
|
||||
else:
|
||||
logger.error("Failed to store consumption readings")
|
||||
else:
|
||||
logger.warning("No consumption readings to store")
|
||||
else:
|
||||
logger.warning("No consumption data retrieved")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error: {e}")
|
||||
finally:
|
||||
# Close database connection
|
||||
if "db" in locals():
|
||||
db.close()
|
||||
logger.info("Database connection closed")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,55 @@
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "energy-consumption-ingester"
|
||||
version = "0.1.0"
|
||||
description = "Energy consumption data ingester for Eloverblik API with PostgreSQL storage"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
authors = [
|
||||
{name = "Brian Bjarke Jensen", email = "your-email@example.com"},
|
||||
]
|
||||
keywords = ["energy", "consumption", "eloverblik", "postgresql", "data"]
|
||||
classifiers = [
|
||||
"Development Status :: 3 - Alpha",
|
||||
"Intended Audience :: Developers",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
]
|
||||
|
||||
dependencies = [
|
||||
"pydantic>=2.12.3",
|
||||
"psycopg2-binary>=2.9.0",
|
||||
]
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"ipykernel>=7.0.1",
|
||||
"python-dotenv>=1.0.0",
|
||||
"pytest>=7.0.0",
|
||||
"black>=23.0.0",
|
||||
"flake8>=6.0.0",
|
||||
"mypy>=1.0.0",
|
||||
"ruff>=0.14.2",
|
||||
"pyupgrade>=3.21.0",
|
||||
"types-psycopg2>=2.9.21.20251012",
|
||||
"pre-commit>=4.3.0",
|
||||
]
|
||||
|
||||
[project.scripts]
|
||||
energy-ingester = "energy_consumption_ingester.cli:main"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["src/energy_consumption_ingester"]
|
||||
|
||||
[tool.black]
|
||||
line-length = 88
|
||||
target-version = ['py312']
|
||||
|
||||
[tool.mypy]
|
||||
python_version = "3.12"
|
||||
warn_return_any = true
|
||||
warn_unused_configs = true
|
||||
disallow_untyped_defs = true
|
||||
@@ -0,0 +1,21 @@
|
||||
"""Energy consumption data ingester for Eloverblik API."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
|
||||
from .eloverblik import EloverblikClient
|
||||
from .database import DatabaseStorage
|
||||
from .scheduler import ScheduledSync
|
||||
from .utils import (
|
||||
transform_metering_point_data,
|
||||
transform_consumption_data,
|
||||
calculate_consumption_stats,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"EloverblikClient",
|
||||
"DatabaseStorage",
|
||||
"ScheduledSync",
|
||||
"transform_metering_point_data",
|
||||
"transform_consumption_data",
|
||||
"calculate_consumption_stats",
|
||||
]
|
||||
@@ -0,0 +1,6 @@
|
||||
"""CLI module entry point for python -m execution."""
|
||||
|
||||
from .cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,374 @@
|
||||
"""Command line interface for energy consumption ingester."""
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
# Optional import for development
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
|
||||
_load_dotenv: Any | None = load_dotenv
|
||||
except ImportError:
|
||||
_load_dotenv = None
|
||||
|
||||
from .eloverblik import EloverblikClient
|
||||
from .database import DatabaseStorage
|
||||
from .utils import transform_metering_point_data, transform_consumption_data
|
||||
from .scheduler import ScheduledSync
|
||||
|
||||
|
||||
def load_environment() -> None:
|
||||
"""Load environment variables from .env file if available."""
|
||||
if _load_dotenv:
|
||||
_load_dotenv()
|
||||
|
||||
|
||||
def setup_logging(verbose: bool = False) -> None:
|
||||
"""Set up logging configuration."""
|
||||
level = logging.DEBUG if verbose else logging.INFO
|
||||
logging.basicConfig(
|
||||
level=level, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
|
||||
|
||||
def sync_incremental(hours_back: int = 25, verbose: bool = False) -> int:
|
||||
"""Sync consumption data incrementally for the last specified hours."""
|
||||
setup_logging(verbose)
|
||||
logger = logging.getLogger(__name__)
|
||||
load_environment()
|
||||
|
||||
# Get required environment variables
|
||||
api_token = os.getenv("ELOVERBLIK_API_TOKEN")
|
||||
if not api_token:
|
||||
logger.error("ELOVERBLIK_API_TOKEN environment variable is required")
|
||||
return 1
|
||||
|
||||
try:
|
||||
# Initialize scheduled sync
|
||||
sync = ScheduledSync.from_env()
|
||||
|
||||
# Connect to database
|
||||
if not sync.db.connect():
|
||||
logger.error("Failed to connect to database")
|
||||
return 1
|
||||
|
||||
if not sync.db.create_tables():
|
||||
logger.error("Failed to create database tables")
|
||||
return 1
|
||||
|
||||
# Perform incremental sync
|
||||
logger.info("Starting incremental synchronization")
|
||||
if sync.sync_incremental(hours_back=hours_back):
|
||||
logger.info("Incremental sync completed successfully")
|
||||
return 0
|
||||
else:
|
||||
logger.error("Incremental sync failed")
|
||||
return 1
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error during incremental sync: {e}")
|
||||
return 1
|
||||
finally:
|
||||
if "sync" in locals():
|
||||
sync.db.close()
|
||||
|
||||
|
||||
def setup_database(reset: bool = False, verbose: bool = False) -> int:
|
||||
"""Setup the database schema."""
|
||||
setup_logging(verbose)
|
||||
logger = logging.getLogger(__name__)
|
||||
load_environment()
|
||||
|
||||
db_params: dict[str, Any] = {
|
||||
"host": os.getenv("DB_HOST", "localhost"),
|
||||
"port": int(os.getenv("DB_PORT", "5432")),
|
||||
"database": os.getenv("DB_NAME", "energy_consumption"),
|
||||
"user": os.getenv("DB_USER"),
|
||||
"password": os.getenv("DB_PASSWORD"),
|
||||
}
|
||||
|
||||
if not db_params["user"] or not db_params["password"]:
|
||||
logger.error("DB_USER and DB_PASSWORD environment variables are required")
|
||||
return 1
|
||||
|
||||
try:
|
||||
db = DatabaseStorage(db_params)
|
||||
|
||||
# Connect to database
|
||||
if not db.connect():
|
||||
logger.error("Failed to connect to database")
|
||||
return 1
|
||||
|
||||
# Drop tables if reset is requested
|
||||
if reset or os.getenv("RESET_DB", "").lower() in ("true", "1", "yes"):
|
||||
logger.info("Resetting database (dropping existing tables)")
|
||||
if db.connection is not None:
|
||||
cursor = db.connection.cursor()
|
||||
cursor.execute("DROP TABLE IF EXISTS consumption_readings CASCADE;")
|
||||
cursor.execute("DROP TABLE IF EXISTS metering_points CASCADE;")
|
||||
cursor.execute(
|
||||
"DROP FUNCTION IF EXISTS update_updated_at_column() CASCADE;"
|
||||
)
|
||||
db.connection.commit()
|
||||
logger.info("Existing tables dropped")
|
||||
|
||||
# Create database schema
|
||||
if db.create_tables():
|
||||
logger.info("Database schema created successfully")
|
||||
return 0
|
||||
else:
|
||||
logger.error("Failed to create database schema")
|
||||
return 1
|
||||
except Exception as e:
|
||||
logger.error(f"Error setting up database: {e}")
|
||||
return 1
|
||||
finally:
|
||||
if "db" in locals():
|
||||
db.close()
|
||||
|
||||
|
||||
def health_check(verbose: bool = False) -> int:
|
||||
"""Perform a health check of the system."""
|
||||
setup_logging(verbose)
|
||||
logger = logging.getLogger(__name__)
|
||||
load_environment()
|
||||
|
||||
# Get required environment variables
|
||||
api_token = os.getenv("ELOVERBLIK_API_TOKEN")
|
||||
if not api_token:
|
||||
logger.error("ELOVERBLIK_API_TOKEN environment variable is required")
|
||||
return 1
|
||||
|
||||
db_params: dict[str, Any] = {
|
||||
"host": os.getenv("DB_HOST", "localhost"),
|
||||
"port": int(os.getenv("DB_PORT", "5432")),
|
||||
"database": os.getenv("DB_NAME", "energy_consumption"),
|
||||
"user": os.getenv("DB_USER"),
|
||||
"password": os.getenv("DB_PASSWORD"),
|
||||
}
|
||||
|
||||
if not db_params["user"] or not db_params["password"]:
|
||||
logger.error("DB_USER and DB_PASSWORD environment variables are required")
|
||||
return 1
|
||||
|
||||
try:
|
||||
from .scheduler import ScheduledSync
|
||||
|
||||
logger.info("Starting health check")
|
||||
|
||||
# Initialize clients
|
||||
eloverblik = EloverblikClient(api_token)
|
||||
db = DatabaseStorage(db_params)
|
||||
|
||||
# Create ScheduledSync with proper instances
|
||||
sync = ScheduledSync(eloverblik, db)
|
||||
|
||||
if sync.health_check():
|
||||
logger.info("Health check passed")
|
||||
return 0
|
||||
else:
|
||||
logger.error("Health check failed")
|
||||
return 1
|
||||
except Exception as e:
|
||||
logger.error(f"Error during health check: {e}")
|
||||
return 1
|
||||
|
||||
|
||||
def sync_full_backfill(verbose: bool = False) -> int:
|
||||
"""Perform a full backfill synchronization."""
|
||||
setup_logging(verbose)
|
||||
logger = logging.getLogger(__name__)
|
||||
load_environment()
|
||||
|
||||
# Get required environment variables
|
||||
api_token = os.getenv("ELOVERBLIK_API_TOKEN")
|
||||
if not api_token:
|
||||
logger.error("ELOVERBLIK_API_TOKEN environment variable is required")
|
||||
return 1
|
||||
|
||||
try:
|
||||
# Initialize scheduled sync
|
||||
sync = ScheduledSync.from_env()
|
||||
|
||||
# Connect to database
|
||||
if not sync.db.connect():
|
||||
logger.error("Failed to connect to database")
|
||||
return 1
|
||||
|
||||
if not sync.db.create_tables():
|
||||
logger.error("Failed to create database tables")
|
||||
return 1
|
||||
|
||||
# Perform full backfill sync
|
||||
logger.info("Starting full backfill synchronization")
|
||||
if sync.sync_full_backfill():
|
||||
logger.info("Full backfill sync completed successfully")
|
||||
return 0
|
||||
else:
|
||||
logger.error("Full backfill sync failed")
|
||||
return 1
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error during full backfill sync: {e}")
|
||||
return 1
|
||||
finally:
|
||||
if "sync" in locals():
|
||||
sync.db.close()
|
||||
|
||||
|
||||
def sync_data(days: int = 7, verbose: bool = False) -> None:
|
||||
"""
|
||||
Sync energy consumption data from Eloverblik to PostgreSQL.
|
||||
|
||||
Args:
|
||||
days: Number of days to sync (from today backwards)
|
||||
verbose: Enable verbose logging
|
||||
"""
|
||||
setup_logging(verbose)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Load environment variables
|
||||
load_environment()
|
||||
|
||||
try:
|
||||
# Initialize clients
|
||||
eloverblik = EloverblikClient.from_env()
|
||||
db = DatabaseStorage.from_env()
|
||||
|
||||
# Connect to database and create tables
|
||||
if not db.connect():
|
||||
logger.error("Failed to connect to database")
|
||||
sys.exit(1)
|
||||
|
||||
if not db.create_tables():
|
||||
logger.error("Failed to create database tables")
|
||||
sys.exit(1)
|
||||
|
||||
# Get metering points
|
||||
logger.info("Fetching metering points...")
|
||||
metering_points = eloverblik.get_metering_points()
|
||||
|
||||
if not metering_points:
|
||||
logger.error("No metering points found")
|
||||
sys.exit(1)
|
||||
|
||||
logger.info(f"Found {len(metering_points)} metering points")
|
||||
|
||||
# Process each metering point
|
||||
for mp_data in metering_points:
|
||||
metering_point_id = mp_data["meteringPointId"]
|
||||
logger.info(f"Processing metering point: {metering_point_id}")
|
||||
|
||||
# Store metering point data
|
||||
transformed_mp = transform_metering_point_data(mp_data)
|
||||
if not db.store_metering_point(transformed_mp):
|
||||
logger.error(f"Failed to store metering point {metering_point_id}")
|
||||
continue
|
||||
|
||||
# Get consumption data
|
||||
logger.info(f"Fetching {days} days of consumption data...")
|
||||
consumption_data = eloverblik.get_consumption_data_last_days(
|
||||
metering_point_id, days
|
||||
)
|
||||
|
||||
if not consumption_data:
|
||||
logger.warning(f"No consumption data for {metering_point_id}")
|
||||
continue
|
||||
|
||||
# Transform and store consumption data
|
||||
_, readings = transform_consumption_data(consumption_data)
|
||||
|
||||
if readings:
|
||||
if db.store_consumption_readings(metering_point_id, readings):
|
||||
logger.info(
|
||||
f"Stored {len(readings)} readings for {metering_point_id}"
|
||||
)
|
||||
else:
|
||||
logger.error(f"Failed to store readings for {metering_point_id}")
|
||||
else:
|
||||
logger.warning(f"No readings to store for {metering_point_id}")
|
||||
|
||||
logger.info("Data synchronization completed successfully")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error during data sync: {e}")
|
||||
sys.exit(1)
|
||||
finally:
|
||||
if "db" in locals():
|
||||
db.close()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Main CLI entry point."""
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Energy consumption data ingester for Eloverblik API"
|
||||
)
|
||||
|
||||
subparsers = parser.add_subparsers(dest="command", help="Available commands")
|
||||
|
||||
# Sync command (legacy - full sync)
|
||||
sync_parser = subparsers.add_parser(
|
||||
"sync", help="Full sync from Eloverblik to PostgreSQL"
|
||||
)
|
||||
sync_parser.add_argument(
|
||||
"--days", type=int, default=7, help="Number of days to sync (default: 7)"
|
||||
)
|
||||
sync_parser.add_argument(
|
||||
"--verbose", action="store_true", help="Enable verbose logging"
|
||||
)
|
||||
|
||||
# Incremental sync command (optimized for CI)
|
||||
incr_parser = subparsers.add_parser(
|
||||
"sync-incremental",
|
||||
help="Incremental sync for CI/CD pipelines (recommended for hourly jobs)",
|
||||
)
|
||||
incr_parser.add_argument(
|
||||
"--hours-back",
|
||||
type=int,
|
||||
default=25,
|
||||
help="Hours to look back for gaps (default: 25)",
|
||||
)
|
||||
incr_parser.add_argument(
|
||||
"--verbose", action="store_true", help="Enable verbose logging"
|
||||
)
|
||||
|
||||
# Health check command
|
||||
health_parser = subparsers.add_parser(
|
||||
"health-check", help="Check system health (API connectivity, database, etc.)"
|
||||
)
|
||||
health_parser.add_argument(
|
||||
"--verbose", action="store_true", help="Enable verbose logging"
|
||||
)
|
||||
|
||||
# Database setup command
|
||||
db_parser = subparsers.add_parser(
|
||||
"setup-database", help="Set up database tables and schema"
|
||||
)
|
||||
db_parser.add_argument(
|
||||
"--reset",
|
||||
action="store_true",
|
||||
help="Reset database (DESTRUCTIVE - drops all tables)",
|
||||
)
|
||||
db_parser.add_argument(
|
||||
"--verbose", action="store_true", help="Enable verbose logging"
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.command == "sync":
|
||||
sync_data(days=args.days, verbose=args.verbose)
|
||||
elif args.command == "sync-incremental":
|
||||
sync_incremental(hours_back=args.hours_back, verbose=args.verbose)
|
||||
elif args.command == "health-check":
|
||||
health_check(verbose=args.verbose)
|
||||
elif args.command == "setup-database":
|
||||
setup_database(reset=args.reset, verbose=args.verbose)
|
||||
else:
|
||||
parser.print_help()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,48 @@
|
||||
"""Test configuration and example usage."""
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load environment variables
|
||||
load_dotenv()
|
||||
|
||||
# Example configuration
|
||||
EXAMPLE_ENV_FILE = """
|
||||
# Eloverblik API Configuration
|
||||
ELOVERBLIK_API_TOKEN=your_refresh_token_here
|
||||
|
||||
# Database Configuration
|
||||
DB_HOST=localhost
|
||||
DB_NAME=energy_consumption
|
||||
DB_USER=energy_user
|
||||
DB_PASSWORD=your_password_here
|
||||
DB_PORT=5432
|
||||
"""
|
||||
|
||||
# Example usage
|
||||
EXAMPLE_USAGE = """
|
||||
# Basic usage example
|
||||
from energy_consumption_ingester import EloverblikClient, DatabaseStorage
|
||||
|
||||
# Initialize clients
|
||||
client = EloverblikClient.from_env()
|
||||
db = DatabaseStorage.from_env()
|
||||
|
||||
# Connect and sync data
|
||||
db.connect()
|
||||
db.create_tables()
|
||||
|
||||
# Get metering points
|
||||
metering_points = client.get_metering_points()
|
||||
|
||||
# Get consumption data for last 7 days
|
||||
consumption_data = client.get_consumption_data_last_days(
|
||||
metering_points[0]['meteringPointId'],
|
||||
days=7
|
||||
)
|
||||
"""
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Example .env file content:")
|
||||
print(EXAMPLE_ENV_FILE)
|
||||
print("\nExample usage:")
|
||||
print(EXAMPLE_USAGE)
|
||||
@@ -0,0 +1,310 @@
|
||||
"""Database storage module for energy consumption data."""
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Any, TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import psycopg2
|
||||
from psycopg2.extras import RealDictCursor
|
||||
else:
|
||||
try:
|
||||
import psycopg2
|
||||
from psycopg2.extras import RealDictCursor
|
||||
except ImportError:
|
||||
psycopg2 = None # type: ignore
|
||||
RealDictCursor = None # type: ignore
|
||||
|
||||
|
||||
class DatabaseStorage:
|
||||
"""PostgreSQL database storage for energy consumption data."""
|
||||
|
||||
def __init__(self, db_config: dict[str, Any]):
|
||||
"""
|
||||
Initialize database storage.
|
||||
|
||||
Args:
|
||||
db_config: Database connection configuration dict
|
||||
"""
|
||||
if psycopg2 is None:
|
||||
raise ImportError(
|
||||
"psycopg2 not installed. Install with: uv add psycopg2-binary"
|
||||
)
|
||||
|
||||
self.db_config = db_config
|
||||
self.connection = None
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
def connect(self) -> bool:
|
||||
"""
|
||||
Establish database connection.
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
self.connection = psycopg2.connect(**self.db_config)
|
||||
self.logger.info("Database connection established")
|
||||
return True
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error connecting to database: {e}")
|
||||
return False
|
||||
|
||||
def create_tables(self) -> bool:
|
||||
"""
|
||||
Create database tables if they don't exist.
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
if not self.connection:
|
||||
self.logger.error("No database connection")
|
||||
return False
|
||||
|
||||
create_tables_sql = """
|
||||
-- Create metering_points table
|
||||
CREATE TABLE IF NOT EXISTS metering_points (
|
||||
id SERIAL PRIMARY KEY,
|
||||
metering_point_id VARCHAR(50) UNIQUE NOT NULL,
|
||||
street_code VARCHAR(10),
|
||||
street_name VARCHAR(255),
|
||||
building_number VARCHAR(20),
|
||||
floor_id VARCHAR(10),
|
||||
room_id VARCHAR(10),
|
||||
city_subdivision_name VARCHAR(255),
|
||||
municipality_code VARCHAR(10),
|
||||
location_description TEXT,
|
||||
settlement_method VARCHAR(10),
|
||||
meter_reading_occurrence VARCHAR(20),
|
||||
first_consumer_party_name VARCHAR(255),
|
||||
second_consumer_party_name VARCHAR(255),
|
||||
meter_number VARCHAR(50),
|
||||
consumer_start_date TIMESTAMP WITH TIME ZONE,
|
||||
type_of_mp VARCHAR(10),
|
||||
balance_supplier_name VARCHAR(255),
|
||||
postcode VARCHAR(10),
|
||||
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
-- Create consumption_readings table
|
||||
CREATE TABLE IF NOT EXISTS consumption_readings (
|
||||
id SERIAL PRIMARY KEY,
|
||||
metering_point_id VARCHAR(50) REFERENCES metering_points(metering_point_id),
|
||||
timestamp TIMESTAMP WITH TIME ZONE NOT NULL,
|
||||
consumption_kwh DECIMAL(10, 3) NOT NULL,
|
||||
quality VARCHAR(10),
|
||||
period_resolution VARCHAR(10) DEFAULT 'PT1H',
|
||||
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
|
||||
UNIQUE(metering_point_id, timestamp)
|
||||
);
|
||||
|
||||
-- Create indexes for performance
|
||||
CREATE INDEX IF NOT EXISTS idx_consumption_metering_point
|
||||
ON consumption_readings(metering_point_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_consumption_timestamp
|
||||
ON consumption_readings(timestamp);
|
||||
CREATE INDEX IF NOT EXISTS idx_consumption_metering_point_timestamp
|
||||
ON consumption_readings(metering_point_id, timestamp);
|
||||
|
||||
-- Create a function to update the updated_at timestamp
|
||||
CREATE OR REPLACE FUNCTION update_updated_at_column()
|
||||
RETURNS TRIGGER AS $$
|
||||
BEGIN
|
||||
NEW.updated_at = CURRENT_TIMESTAMP;
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$ language 'plpgsql';
|
||||
|
||||
-- Create trigger for metering_points
|
||||
CREATE TRIGGER update_metering_points_updated_at
|
||||
BEFORE UPDATE ON metering_points
|
||||
FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
|
||||
"""
|
||||
|
||||
try:
|
||||
with self.connection.cursor() as cursor:
|
||||
cursor.execute(create_tables_sql)
|
||||
self.connection.commit()
|
||||
self.logger.info("Tables created successfully")
|
||||
return True
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error creating tables: {e}")
|
||||
self.connection.rollback()
|
||||
return False
|
||||
|
||||
def store_metering_point(self, metering_point_data: dict[str, Any]) -> bool:
|
||||
"""
|
||||
Store or update metering point information.
|
||||
|
||||
Args:
|
||||
metering_point_data: Dictionary containing metering point data
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
if not self.connection:
|
||||
self.logger.error("No database connection")
|
||||
return False
|
||||
|
||||
try:
|
||||
with self.connection.cursor() as cursor:
|
||||
upsert_sql = """
|
||||
INSERT INTO metering_points (
|
||||
metering_point_id, street_code, street_name, building_number,
|
||||
floor_id, room_id, city_subdivision_name, municipality_code,
|
||||
location_description, settlement_method, meter_reading_occurrence,
|
||||
first_consumer_party_name, second_consumer_party_name,
|
||||
meter_number, consumer_start_date, type_of_mp,
|
||||
balance_supplier_name, postcode
|
||||
) VALUES (
|
||||
%(meteringPointId)s, %(streetCode)s, %(streetName)s, %(buildingNumber)s,
|
||||
%(floorId)s, %(roomId)s, %(citySubDivisionName)s, %(municipalityCode)s,
|
||||
%(locationDescription)s, %(settlementMethod)s, %(meterReadingOccurrence)s,
|
||||
%(firstConsumerPartyName)s, %(secondConsumerPartyName)s,
|
||||
%(meterNumber)s, %(consumerStartDate)s, %(typeOfMP)s,
|
||||
%(balanceSupplierName)s, %(postcode)s
|
||||
)
|
||||
ON CONFLICT (metering_point_id)
|
||||
DO UPDATE SET
|
||||
street_code = EXCLUDED.street_code,
|
||||
street_name = EXCLUDED.street_name,
|
||||
building_number = EXCLUDED.building_number,
|
||||
updated_at = CURRENT_TIMESTAMP;
|
||||
"""
|
||||
|
||||
cursor.execute(upsert_sql, metering_point_data)
|
||||
self.connection.commit()
|
||||
self.logger.info(
|
||||
f"Metering point {metering_point_data['meteringPointId']} stored"
|
||||
)
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error storing metering point: {e}")
|
||||
self.connection.rollback()
|
||||
return False
|
||||
|
||||
def store_consumption_readings(
|
||||
self, metering_point_id: str, consumption_data: list[dict[str, Any]]
|
||||
) -> bool:
|
||||
"""
|
||||
Store consumption readings with conflict handling.
|
||||
|
||||
Args:
|
||||
metering_point_id: The metering point identifier
|
||||
consumption_data: List of consumption reading dictionaries
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
if not self.connection:
|
||||
self.logger.error("No database connection")
|
||||
return False
|
||||
|
||||
try:
|
||||
with self.connection.cursor() as cursor:
|
||||
insert_sql = """
|
||||
INSERT INTO consumption_readings (
|
||||
metering_point_id, timestamp, consumption_kwh, quality, period_resolution
|
||||
) VALUES (
|
||||
%s, %s, %s, %s, %s
|
||||
)
|
||||
ON CONFLICT (metering_point_id, timestamp)
|
||||
DO UPDATE SET
|
||||
consumption_kwh = EXCLUDED.consumption_kwh,
|
||||
quality = EXCLUDED.quality;
|
||||
"""
|
||||
|
||||
readings_data = []
|
||||
for reading in consumption_data:
|
||||
readings_data.append(
|
||||
(
|
||||
metering_point_id,
|
||||
reading["timestamp"],
|
||||
reading["consumption_kwh"],
|
||||
reading["quality"],
|
||||
reading.get("period_resolution", "PT1H"),
|
||||
)
|
||||
)
|
||||
|
||||
cursor.executemany(insert_sql, readings_data)
|
||||
self.connection.commit()
|
||||
self.logger.info(f"Stored {len(readings_data)} consumption readings")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error storing consumption readings: {e}")
|
||||
self.connection.rollback()
|
||||
return False
|
||||
|
||||
def get_latest_reading_timestamp(self, metering_point_id: str) -> datetime | None:
|
||||
"""
|
||||
Get the timestamp of the latest reading for a metering point.
|
||||
|
||||
Args:
|
||||
metering_point_id: The metering point identifier
|
||||
|
||||
Returns:
|
||||
Latest timestamp or None if no readings exist
|
||||
"""
|
||||
if not self.connection:
|
||||
self.logger.error("No database connection")
|
||||
return None
|
||||
|
||||
try:
|
||||
with self.connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT MAX(timestamp)
|
||||
FROM consumption_readings
|
||||
WHERE metering_point_id = %s
|
||||
""",
|
||||
(metering_point_id,),
|
||||
)
|
||||
result = cursor.fetchone()
|
||||
return result[0] if result[0] else None
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error getting latest reading timestamp: {e}")
|
||||
return None
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close database connection."""
|
||||
if self.connection:
|
||||
self.connection.close()
|
||||
self.logger.info("Database connection closed")
|
||||
|
||||
@classmethod
|
||||
def from_env(cls) -> "DatabaseStorage":
|
||||
"""
|
||||
Create DatabaseStorage from environment variables.
|
||||
|
||||
Expected environment variables:
|
||||
- DB_HOST
|
||||
- DB_NAME
|
||||
- DB_USER
|
||||
- DB_PASSWORD
|
||||
- DB_PORT (optional, defaults to 5432)
|
||||
|
||||
Returns:
|
||||
DatabaseStorage instance
|
||||
"""
|
||||
import os
|
||||
|
||||
db_config = {
|
||||
"host": os.getenv("DB_HOST", "localhost"),
|
||||
"database": os.getenv("DB_NAME"),
|
||||
"user": os.getenv("DB_USER"),
|
||||
"password": os.getenv("DB_PASSWORD"),
|
||||
"port": int(os.getenv("DB_PORT", 5432)),
|
||||
}
|
||||
|
||||
# Check for required environment variables
|
||||
required_vars = ["DB_NAME", "DB_USER", "DB_PASSWORD"]
|
||||
missing_vars = [var for var in required_vars if not os.getenv(var)]
|
||||
|
||||
if missing_vars:
|
||||
raise ValueError(f"Missing required environment variables: {missing_vars}")
|
||||
|
||||
return cls(db_config)
|
||||
@@ -0,0 +1,220 @@
|
||||
"""Eloverblik API client for energy consumption data."""
|
||||
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Any
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from pydantic import AnyUrl
|
||||
|
||||
|
||||
class EloverblikClient:
|
||||
"""Client for interacting with the Eloverblik API."""
|
||||
|
||||
def __init__(
|
||||
self, refresh_token: str, server_url: str = "https://api.eloverblik.dk"
|
||||
):
|
||||
"""
|
||||
Initialize the Eloverblik client.
|
||||
|
||||
Args:
|
||||
refresh_token: The refresh token for authentication
|
||||
server_url: The base URL of the Eloverblik API
|
||||
"""
|
||||
self.refresh_token = refresh_token
|
||||
self.server_url = AnyUrl(server_url)
|
||||
self.access_token: str | None = None
|
||||
|
||||
# Ensure host is available
|
||||
if not self.server_url.host:
|
||||
raise ValueError(f"Invalid server URL: {server_url}")
|
||||
self.host = self.server_url.host
|
||||
|
||||
def get_access_token(self) -> bool:
|
||||
"""
|
||||
Exchange refresh token for access token.
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
conn = http.client.HTTPSConnection(self.host)
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.refresh_token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
try:
|
||||
conn.request("GET", "/customerapi/api/token", headers=headers)
|
||||
res = conn.getresponse()
|
||||
|
||||
if res.status == 200:
|
||||
token_response = res.read().decode("utf-8")
|
||||
token_data = json.loads(token_response)
|
||||
self.access_token = token_data["result"]
|
||||
return True
|
||||
else:
|
||||
error_data = res.read().decode("utf-8")
|
||||
print(f"Error getting access token: {error_data}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"Exception getting access token: {e}")
|
||||
return False
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_metering_points(
|
||||
self, include_all: bool = True
|
||||
) -> list[dict[str, Any]] | None:
|
||||
"""
|
||||
Get all metering points for the authenticated user.
|
||||
|
||||
Args:
|
||||
include_all: Whether to include all metering points
|
||||
|
||||
Returns:
|
||||
List of metering point data or None if error
|
||||
"""
|
||||
if not self.access_token:
|
||||
if not self.get_access_token():
|
||||
return None
|
||||
|
||||
conn = http.client.HTTPSConnection(self.host)
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.access_token}",
|
||||
"Content-Type": "application/json",
|
||||
"api-version": "1.0",
|
||||
}
|
||||
|
||||
params = {"includeAll": include_all}
|
||||
|
||||
try:
|
||||
conn.request(
|
||||
"GET",
|
||||
"/customerapi/api/meteringpoints/meteringpoints?" + urlencode(params),
|
||||
headers=headers,
|
||||
)
|
||||
res = conn.getresponse()
|
||||
|
||||
if res.status == 200:
|
||||
data = res.read().decode("utf-8")
|
||||
metering_data = json.loads(data)
|
||||
result = metering_data.get("result", [])
|
||||
return result if isinstance(result, list) else []
|
||||
else:
|
||||
error_data = res.read().decode("utf-8")
|
||||
print(f"Error getting metering points: {error_data}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
print(f"Exception getting metering points: {e}")
|
||||
return None
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_consumption_data(
|
||||
self,
|
||||
metering_point_id: str,
|
||||
date_from: str,
|
||||
date_to: str,
|
||||
aggregation: str = "Hour",
|
||||
) -> dict[str, Any] | None:
|
||||
"""
|
||||
Get consumption data for a specific metering point.
|
||||
|
||||
Args:
|
||||
metering_point_id: The metering point identifier
|
||||
date_from: Start date in YYYY-MM-DD format
|
||||
date_to: End date in YYYY-MM-DD format
|
||||
aggregation: Data aggregation level (Hour, Day, Month)
|
||||
|
||||
Returns:
|
||||
Consumption data or None if error
|
||||
"""
|
||||
if not self.access_token:
|
||||
if not self.get_access_token():
|
||||
return None
|
||||
|
||||
conn = http.client.HTTPSConnection(self.host)
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.access_token}",
|
||||
"Content-Type": "application/json",
|
||||
"api-version": "1.0",
|
||||
}
|
||||
|
||||
request_body = {"meteringPoints": {"meteringPoint": [metering_point_id]}}
|
||||
|
||||
try:
|
||||
conn.request(
|
||||
"POST",
|
||||
f"/customerapi/api/meterdata/gettimeseries/{date_from}/{date_to}/{aggregation}",
|
||||
body=json.dumps(request_body),
|
||||
headers=headers,
|
||||
)
|
||||
res = conn.getresponse()
|
||||
|
||||
if res.status == 200:
|
||||
data = res.read().decode("utf-8")
|
||||
result = json.loads(data)
|
||||
return result if isinstance(result, dict) else {}
|
||||
else:
|
||||
error_data = res.read().decode("utf-8")
|
||||
print(f"Error getting consumption data: {error_data}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
print(f"Exception getting consumption data: {e}")
|
||||
return None
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_consumption_data_last_days(
|
||||
self, metering_point_id: str, days: int = 7, aggregation: str = "Hour"
|
||||
) -> dict[str, Any] | None:
|
||||
"""
|
||||
Get consumption data for the last N days.
|
||||
|
||||
Args:
|
||||
metering_point_id: The metering point identifier
|
||||
days: Number of days to look back
|
||||
aggregation: Data aggregation level (Hour, Day, Month)
|
||||
|
||||
Returns:
|
||||
Consumption data or None if error
|
||||
"""
|
||||
end_date = datetime.now()
|
||||
start_date = end_date - timedelta(days=days)
|
||||
|
||||
date_from = start_date.strftime("%Y-%m-%d")
|
||||
date_to = end_date.strftime("%Y-%m-%d")
|
||||
|
||||
return self.get_consumption_data(
|
||||
metering_point_id, date_from, date_to, aggregation
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_env(
|
||||
cls, token_env_var: str = "ELOVERBLIK_API_TOKEN"
|
||||
) -> "EloverblikClient":
|
||||
"""
|
||||
Create client from environment variable.
|
||||
|
||||
Args:
|
||||
token_env_var: Name of environment variable containing the refresh token
|
||||
|
||||
Returns:
|
||||
EloverblikClient instance
|
||||
|
||||
Raises:
|
||||
ValueError: If token environment variable is not set
|
||||
"""
|
||||
token = os.getenv(token_env_var)
|
||||
if not token:
|
||||
raise ValueError(f"Environment variable {token_env_var} is not set")
|
||||
|
||||
return cls(refresh_token=token)
|
||||
@@ -0,0 +1,235 @@
|
||||
"""Scheduled synchronization functionality for CI/CD pipelines."""
|
||||
|
||||
import logging
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from .eloverblik import EloverblikClient
|
||||
from .database import DatabaseStorage
|
||||
from .utils import transform_metering_point_data, transform_consumption_data
|
||||
|
||||
|
||||
class ScheduledSync:
|
||||
"""Handles scheduled synchronization of energy consumption data."""
|
||||
|
||||
def __init__(self, client: EloverblikClient, db: DatabaseStorage):
|
||||
"""
|
||||
Initialize scheduled sync.
|
||||
|
||||
Args:
|
||||
client: Eloverblik API client
|
||||
db: Database storage instance
|
||||
"""
|
||||
self.client = client
|
||||
self.db = db
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
def sync_incremental(self, hours_back: int = 25) -> bool:
|
||||
"""
|
||||
Perform incremental sync - only sync recent data.
|
||||
|
||||
Args:
|
||||
hours_back: Number of hours to look back (default 25 to handle
|
||||
timezone issues and ensure no gaps)
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
self.logger.info(f"Starting incremental sync for last {hours_back} hours")
|
||||
|
||||
# Get metering points
|
||||
metering_points = self.client.get_metering_points()
|
||||
if not metering_points:
|
||||
self.logger.error("No metering points found")
|
||||
return False
|
||||
|
||||
success_count = 0
|
||||
total_readings = 0
|
||||
|
||||
for mp_data in metering_points:
|
||||
metering_point_id = mp_data["meteringPointId"]
|
||||
|
||||
try:
|
||||
# Store/update metering point data
|
||||
transformed_mp = transform_metering_point_data(mp_data)
|
||||
self.db.store_metering_point(transformed_mp)
|
||||
|
||||
# Get latest reading timestamp from database
|
||||
latest_timestamp = self.db.get_latest_reading_timestamp(
|
||||
metering_point_id
|
||||
)
|
||||
|
||||
# Calculate date range for sync
|
||||
end_date = datetime.now()
|
||||
|
||||
if latest_timestamp:
|
||||
# Sync from last reading with some overlap
|
||||
start_date = latest_timestamp - timedelta(hours=1)
|
||||
self.logger.info(
|
||||
f"Syncing {metering_point_id} from {start_date} (last reading: {latest_timestamp})"
|
||||
)
|
||||
else:
|
||||
# No previous data, sync last N hours
|
||||
start_date = end_date - timedelta(hours=hours_back)
|
||||
self.logger.info(
|
||||
f"No previous data for {metering_point_id}, syncing last {hours_back} hours"
|
||||
)
|
||||
|
||||
# Format dates for API
|
||||
date_from = start_date.strftime("%Y-%m-%d")
|
||||
date_to = end_date.strftime("%Y-%m-%d")
|
||||
|
||||
# Get consumption data
|
||||
consumption_data = self.client.get_consumption_data(
|
||||
metering_point_id, date_from, date_to
|
||||
)
|
||||
|
||||
if consumption_data:
|
||||
# Transform and store
|
||||
_, readings = transform_consumption_data(consumption_data)
|
||||
|
||||
if readings:
|
||||
if self.db.store_consumption_readings(
|
||||
metering_point_id, readings
|
||||
):
|
||||
total_readings += len(readings)
|
||||
self.logger.info(
|
||||
f"Stored {len(readings)} readings for {metering_point_id}"
|
||||
)
|
||||
else:
|
||||
self.logger.error(
|
||||
f"Failed to store readings for {metering_point_id}"
|
||||
)
|
||||
continue
|
||||
|
||||
success_count += 1
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error processing {metering_point_id}: {e}")
|
||||
continue
|
||||
|
||||
self.logger.info(
|
||||
f"Incremental sync completed: {success_count}/{len(metering_points)} "
|
||||
f"metering points, {total_readings} total readings"
|
||||
)
|
||||
|
||||
return success_count > 0
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Incremental sync failed: {e}")
|
||||
return False
|
||||
|
||||
def sync_full_backfill(self, days: int = 30) -> bool:
|
||||
"""
|
||||
Perform full backfill sync - sync large amount of historical data.
|
||||
|
||||
Args:
|
||||
days: Number of days to backfill
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
self.logger.info(f"Starting full backfill sync for last {days} days")
|
||||
|
||||
metering_points = self.client.get_metering_points()
|
||||
if not metering_points:
|
||||
self.logger.error("No metering points found")
|
||||
return False
|
||||
|
||||
success_count = 0
|
||||
total_readings = 0
|
||||
|
||||
for mp_data in metering_points:
|
||||
metering_point_id = mp_data["meteringPointId"]
|
||||
|
||||
try:
|
||||
# Store metering point data
|
||||
transformed_mp = transform_metering_point_data(mp_data)
|
||||
self.db.store_metering_point(transformed_mp)
|
||||
|
||||
# Get consumption data for the full period
|
||||
consumption_data = self.client.get_consumption_data_last_days(
|
||||
metering_point_id, days
|
||||
)
|
||||
|
||||
if consumption_data:
|
||||
_, readings = transform_consumption_data(consumption_data)
|
||||
|
||||
if readings:
|
||||
if self.db.store_consumption_readings(
|
||||
metering_point_id, readings
|
||||
):
|
||||
total_readings += len(readings)
|
||||
self.logger.info(
|
||||
f"Stored {len(readings)} readings for {metering_point_id}"
|
||||
)
|
||||
else:
|
||||
self.logger.error(
|
||||
f"Failed to store readings for {metering_point_id}"
|
||||
)
|
||||
continue
|
||||
|
||||
success_count += 1
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error processing {metering_point_id}: {e}")
|
||||
continue
|
||||
|
||||
self.logger.info(
|
||||
f"Full backfill completed: {success_count}/{len(metering_points)} "
|
||||
f"metering points, {total_readings} total readings"
|
||||
)
|
||||
|
||||
return success_count > 0
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Full backfill sync failed: {e}")
|
||||
return False
|
||||
|
||||
def health_check(self) -> bool:
|
||||
"""
|
||||
Perform a health check of the system.
|
||||
|
||||
Returns:
|
||||
True if all systems are healthy, False otherwise
|
||||
"""
|
||||
try:
|
||||
self.logger.info("Starting health check")
|
||||
|
||||
# Check API connection
|
||||
if not self.client.get_access_token():
|
||||
self.logger.error("Failed to get API access token")
|
||||
return False
|
||||
|
||||
# Check database connection
|
||||
if not self.db.connection:
|
||||
self.logger.error("No database connection")
|
||||
return False
|
||||
|
||||
# Try to get metering points
|
||||
metering_points = self.client.get_metering_points()
|
||||
if not metering_points:
|
||||
self.logger.error("No metering points found")
|
||||
return False
|
||||
|
||||
self.logger.info(
|
||||
f"Health check passed: {len(metering_points)} metering points available"
|
||||
)
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Health check failed: {e}")
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def from_env(cls) -> "ScheduledSync":
|
||||
"""
|
||||
Create ScheduledSync from environment variables.
|
||||
|
||||
Returns:
|
||||
ScheduledSync instance
|
||||
"""
|
||||
client = EloverblikClient.from_env()
|
||||
db = DatabaseStorage.from_env()
|
||||
return cls(client, db)
|
||||
@@ -0,0 +1,124 @@
|
||||
"""Utility functions for data transformation and processing."""
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
|
||||
def transform_metering_point_data(api_data: dict[str, Any]) -> dict[str, Any]:
|
||||
"""
|
||||
Transform API metering point data to database format.
|
||||
|
||||
Args:
|
||||
api_data: Raw metering point data from API
|
||||
|
||||
Returns:
|
||||
Dictionary formatted for database storage
|
||||
"""
|
||||
# Parse consumer start date if present
|
||||
consumer_start_date = None
|
||||
if api_data.get("consumerStartDate"):
|
||||
try:
|
||||
consumer_start_date = datetime.fromisoformat(
|
||||
api_data["consumerStartDate"].replace("Z", "+00:00")
|
||||
)
|
||||
except (ValueError, AttributeError):
|
||||
pass
|
||||
|
||||
return {
|
||||
"meteringPointId": api_data.get("meteringPointId"),
|
||||
"streetCode": api_data.get("streetCode"),
|
||||
"streetName": api_data.get("streetName"),
|
||||
"buildingNumber": api_data.get("buildingNumber"),
|
||||
"floorId": api_data.get("floorId"),
|
||||
"roomId": api_data.get("roomId"),
|
||||
"citySubDivisionName": api_data.get("citySubDivisionName"),
|
||||
"municipalityCode": api_data.get("municipalityCode"),
|
||||
"locationDescription": api_data.get("locationDescription"),
|
||||
"settlementMethod": api_data.get("settlementMethod"),
|
||||
"meterReadingOccurrence": api_data.get("meterReadingOccurrence"),
|
||||
"firstConsumerPartyName": api_data.get("firstConsumerPartyName"),
|
||||
"secondConsumerPartyName": api_data.get("secondConsumerPartyName"),
|
||||
"meterNumber": api_data.get("meterNumber"),
|
||||
"consumerStartDate": consumer_start_date,
|
||||
"typeOfMP": api_data.get("typeOfMP"),
|
||||
"balanceSupplierName": api_data.get("balanceSupplierName"),
|
||||
"postcode": api_data.get("postcode"),
|
||||
}
|
||||
|
||||
|
||||
def transform_consumption_data(
|
||||
consumption_json: dict[str, Any],
|
||||
) -> tuple[str | None, list[dict[str, Any]]]:
|
||||
"""
|
||||
Transform API consumption data to database format.
|
||||
|
||||
Args:
|
||||
consumption_json: Raw consumption data from API
|
||||
|
||||
Returns:
|
||||
Tuple of (metering_point_id, list_of_consumption_readings)
|
||||
"""
|
||||
if not consumption_json or "result" not in consumption_json:
|
||||
return None, []
|
||||
|
||||
try:
|
||||
# Extract the time series data
|
||||
market_document = consumption_json["result"][0]["MyEnergyData_MarketDocument"]
|
||||
time_series = market_document["TimeSeries"][0]
|
||||
metering_point_id = time_series["mRID"]
|
||||
periods = time_series["Period"]
|
||||
|
||||
consumption_readings = []
|
||||
|
||||
for period in periods:
|
||||
period_start = datetime.fromisoformat(
|
||||
period["timeInterval"]["start"].replace("Z", "+00:00")
|
||||
)
|
||||
resolution = period["resolution"]
|
||||
|
||||
if "Point" in period:
|
||||
for point in period["Point"]:
|
||||
# Calculate the actual timestamp for this point
|
||||
position = int(point["position"])
|
||||
# Position is 1-based, so subtract 1 to get hours offset
|
||||
hours_offset = position - 1
|
||||
|
||||
point_timestamp = period_start + timedelta(hours=hours_offset)
|
||||
|
||||
reading = {
|
||||
"timestamp": point_timestamp,
|
||||
"consumption_kwh": float(point["out_Quantity.quantity"]),
|
||||
"quality": point["out_Quantity.quality"],
|
||||
"period_resolution": resolution,
|
||||
}
|
||||
consumption_readings.append(reading)
|
||||
|
||||
return metering_point_id, consumption_readings
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error transforming consumption data: {e}")
|
||||
return None, []
|
||||
|
||||
|
||||
def calculate_consumption_stats(readings: list[dict[str, Any]]) -> dict[str, float]:
|
||||
"""
|
||||
Calculate consumption statistics from readings.
|
||||
|
||||
Args:
|
||||
readings: List of consumption reading dictionaries
|
||||
|
||||
Returns:
|
||||
Dictionary with consumption statistics
|
||||
"""
|
||||
if not readings:
|
||||
return {}
|
||||
|
||||
consumptions = [r["consumption_kwh"] for r in readings]
|
||||
|
||||
return {
|
||||
"total_kwh": sum(consumptions),
|
||||
"average_kwh": sum(consumptions) / len(consumptions),
|
||||
"min_kwh": min(consumptions),
|
||||
"max_kwh": max(consumptions),
|
||||
"count": len(consumptions),
|
||||
}
|
||||
Reference in New Issue
Block a user