Skip to content

Commit

Permalink
Add start of boundary_condition
Browse files Browse the repository at this point in the history
  • Loading branch information
@WWGolay
WWGolay committed Jan 30, 2025
1 parent eaae7ca commit 57de984
Show file tree
Hide file tree
Showing 3 changed files with 179 additions and 70 deletions.
7 changes: 5 additions & 2 deletions pyscope/db.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ def uuid(cls) -> Mapped[Uuid]:
attempting to update it again.
For more information, see the
[SQLAlchemy documentation](https://docs.sqlalchemy.org/en/20/orm/versioning.html)
`SQLAlchemy documentation <https://docs.sqlalchemy.org/en/20/orm/versioning.html>`__
on versioning.
"""

Expand Down Expand Up @@ -128,6 +128,9 @@ def uuid(cls) -> Mapped[Uuid]:
that identifies the new class.
For more information, see the
[SQLAlchemy documentation](https://docs.sqlalchemy.org/en/20/orm/inheritance.html)
`SQLAlchemy documentation <https://docs.sqlalchemy.org/en/20/orm/inheritance.html)>`__
on inheritance.
"""


Base.__init__.__doc__ = ""
15 changes: 15 additions & 0 deletions pyscope/scheduling/_block.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -315,3 +315,18 @@ def _end_time_update_expression(
cls, value: DateTime
) -> List[Tuple[DateTime, DateTime]]:
return [(cls.start_time, type_coerce(value - cls.duration, DateTime))]


'''
_Block.__annotations__["uuid"] = Mapped[Uuid]
_Block.uuid.__doc__ = \
"""
The universally unique identifier (UUID) for the database entry
corresponding to the object. This UUID is generated automatically with
`uuid.uuid4` when the object is created and is used to uniquely
identify the object in the database. The UUID is a primary key for
the table and is required to be unique for each entry. The UUID is
not intended to be used as a human-readable identifier and should
not be relied upon for that purpose.
"""
'''
227 changes: 159 additions & 68 deletions pyscope/scheduling/boundary_condition.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,10 @@

import logging
from dataclasses import InitVar
from typing import Callable, Tuple

from sqlalchemy import Uuid
from sqlalchemy.orm import Mapped
from sqlalchemy import Float, Uuid
from sqlalchemy.orm import Mapped, mapped_column

logger = logging.getLogger(__name__)
logger.debug("boundary_condition.py")
Expand Down Expand Up @@ -46,87 +47,175 @@ class BoundaryCondition(Base):
as the image simulation tool for previewing the expected image and
sources in the field.
.. important::
Developers and observatory managers should be aware that the `bc_func`
and `lqs_func` attributes of this class are not mapped, so committing
an instance of this class to the database will not store the functions
in the database. This is by design, as storing functions in the
database is *not recommended* due to the potential for security risks
associated with the execution of arbitrary serialized code. Instead,
the `~pyscope.scheduling.BoundaryCondition` class is designed to be
used as a base class for defining custom boundary conditions that can
be used in the scheduling optimization process. The `bc_func` and
`lqs_func` functions should be defined in the subclass as static
methods or regular methods.
Parameters
----------
bc_func : `function`, default : `None`
The function to evaluate the condition. This function should take a `~astropy.coordinates.SkyCoord`,
a `~astropy.time.Time`, and a `~astropy.coordinates.EarthLocation` as arguments and return a value
that will be used by the `lqs_func` to evaluate the condition. If `None`, the `lqs_func` function will
be used directly.
lqs_func : `function`, default : `None`
The function to convert the output of the `bc_func` into a linear quality score between `0` and `1`.
This function should take the output of the `bc_func` and return a value between `0` and `1` that represents
the quality of the condition. If `None`, the output of the `bc_func` will be used directly.
bc_func : `Callable`, default : `None`
The function to evaluate the condition. This function should take a
`~astropy.coordinates.SkyCoord`, a `~astropy.time.Time`, and a
`~astropy.coordinates.EarthLocation` as arguments and return a value
that will be used by the `lqs_func` to evaluate the condition. If
`None`, the `lqs_func` function will be used directly.
lqs_func : `Callable`, default : `None`
The function to convert the output of the `bc_func` into a linear
quality score between `0` and `1`. This function should take the
output of the `bc_func` and return a value between `0` and `1` that
represents the quality of the condition. If `None`, the output of the
`bc_func` will be used directly. In this case, the
`~pyscope.scheduling.Scheduler` will typically try to normalize over
all evaluated values from this boundary condition to a value between
`0` and `1`, i.e., a `~pyscope.scheduling.LQSMinMax` with `min` and
`max` values set to the minimum and maximum values of the `bc_func`
output.
weight : `float`, default : 1
The weight of this condition relative to other conditions. This value is used by the `~pyscope.telrun.Optimizer`
inside the `~pyscope.telrun.Scheduler` to compute the overall quality of a `~pyscope.telrun.Field` or
or `~pyscope.telrun.ScheduleBlock` based on the conditions that are evaluated. The weight should be a positive value,
and the relative weight of each condition will be used to scale the output of the `lqs_func` function when computing the
overall score. A weight of `0` will effectively disable the condition from being used in the optimization, and a
weight of `1` is the default value. The weight can be set to any positive value to increase the relative importance of
the condition. The composite linear quality score is typically computed as the geometric mean of the individual condition
scores, so the weights are used to increase the power index of the geometric mean for each condition. Expressed
mathematically:
The weight of this condition relative to other conditions. This value
is used by the `~pyscope.scheduling.Optimizer` inside the
`~pyscope.scheduling.Scheduler` to compute the overall quality of a
`~pyscope.scheduling.ScheduleBlock` based on the conditions that are
evaluated. The weight should be a positive value, and the relative
weight of each condition will be used to scale the output of the
`lqs_func` function when computing the overall score. A weight of
`0` will effectively disable the condition from being used in the
optimization, and a weight of `1` is the default value. The weight
can be set to any positive value to increase the relative importance
of the condition. The composite linear quality score is typically
computed as the geometric mean of the individual condition scores, so
the weights are used to increase the power index of the geometric mean
for each condition. Expressed mathematically:
.. math::
Q = \\left( \\prod_{i=1}^{N} q_i^{w_i} \\right)^{1 / \\sum_{i=1}^{N} w_i}
where :math:`Q` is the composite quality score, :math:`q_i` is the quality score of the :math:`i`-th condition,
and :math:`w_i` is the weight of the :math:`i`-th condition. The sum of the weights is used to normalize the
composite quality score to a value between `0` and `1`. The default weight of `1` is used to give equal weight to
all conditions, but users can adjust the weights to prioritize certain conditions over others. Since the weights
are used as exponents in the geometric mean, `float` weights are possible.
where :math:`Q` is the composite quality score, :math:`q_i` is the
quality score of the :math:`i`-th condition, and :math:`w_i` is the
weight of the :math:`i`-th condition. The sum of the weights is used
to normalize the composite quality score to a value between
`0` and `1`. The default weight of `1` is used to give equal weight to
all conditions, but users can adjust the weights to prioritize certain
conditions over others. Since the weights are used as exponents in the
geometric mean, `float` weights are possible.
See Also
--------
pyscope.telrun.CoordinateCondition
pyscope.telrun.HourAngleCondition
pyscope.telrun.AirmassCondition
pyscope.telrun.MoonCondition
pyscope.telrun.SunCondition
pyscope.telrun.TimeCondition
pyscope.telrun.SNRCondition
pyscope.scheduling.CoordinateCondition
pyscope.scheduling.HourAngleCondition
pyscope.scheduling.AirmassCondition
pyscope.scheduling.MoonCondition
pyscope.scheduling.SunCondition
pyscope.scheduling.TimeCondition
pyscope.scheduling.SNRCondition
"""

bc_func: InitVar[Callable] = None
"""
The function to evaluate the condition. This function should take a
`~astropy.coordinates.SkyCoord`, a `~astropy.time.Time`, and a
`~astropy.coordinates.EarthLocation` as arguments and return a value
that will be used by the `lqs_func` to evaluate the condition. If
`None`, the `lqs_func` function will be used directly.
.. important::
This is not a mapped attribute and therefore will not be stored in
the database. The `bc_func` is typically defined as a static method
or regular method in a subclass of
`~pyscope.scheduling.BoundaryCondition`.
"""

lqs_func: InitVar[Callable] = None
"""
The function to convert the output of the `bc_func` into a linear
quality score between `0` and `1`. This function should take the
output of the `bc_func` and return a value between `0` and `1` that
represents the quality of the condition. If `None`, the output of the
`bc_func` will be used directly. In this case, the
`~pyscope.scheduling.Scheduler` will typically try to normalize over all
evaluated values from this boundary condition to a value between `0` and
`1`, i.e., a `~pyscope.scheduling.LQSMinMax` with `min` and `max` values
set to the minimum and maximum values of the `bc_func` output.
.. important::
This is not a mapped attribute and therefore will not be stored in
the database. The `lqs_func` is typically a subclass of the
`~pyscope.scheduling.LQS` class with a `__call__` method that
accepts the output of the `bc_func` as an argument and returns a
value between `0` and `1`.
"""

# weight: Mapped[float | None] = mapped_column(Float, default=1)
weight: Mapped[float | None] = mapped_column(Float, default=1)
"""
The weight of this condition relative to other conditions. This value
is used by the `~pyscope.scheduling.Optimizer` inside the
`~pyscope.scheduling.Scheduler` to compute the overall quality of a
`~pyscope.scheduling.ScheduleBlock` based on the conditions that are
evaluated. The weight should be a positive value, and the relative
weight of each condition will be used to scale the output of the
`lqs_func` function when computing the overall score. A weight of
`0` will effectively disable the condition from being used in the
optimization, and a weight of `1` is the default value. The weight
can be set to any positive value to increase the relative importance
of the condition. The composite linear quality score is typically
computed as the geometric mean of the individual condition scores, so
the weights are used to increase the power index of the geometric mean
for each condition. Expressed mathematically:
.. math::
Q = \\left( \\prod_{i=1}^{N} q_i^{w_i} \\right)^{1 / \\sum_{i=1}^{N} w_i}
where :math:`Q` is the composite quality score, :math:`q_i` is the
quality score of the :math:`i`-th condition, and :math:`w_i` is the
weight of the :math:`i`-th condition. The sum of the weights is used
to normalize the composite quality score to a value between
`0` and `1`. The default weight of `1` is used to give equal weight to
all conditions, but users can adjust the weights to prioritize certain
conditions over others. Since the weights are used as exponents in the
geometric mean, `float` weights are possible.
"""

def __post_init__(self):
def __post_init__(
self, bc_func: Callable | None, lqs_func: Callable | None
) -> None:
self._bc_func = bc_func
self._lqs_func = lqs_func
logger.debug("BoundaryCondition = %s" % self.__repr__)

def __call__(self, target, time, location):
def __call__(
self, target: SkyCoord, time: Time, location: EarthLocation
) -> float:
"""
Evaluate the `~pyscope.scheduling.BoundaryCondition` for a given target, time, and location.
This is a shortcut for calling the `func` and `lqs_func` functions directly. The `func` function
evaluates the condition for the target, time, and location and returns a value that is then passed
to the `lqs_func` function to convert the value into a linear quality score between `0` and `1`. The
`lqs_func` function is optional, and if not provided, the output of the `func` function will be used
Evaluate the `~pyscope.scheduling.BoundaryCondition` for a given
target, time, and location.
This is a shortcut for calling the `bc_func` and `lqs_func` functions
directly. The `bc_func` function evaluates the condition for the
target, time, and location and returns a value that is then passed to
the `lqs_func` function to convert the value into a linear quality
score between `0` and `1`. The `lqs_func` function is optional, and if
not provided, the output of the `bc_func` function will be used
directly as the quality score. The code is essentially equivalent to:
.. code-block:: python
if func is not None and lqs_func is None:
value = func(target, time, location)
if bc_func is not None and lqs_func is None:
value = bc_func(target, time, location)
elif lqs_func is not None:
value = lqs_func(func(target, time, location))
else:
raise ValueError("Either func or lqs_func must be provided.")
raise ValueError("Either bc_func or lqs_func must be provided.")
Parameters
----------
Expand All @@ -142,15 +231,16 @@ def __call__(self, target, time, location):
Returns
-------
`float`
A `float` value between `0` and `1` that represents the linear quality score of the condition.
A `float` value between `0` and `1` that represents the linear
quality score of the condition.
"""
logger.debug(
"BoundaryCondition().__call__(target=%s, time=%s, location=%s)"
% (target, time, location)
)

if self._func is not None and self._lqs_func is None:
if self._bc_func is not None and self._lqs_func is None:
value = self.calculate(target, time, location)
elif self._lqs_func is not None:
value = self.score(self.calculate(target, time, location))
Expand All @@ -159,23 +249,24 @@ def __call__(self, target, time, location):

return value

def calculate(self, target, time, location):
return self._func(target, time, location)
def calculate(
self, target: SkyCoord, time: Time, location: EarthLocation
) -> Quantity | float:
"""
Shorthand for calling the `bc_func` function directly.
"""
return self._bc_func(target, time, location)

def score(self, value):
def score(self, value: Quantity | float) -> float:
"""
Shorthand for calling the `lqs_func` function directly.
"""
return self._lqs_func(value)

def plot(self, target, time, location):
pass


BoundaryCondition.__annotations__["uuid"] = Mapped[Uuid]
BoundaryCondition.uuid.__doc__ = """
The universally unique identifier (UUID) for the database entry
corresponding to the object. This UUID is generated automatically with
`uuid.uuid4` when the object is created and is used to uniquely
identify the object in the database. The UUID is a primary key for
the table and is required to be unique for each entry. The UUID is
not intended to be used as a human-readable identifier and should
not be relied upon for that purpose.
"""
def plot(
self, target: SkyCoord, time: Time, location: EarthLocation
) -> Tuple[Figure, Axes]:
"""
To be implemented
"""
raise NotImplementedError

0 comments on commit 57de984

Please sign in to comment.