stream.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# System modules
import json
from time import sleep
import subprocess
import requests

# Third-party modules
from astropy.time import Time
from flask import Response, Blueprint

# Custom modules
from utils.url_stuff import api_base_url
from web.server.scheduler import scheduler

web = Blueprint('stream', __name__)

def req(address, timeout=1, json=True):
    """Simple GET request managing some exceptions"""
    
    try:
        res = requests.get(address, timeout=timeout)
        if res.status_code != 200:
            return False
        else:
            return res.json() if json else res
    except requests.exceptions.ConnectTimeout as e:
        return None
    except requests.exceptions.ConnectionError as e:
        return None
    except requests.exceptions.ReadTimeout as e:
        return None

class Container():
    def __init__(self):
        self.tel_last = {}
        self.dom_last = {}
        self.cam_last = {}
        self.env_last = {}
        self.ipcam_last = {}
        self.seq_last = {}

container = Container()
        
############################
# Schedulers
############################

@scheduler.task('interval', seconds=2, id="tel", max_instances=1)
def tel_job():
    try:
        container.tel_last = req(api_base_url+"/api/telescope/status")
    except Exception as e:
        print(f"Probably killed web app: {e}")

@scheduler.task('interval', seconds=2, id="dom", max_instances=1)
def dom_job():
    try:
        container.dom_last = req(api_base_url+"/api/dome/status")
    except Exception as e:
        print(f"Probably killed web app: {e}")

@scheduler.task('interval', seconds=2, id="cam", max_instances=1)
def cam_job():
    try:
        container.cam_last = req(api_base_url+"/api/camera/status")
    except Exception as e:
        print(f"cam job Probably killed web app: {e}")

@scheduler.task('interval', seconds=60, id="env", max_instances=1)
def env_job():
    try:
        container.env_last = req(api_base_url+"/api/environment/status", timeout=50)
    except SystemExit as e:
        print(f"Env status get error: {e}")
    except Exception as e:
        print(f"Probably killed web app: {e}")

@scheduler.task('interval', seconds=1, id="ipcam", max_instances=1)
def ipcam_job():
    try:
        container.ipcam_last = req(api_base_url+"/api/webcam/snapshot")
    except SystemExit as e:
        print(f"Webcam status get error: {e}")
    except RuntimeError as e:
        print(f"Probably killed web app: {e}")

@scheduler.task('interval', seconds=1, id="seq", max_instances=1)
def seq_job():
    try:
        container.seq_last = req(api_base_url+"/api/sequencer/run")
        #print(cam_container.last)
    except Exception as e:
        print(f"seq job Probably killed web app: {e}")

########
# STREAM
########

@web.route('/status')
def stream_status():

    def inner():

        previous_ts = {}
        previous_ds = {}
        previous_cs = {}
        previous_sq = {}

        while True:

            try:
                sleep(1)

                ###########################
                # Stream telescope status #
                ###########################

                # Complete for control panel
                ts = container.tel_last

                if previous_ts != ts:
                    yield 'event: telescope\ndata: {}\n\n'.format(json.dumps(ts))
                previous_ts = ts

                ######################
                # Stream dome status #
                ######################

                ds = container.dom_last

                if previous_ds != ds:
                    yield 'event: dome\ndata: {}\n\n'.format(json.dumps(ds))
                previous_ds = ds

                ########################
                # Stream camera status #
                ########################

                cs = container.cam_last
                #print(cs)
                if previous_cs != cs:
                    yield 'event: camera\ndata: {}\n\n'.format(json.dumps(cs))
                previous_cs = cs

                ###########################
                # Stream sequencer status #
                ###########################

                sq = container.seq_last
                #print(sq)
                if previous_sq != sq:
                    yield 'event: sequencer\ndata: {}\n\n'.format(json.dumps(sq))
                previous_sq = sq

            except GeneratorExit:
                print(f"Stop streaming sequencer")
                return

    print("Streaming status...")
    return Response(inner(), mimetype="text/event-stream")


@web.route('/environment')
def stream_environment():

    def inner():
        previous_es = {}
        while True:

            try:
                sleep(20)

                #############################
                # Stream environment status #
                #############################

                es = container.env_last

                if previous_es != es:
                    yield 'event: environment\ndata: {}\n\n'.format(json.dumps(es))
                previous_es = es

            except GeneratorExit:
                print(f"Stop streaming environment")
                return

    print("Streaming environment...")
    return Response(inner(), mimetype="text/event-stream")



@web.route('/webcam')
def stream_webcam():
    def inner():
        while True:

            try:
                sleep(0.5)

                ##########################
                # Stream webcam snapshot #
                ##########################

                try:
                    img = container.ipcam_last["response"].encode("ISO-8859-1")
                except:
                    img = None
                    
                if img:
                    yield (b'--frame\r\n'
                           b'Content-Type: image/jpeg\r\n\r\n' + bytes(img) + b'\r\n')

            except GeneratorExit:
                print(f"Stop streaming webcam")
                return

    return Response(inner(), mimetype="multipart/x-mixed-replace; boundary=frame")


@web.route('/logfile')
def stream_logfile():
    def inner(today):

        proc = subprocess.Popen(
            [f"tail -f -n30 ./data/log/OARPAF.{today}.log"],
            shell=True,
            stdout=subprocess.PIPE
        )

        try:

            for line in iter(proc.stdout.readline, ''):
                line = line.rstrip().decode()

                if not line:
                    sleep(0.05)
                    yield "event: logfile\ndata: \n\n"
                    continue

                sleep(0.05)
                yield f"event: logfile\ndata: {line}\n\n"

        except GeneratorExit:
            print("qui")
            print("Stop streaming log")
            print("Closing subprocess")
            proc.terminate()
            proc.wait()
            print("Subprocess closed")

    print("Streaming template log...")
    today = Time.now().iso.split()[0]
    return Response(inner(today), mimetype="text/event-stream")