Task Manager

.github/workflows/pythontest.yaml

@@ -6,7 +6,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [ "3.8" ]
+        python-version: [ "3.10" ]
     steps:
       - name: Checkout code
         uses: actions/checkout@v2
@@ -25,7 +25,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [ "3.8", "3.9", "3.10" ]
+        python-version: [ "3.10" ]
     steps:
       - name: Checkout code
         uses: actions/checkout@v2
@@ -34,7 +34,7 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
       - name: Install
-        run: pip install ".[google,focuser,sensors,testing]"
+        run: pip install ".[google,focuser,sensors,tasks,testing]"
       - name: Test
         run: pytest
       - name: Upload coverage report to codecov.io

conf_files/pocs.yaml

@@ -32,6 +32,16 @@ directories:
   mounts: resources/mounts
   fields: conf_files/fields
 
+# Celery task manager for distributed tasks.
+celery:
+  broker_url: redis://localhost:6379
+  result_backend: redis://localhost:6379
+  service:
+    - name: pocs-celery
+      image: redis
+      ports:
+        6379: 6379
+
 db:
   name: panoptes
   type: file

setup.cfg

@@ -18,7 +18,7 @@ classifiers =
     License :: OSI Approved :: MIT License
     Operating System :: POSIX
     Programming Language :: Python :: 3
-    Programming Language :: Python :: 3.8
+    Programming Language :: Python :: 3.10
     Programming Language :: Python :: 3 :: Only
     Topic :: Scientific/Engineering :: Astronomy
     Topic :: Scientific/Engineering :: Physics
@@ -52,7 +52,7 @@ install_requires =
 # The usage of test_requires is discouraged, see `Dependency Management` docs
 # tests_require = pytest; pytest-cov
 # Require a specific Python version, e.g. Python 2.7 or >= 3.4
-python_requires = >=3.8
+python_requires = >="3.10"
 
 packages = find_namespace:
 [options.packages.find]
@@ -72,6 +72,9 @@ google =
     gsutil
     protobuf
     rsa
+tasks =
+    celery[redis]
+    docker
 testing =
     coverage
     pycodestyle

src/panoptes/pocs/camera/gphoto/celery/controller.py

@@ -0,0 +1,57 @@
+from typing import List, Union
+
+from panoptes.pocs.camera.gphoto.canon import Camera as CanonCamera
+from panoptes.pocs.utils import error
+from panoptes.pocs.utils.tasks import TaskManager, RunTaskMixin
+
+
+class Camera(CanonCamera, RunTaskMixin):
+    """A remote gphoto2 camera class that can call local or remote celery tasks."""
+
+    def __init__(self, queue: str | None = None, *args, **kwargs):
+        """Control a remote gphoto2 camera via a celery task. """
+        super().__init__(connect=False, *args, **kwargs)
+
+        self.celery_app = TaskManager.create_celery_app_from_config()
+
+        self.task = None
+        self.queue = queue or self.name
+        self.connect()
+        self.logger.debug(f'Canon DSLR GPhoto2 camera celery task manager with queue={self.queue}')
+
+    @property
+    def is_exposing(self):
+        return self.task and self.task.state == 'EXPOSING'
+
+    def command(self, cmd, queue=None, **kwargs):
+        """Run a remote celery task attached to a camera. """
+
+        if self.is_exposing:
+            raise error.CameraBusy()
+
+        queue = queue or self.queue
+
+        arguments = ' '.join(cmd)
+
+        self.logger.debug(f'Running remote gphoto2 task with {arguments=} to {queue=}')
+        self.task = self.call_task('camera.command', args=[arguments], queue=queue)
+
+    def get_command_result(self, timeout: float = 10) -> Union[List[str], None]:
+        """Get the output from the remote camera task, blocking up to timeout."""
+        cmd_result = self.task.get(timeout=timeout)
+
+        self.logger.debug(f'Full results from command {cmd_result!r}')
+
+        # Clear task.
+        self.task = None
+
+        # Return just the actual output. TODO error checking?
+        return cmd_result['output']
+
+    def _create_fits_header(self, seconds, dark=None, metadata=None) -> dict:
+        fits_header = super(Camera, self)._create_fits_header(seconds, dark=dark, metadata=metadata)
+        return {k.lower(): v for k, v in dict(fits_header).items()}
+
+    def _start_exposure(self, seconds=None, *args, **kwargs):
+        # TODO more here
+        self.task = self.call_task('camera.release_shutter', args=[seconds], queue=self.queue)

src/panoptes/pocs/camera/gphoto/celery/settings.py

@@ -0,0 +1,44 @@
+import typing
+from enum import IntEnum
+from typing import Dict, Optional
+
+import pigpio
+from pydantic import BaseSettings, BaseModel, Field
+
+from worker import gpio
+
+
+class State(IntEnum):
+    LOW = 0
+    HIGH = 1
+
+
+class Settings(BaseSettings):
+    camera_name: str
+    camera_port: str
+    camera_pin: int
+    broker_url: str = 'amqp://guest:guest@localhost:5672//'
+    result_backend: str = 'rpc://'
+
+    class Config:
+        env_prefix = 'pocs_'
+
+
+class Camera(BaseModel):
+    """A camera with a shutter release connected to a gpio pin."""
+    name: str
+    port: str
+    pin: int
+    is_tethered: bool = False
+
+    def setup_pin(self):
+        """Sets the mode for the GPIO pin."""
+        # Get GPIO pin and set OUTPUT mode.
+        print(f'Setting {self.pin=} as OUTPUT for {self.name}')
+        gpio.set_mode(self.pin, pigpio.OUTPUT)
+
+
+class AppSettings(BaseModel):
+    celery: Dict = Field(default_factory=dict)
+    camera: Camera
+    process: Optional[typing.Any] = None

src/panoptes/pocs/camera/gphoto/celery/worker.py

@@ -0,0 +1,158 @@
+import re
+import shutil
+import subprocess
+from contextlib import suppress
+from typing import Optional, List, Union
+
+import pigpio
+from celery import Celery
+from panoptes.utils.time import current_time, CountdownTimer
+
+from panoptes.pocs.camera.gphoto.celery.settings import State, Settings, Camera, AppSettings
+
+# Create settings from env vars.
+settings = Settings()
+
+# Build app settings.
+app_settings = AppSettings(
+    camera=Camera(name=settings.camera_name,
+                  port=settings.camera_port,
+                  pin=settings.camera_pin),
+    celery=dict(broker_url=settings.broker_url,
+                result_backend=settings.result_backend),
+)
+
+# Start celery.
+app = Celery()
+app.config_from_object(app_settings.celery)
+
+# Setup GPIO pins.
+gpio = pigpio.pi()
+app_settings.camera.setup_pin()
+
+camera_match_re = re.compile(r'([\w\d\s_.]{30})\s(usb:\d{3},\d{3})')
+file_save_re = re.compile(r'Saving file as (.*)')
+
+
+@app.task(name='camera.release_shutter', bind=True)
+def release_shutter(self, exptime: float):
+    """Trigger the shutter release for given exposure time via the GPIO pin."""
+    # Create a timer.
+    timer = CountdownTimer(exptime, name=f'Pin{app_settings.camera.pin}Expose')
+
+    # Open shutter.
+    self.update_state(state='START_EXPOSING', start_time=current_time(flatten=True))
+    gpio.write(app_settings.camera.pin, State.HIGH)
+
+    # Wait for exptime, send state updates.
+    while timer.expired() is False:
+        self.update_state(state='EXPOSING', meta=dict(secs=f'{exptime - timer.time_left():.02f}', ))
+        timer.sleep(max_sleep=max(1., exptime / 8))  # Divide wait time into eighths.
+
+    # Close shutter.
+    gpio.write(app_settings.camera.pin, State.LOW)
+    self.update_state(state='STOP_EXPOSING', stop_time=current_time(flatten=True))
+
+
+@app.task(name='camera.start_tether', bind=True)
+def start_gphoto2_tether(self, filename_pattern: str):
+    """Start a tether for gphoto2 auto-download."""
+    if app_settings.camera.is_tethered:
+        print(f'{app_settings.camera} is already tethered')
+        return
+    else:
+        print(f'Starting gphoto2 tether for {app_settings.camera.port=} using {filename_pattern=}')
+        app_settings.camera.is_tethered = True
+
+    command = ['--filename', filename_pattern, '--capture-tethered']
+    full_command = _build_gphoto2_command(command)
+
+    # Start tether process.
+    app_settings.process = subprocess.Popen(full_command,
+                                            stderr=subprocess.STDOUT,
+                                            stdout=subprocess.PIPE)
+    print(f'gphoto2 tether started for {app_settings.camera} on {app_settings.process.pid=}')
+
+
+@app.task(name='camera.stop_tether')
+def stop_gphoto2_tether():
+    """Tells camera to stop gphoto2 tether."""
+    print(f'Stopping gphoto2 tether for {app_settings.camera}')
+    # Communicate and kill immediately.
+    try:
+        outs, errs = app_settings.process.communicate(timeout=1)
+    except subprocess.TimeoutExpired:
+        app_settings.process.kill()
+        outs, errs = app_settings.process.communicate()
+
+    app_settings.camera.is_tethered = False
+
+    return dict(outs=outs.decode('utf-8'), errs=errs.decode('utf-8'))
+
+
+@app.task(name='camera.file_download', bind=True)
+def gphoto_file_download(self,
+                         filename_pattern: str,
+                         only_new: bool = True
+                         ):
+    """Downloads (newer) files from the camera on the given port using the filename pattern."""
+    print(f'Starting gphoto2 download for {app_settings.camera} using {filename_pattern=}')
+    command = ['--filename', filename_pattern, '--get-all-files', '--recurse']
+    if only_new:
+        command.append('--new')
+
+    results = gphoto2_command(command, timeout=600)
+    filenames = list()
+    for line in results['output']:
+        file_match = file_save_re.match(line)
+        if file_match is not None:
+            fn = file_match.group(1).strip()
+            print(f'Found match {fn}')
+            filenames.append(fn)
+            self.update_state(state='DOWNLOADING', meta=dict(directory=fn))
+
+    return filenames
+
+
+@app.task(name='camera.delete_files', bind=True)
+def gphoto_file_delete(self):
+    """Removes all files from the camera on the given port."""
+    print(f'Deleting all files for {app_settings.camera}')
+    gphoto2_command('--delete-all-files --recurse')
+
+
+@app.task(name='camera.command', bind=True)
+def gphoto_task(self, command: Union[List[str], str]):
+    """Perform arbitrary gphoto2 command.."""
+    print(f'Calling {command=} on {app_settings.camera}')
+    return gphoto2_command(command)
+
+
+def gphoto2_command(command: Union[List[str], str], timeout: Optional[float] = 300) -> dict:
+    """Perform a gphoto2 command."""
+    full_command = _build_gphoto2_command(command)
+    print(f'Running gphoto2 {full_command=}')
+
+    completed_proc = subprocess.run(full_command, capture_output=True, timeout=timeout)
+
+    # Populate return items.
+    command_output = dict(
+        success=completed_proc.returncode >= 0,
+        returncode=completed_proc.returncode,
+        output=completed_proc.stdout.decode('utf-8').split('\n'),
+        error=completed_proc.stderr.decode('utf-8').split('\n')
+    )
+
+    return command_output
+
+
+def _build_gphoto2_command(command: Union[List[str], str]):
+    full_command = [shutil.which('gphoto2'), '--port', app_settings.camera.port]
+
+    # Turn command into a list if not one already.
+    with suppress(AttributeError):
+        command = command.split(' ')
+
+    full_command.extend(command)
+
+    return full_command

src/panoptes/pocs/utils/cli/main.py

@@ -3,6 +3,7 @@
 from panoptes.pocs.utils.cli import config
 from panoptes.pocs.utils.cli import sensor
 from panoptes.pocs.utils.cli import image
+from panoptes.pocs.utils.cli import tasks
 from panoptes.pocs.utils.cli import power
 
 app = typer.Typer()
@@ -12,6 +13,7 @@
 app.add_typer(sensor.app, name="sensor", help='Interact with system sensors.')
 app.add_typer(power.app, name="power", help='Interact with power relays.')
 app.add_typer(image.app, name="image", help='Interact with images.')
+app.add_typer(tasks.app, name="tasks", help='Interact with the TaskManager.')
 
 
 @app.callback()