Quick Start

To get started with KOA Middleware, follow these basic steps:

  1. Installation: Ensure the package is installed as described in the Installation section.

  2. Initialize the Calibration Store: The CalibrationStore is the central component for managing calibration data. You’ll need to provide an ORM class that defines your calibration data structure.

[ ]:

import os
from koa_middleware.store import CalibrationStore
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# Define a simple ORM class for demonstration
class Base(DeclarativeBase):
    pass

class MyCalibrationORM(Base):
    __tablename__ = 'my_calibrations'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    filename = Column(String) # Required for CalibrationStore

# Set up a temporary cache directory for demonstration
cache_dir = '/tmp/koa_middleware_cache'
os.makedirs(cache_dir, exist_ok=True)
os.environ['KOA_CALIBRATION_CACHE'] = cache_dir

# Initialize the store
store = CalibrationStore(orm_class=MyCalibrationORM)

print(f"CalibrationStore initialized with cache: {store.cache_dir}")

  1. Retrieve Calibrations: You can retrieve calibrations using selectors or by ID.

[ ]:

# Example: Register a dummy calibration (in a real scenario, this would come from a DRP)
class DummyCalibration:
    def save(self, output_dir):
        filepath = os.path.join(output_dir, "dummy_cal.fits")
        with open(filepath, "w") as f:
            f.write("dummy calibration data")
        return filepath
    def to_orm(self):
        return MyCalibrationORM(id=1, name="dummy_cal", filename="dummy_cal.fits")

dummy_cal = DummyCalibration()
local_filepath, orm_instance = store.register_local_calibration(dummy_cal)
print(f"Registered calibration: {orm_instance.name} at {local_filepath}")

# Example: Retrieve by ID
retrieved_filepath, retrieved_orm = store.get_calibration_by_id(1)
if retrieved_orm:
    print(f"Retrieved calibration by ID: {retrieved_orm.name} at {retrieved_filepath}")

# Clean up (optional)
store.close()
# os.remove(os.path.join(cache_dir, 'database', 'my_calibrations_calibrations.db'))
# os.remove(os.path.join(cache_dir, 'calibrations', 'dummy_cal.fits'))
# os.rmdir(os.path.join(cache_dir, 'database'))
# os.rmdir(os.join(cache_dir, 'calibrations'))
# os.rmdir(cache_dir)