controller.py

import win32com.client
import datetime
import time
import os
import sys
import logging
import uuid
from astropy.io import fits

filter_positions = [
    "Luminance",
    "Red",
    "Green",
    "Blue",
    "g'",
    "r'",
    "i'",
    "z'",
    "u'",
    "B",
    "V",
    "R",
    "I'",
    "Ha",
    "dark",
    "open",
]


class Controller:
    telescope = None
    camera = None
    filter_wheel = None
    logger = logging.getLogger(__name__)

    def __init__(self, telescope_prog_id, camera_prog_id, filter_wheel_prog_id):

        logging.basicConfig(
            filename="controller.log",
            format="%(asctime)s %(levelname)s:%(message)s",
            level=logging.DEBUG,
        )
        if Controller.telescope is None:
            try:
                Controller.telescope = win32com.client.Dispatch(telescope_prog_id)
            except Exception as e:
                Controller.logger.error("Error connecting to telescope: " + str(e))
                raise IOError("Error connecting to telescope: " + str(e))
            Controller.telescope.Connected = True
            Controller.logger.info("Connected to telescope")
            print("Connected to telescope")

        if Controller.camera is None:
            try:
                Controller.camera = win32com.client.Dispatch(camera_prog_id)
            except Exception as e:
                Controller.logger.error("Error connecting to camera: " + str(e))
                raise IOError("Error connecting to camera: " + str(e))
            Controller.camera.Connected = True
            Controller.logger.info("Connected to camera")
            print("Connected to camera")

        if Controller.filter_wheel is None and filter_wheel_prog_id is not None and filter_wheel_prog_id != "None":
            with HiddenPrints():
                try:
                    Controller.filter_wheel = win32com.client.Dispatch(
                        filter_wheel_prog_id
                    )
                except Exception as e:
                    Controller.logger.error(
                        "Error connecting to filter wheel: " + str(e)
                    )
                    raise IOError("Error connecting to filter wheel: " + str(e))
                Controller.filter_wheel.Connected = True
            Controller.logger.info("Connected to filter wheel")
            print("Connected to filter wheel")

    def disconnect_all(self):
        self.disconnect_telescope()
        self.disconnect_camera()
        self.disconnect_filter_wheel()

    def disconnect_telescope(self):
        if Controller.telescope and Controller.telescope.Connected:
            Controller.telescope.Tracking = False
            Controller.telescope.Connected = False
            Controller.logger.info("Disconnected from telescope")

    def disconnect_camera(self):
        if Controller.camera:
            Controller.camera.Connected = False
            Controller.logger.info("Disconnected from camera")

    def disconnect_filter_wheel(self):
        if Controller.filter_wheel:
            Controller.filter_wheel.Connected = False
            Controller.logger.info("Disconnected from filter wheel")

    def move_to_target(self, altitude, azimuth):
        try:
            altitude = float(altitude)
            azimuth = float(azimuth)
            if altitude < 35 or altitude > 90:
                raise ValueError("Altitude must be between 30 and 90")
            if azimuth < 0 or azimuth > 360:
                raise ValueError("Azimuth must be between 0 and 360")
        except ValueError as e:
            Controller.logger.error("Coordinates invalid: " + str(e))
            self.disconnect_all()
            raise e

        Controller.logger.info("Slewing to alt: {} az: {}".format(altitude, azimuth))
        Controller.telescope.SlewToAltAz(azimuth, altitude)
        # Wait for telescope to finish slewing
        while Controller.telescope.Slewing:
            time.sleep(0.5)
        # Ensure tracking after slewing
        Controller.telescope.Tracking = True
        Controller.logger.info("Slew finished, tracking on")

    def change_filter(self, filter_number):
        if Controller.filter_wheel is None:
            Controller.logger.warn("No filter wheel")
            return
        try:
            filter_number = int(filter_number)
            if filter_number < 0 or filter_number > 15:
                raise ValueError("Filter number must be between 0 and 15")
        except ValueError as e:
            Controller.logger.error("Filter number invalid: " + str(e))
            self.disconnect_all()
            raise e
        Controller.logger.info("Changing filter to: {}".format(filter_number))
        Controller.filter_wheel.Position = filter_number
        while Controller.filter_wheel.Position == -1:
            time.sleep(0.1)
        Controller.logger.info("Filter change complete")

    def take_exposure(self, exposure_time, save_image=False):
        # Take an exposure with the camera
        Controller.logger.info("Taking exposure of {} seconds".format(exposure_time))
        start_time = datetime.datetime.utcnow()
        Controller.camera.StartExposure(exposure_time, True)
        end_time = start_time + datetime.timedelta(seconds=exposure_time)
        while Controller.camera.ImageReady == False:
            time.sleep(0.1)
        Controller.logger.info("Exposure complete")
        if save_image:
            image = Controller.camera.ImageArray
            file_id = uuid.uuid4().hex
            filename = os.path.join(
                os.path.abspath(os.path.dirname(__file__)),
                "image/{}.fits".format(file_id),
            )
            hdu = fits.PrimaryHDU(image)
            header = hdu.header
            header["EXPTIME"] = exposure_time
            header["DATE-OBS"] = start_time.isoformat()
            header["FILTER"] = filter_positions[Controller.filter_wheel.Position]
            header["END_TIME"] = end_time.isoformat()
            header["FILE-ID"] = file_id
            hdul = fits.HDUList([hdu])
            # Write the FITS file
            hdul.writeto(filename, overwrite=True)
            Controller.logger.info("Image saved to: " + filename)
            return (start_time, end_time, file_id, hdul)
        return (start_time, end_time)


class HiddenPrints:
    def __enter__(self):
        self._original_stdout = sys.stdout
        sys.stdout = open(os.devnull, "w")

    def __exit__(self, exc_type, exc_val, exc_tb):
        sys.stdout.close()
        sys.stdout = self._original_stdout