Quick Guide

				
					import requests
import csv
import time
import os
import pytz
import json
from datetime import datetime
from enum import Enum
from typing import List
from pathlib import Path

class EnergyDataImport:
    """
    A utility class to make batch importing to EnergyData.dk easier from Python.
    
    This class handles building a CSV file in the correct format to be used for 
    importing to EnergyData.dk via the batch API. It manages the entire lifecycle:
    creating the proper CSV, uploading, validating, and ingesting the file.
    """

    # The API host for EnergyDataDK import endpoints
    API_HOST = 'https://admin.energydata.dk/api/v1/import'
    API_HEADERS = { 'Accept' : 'application/json' }

    # The format required by EnergyDataDK for timestamps
    TIMESTAMP_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ'

    class Status(Enum):
        """Internal state machine to track the progress of the import job."""
        UNINITIALIZED = 0
        OPEN = 1
        CLOSED = 2
        UPLOADING = 3
        STORED = 4
        VALIDATING = 5
        READY = 6
        INGESTING = 7
        DONE = 8
        ABORTED = 9
        ERROR = 10
    
    def __init__(
        self,
        upload_filename: str, 
        properties: List,  
        energydata_api_token: str,
        overwrite: bool = False, 
        tmp_dir: str = '/tmp/energydata_batch_upload', 
        autoclean_tmp_files: bool = True):
        """
        Constructor for class. Create an instance for each batch upload. 
        An instance cannot be reused between multiple imports.

        Args:
            upload_filename (str): Name for the import file generated locally and used on the server.
            properties (List): List of property IDs or Topics to which data will be added.
            energydata_api_token (str): API token from https://portal.energydata.dk/user#accesstokens.
            overwrite (bool): Controls whether to overwrite existing local or remote files.
            tmp_dir (str): Directory for temporary CSV storage before upload.
            autoclean_tmp_files (bool): If True, removes local files once the context is closed.
        """
        self.status = self.Status.UNINITIALIZED
        self.import_id = None
        self.upload_url = None
        self.added_lines = 0
        self.previous_ts = None
        self.upload_filename = upload_filename
        self.local_file_path = Path.joinpath(Path(tmp_dir), Path(upload_filename))
        self.properties = properties
        self.energydata_api_token = energydata_api_token
        self.overwrite = overwrite
        self.tmp_dir = tmp_dir
        self.autoclean_tmp_files = autoclean_tmp_files
        self.api_headers = dict(self.API_HEADERS)
        self.api_headers.update({'Authorization': f'Bearer {energydata_api_token}'})

    def __enter__(self):
        """
        Prepares the environment: creates the temp directory and opens the CSV file for writing.
        """
        self.__assert_status(self.Status.UNINITIALIZED)
        Path(self.tmp_dir).mkdir(parents=True, exist_ok=True)
        
        if not self.overwrite and Path.exists(self.local_file_path):
            raise Exception(f"File '{self.upload_filename}' already exists. Set overwrite=True.")
            
        self.fd = open(self.local_file_path, mode='w', newline='')
        # CSV format: delimiter ';', double quotes for strings, non-numeric quoting
        self.writer = csv.writer(self.fd, quoting=csv.QUOTE_NONNUMERIC, delimiter=';', 
                                quotechar='"', escapechar='\\', doublequote=False)
        # Header row: First column is empty (reserved for timestamp), followed by property list
        self.writer.writerow([''] + self.properties)

        self.__change_status(self.Status.UNINITIALIZED, self.Status.OPEN)
        return self

    def __exit__(self, type, value, traceback):
        """Closes the file descriptor and handles automatic cleanup of local files."""
        if hasattr(self, 'fd') and not self.fd.closed: self.fd.close()
        if self.autoclean_tmp_files and os.path.exists(self.local_file_path): 
            os.remove(self.local_file_path)
        self.status = self.Status.CLOSED

    def add_values(self, time: datetime, values: List):
        """
        Adds a row of data to the import buffer.

        Args:
            time (datetime): Timestamp. Must be timezone-aware.
            values (List): Data values matching the number of properties in the constructor.

        Raises:
            Exception: If time is not timezone-aware or if timestamps are not monotonically increasing.
        """
        self.__assert_status(self.Status.OPEN)
        if time.tzinfo == None:
            raise Exception("No timezone specified for the datetime object.")
        if len(values) != len(self.properties):
            raise Exception(f"Expected {len(self.properties)} values, found {len(values)}.")
        if self.previous_ts is not None and time <= self.previous_ts:
            raise Exception("Added timestamps must be strictly increasing.")
        
        self.previous_ts = time
        # Convert to UTC ISO format before writing
        self.writer.writerow(
            [time.astimezone(pytz.UTC).strftime(self.TIMESTAMP_FORMAT)] + list(values)
        )
        self.added_lines += 1

    def upload(self, print_progress = True):
        """
        Finalizes the CSV and uploads it to the server storage. Once called, state 
        transitions from OPEN to UPLOADING and finally to STORED.
        """
        self.__change_status(self.Status.OPEN, self.Status.UPLOADING)
        self.fd.close()
        
        # 1. Create the import job record
        res = requests.post(url=self.API_HOST, headers=self.api_headers, params={'importname': self.upload_filename})
        res.raise_for_status()
        self.import_id = res.json()['id']

        # 2. Get the secure S3 upload URL
        res = requests.get(url=f'{self.API_HOST}/{self.import_id}/upload_url', headers=self.api_headers)
        res.raise_for_status()
        self.upload_url = res.json()['upload_url'].replace("\\", "") 
        
        if print_progress: print(f'Starting file upload for job id: {self.import_id}')
        
        # 3. Upload the binary data
        with open(self.local_file_path, 'rb') as f:
            res = requests.put(self.upload_url, data=f, headers={"Content-Type": "application/octet-stream"})
        res.raise_for_status()

        self.__change_status(self.Status.UPLOADING, self.Status.STORED)
        if print_progress: print(f'Successfully stored job id {self.import_id}')

    def validate(self, errors_limit = 0, block = True, print_progress = True):
        """
        Triggers server-side validation. State transitions from STORED to VALIDATING 
        and finally to READY.
        """
        self.__change_status(self.Status.STORED, self.Status.VALIDATING)
        self.__api_put_request(block, 'validate', lambda s: self.__validation_progress(s, print_progress), {'errors_limit': errors_limit})

    def ingest(self, block = True, print_progress = True):
        """
        Triggers final ingestion. State transitions from READY to INGESTING 
        and finally to DONE.
        """
        self.__change_status(self.Status.READY, self.Status.INGESTING)
        self.__api_put_request(block, 'ingest', lambda s: self.__ingestion_progress(s, print_progress))

    def __api_put_request(self, block, path, progress_callback, body = None):
        """Helper to send PUT requests and poll for status updates."""
        res = requests.put(url=f'{self.API_HOST}/{self.import_id}/{path}', headers=self.api_headers, json=body)
        res.raise_for_status()
        while block:
            time.sleep(5)
            status = self.__get_status()
            if not progress_callback(status): break

    def __get_status(self):
        """Retrieves the current job metadata from the server."""
        res = requests.get(url=f'{self.API_HOST}/{self.import_id}', headers=self.api_headers)
        res.raise_for_status()
        return res.json()

    def __assert_status(self, expected):
        if expected != self.status:
            raise Exception(f"Expected status '{expected}', but found '{self.status}'")

    def __change_status(self, expected, to):
        self.__assert_status(expected)
        self.status = to
    
    def __validation_progress(self, status, print_progress):
        if print_progress: print(f'Validated {status["validated_lines"]}/{self.added_lines} lines')
        if status['status'] == 'validating': return True
        if status['status'] == 'ready':
            self.__change_status(self.Status.VALIDATING, self.Status.READY)
            return False
        raise Exception(f'Validation failed: {status["status"]}')

    def __ingestion_progress(self, status, print_progress):
        if print_progress: print(f'Ingested {status["ingested_lines"]}/{self.added_lines} lines')
        if status['status'] == 'ingesting': return True
        if status['status'] == 'done':
            self.__change_status(self.Status.INGESTING, self.Status.DONE)
            print("******Data uploaded successfully******")
            return False
        raise Exception(f'Ingestion failed: {status["status"]}')