API reference

Processing units

class ampel.abstract.AbsLightCurveT2Unit.AbsLightCurveT2Unit(**kwargs)

Base class for T2s that operate on light curves.

static build(compound, datapoints)

Create the parametrized type using compound and datapoints. For example, AbsCustomStateT2Unit[LightCurve] would return a LightCurve instance.

Return type:

LightCurve

require: ClassVar[None | tuple[str, ...]] = {}

Resources requirements as class variable (passed on to and merged with subclasses).

class ampel.abstract.AbsTiedLightCurveT2Unit.AbsTiedLightCurveT2Unit(*args, **kwargs)

Bases: AbsTiedCustomStateT2Unit[LightCurve, T]

abstract process(light_curve, t2_views)

Returned object should contain computed science results to be saved into the DB. Notes: dict must have only string keys and values must be bson encodable

Return type:

Union[None, str, int, float, bool, bytes, list[Any], dict[str, Any], UnitResult]

static get_link(link_override, light_curve)

Method used by T2Processor. :type link_override: dict[Literal['pps', 'uls'], Literal['first', 'middle', 'last']] :param link_override: value associated with key ‘link_override’ of T2Dependency <ampel.struct.T2Dependency.T2Dependency> :rtype: UnionType[None, int, bytes] :returns: the value of ‘link’ (tied t2 documents) to be matched

static build(compound, datapoints)

Create the parametrized type using compound and datapoints. For example, AbsTiedCustomStateT2Unit[LightCurve] would return a LightCurve instance.

Return type:

LightCurve

require: ClassVar[None | tuple[str, ...]] = {}

Resources requirements as class variable (passed on to and merged with subclasses).

class ampel.abstract.AbsPhotoT3Unit.AbsPhotoT3Unit(**kwargs)

Parametrized abstract class for T3 units receiving TransientView instances (and potentially LightCurve instances as well)

require: ClassVar[None | tuple[str, ...]] = {}

Resources requirements as class variable (passed on to and merged with subclasses).

Data classes

class ampel.view.LightCurve.LightCurve(compound_id, stock_id, photopoints=None, upperlimits=None)

Contains a collection of DataPoint (photo points and upper limits), and a few convenience methods to return values from internal collections.

get_values(key, filters=None, of_upper_limits=False)

Parameters:

• filters (Union[None, Sequence[dict[str, Any]], dict[str, Any]]) – filter criteria for data points, e.g. {'attribute': 'magpsf', 'operator': '<', 'value': 18}

• of_upper_limits (bool) – return upper limits instead of photo points

Return type:

Optional[list[Any]]

usage example:

lightcurve.get_values('jd')

See also

get_tuples(), get_ntuples()

get_tuples(key1, key2, filters=None, of_upper_limits=False)

Parameters:

• filters (Union[None, Sequence[dict[str, Any]], dict[str, Any]]) – filter criteria for datapoints

• of_upper_limits (bool) – return upper limits instead of photo points

Return type:

Optional[list[tuple[Any, Any]]]

usage example:

lightcurve.get_tuples('jd', 'magpsf', {'attribute': 'magpsf', 'operator': '<', 'value': 18})

See also

get_values(), get_ntuples()

get_ntuples(params, filters=None, of_upper_limits=False)

Parameters:

• params (Sequence[str]) – list of keys

• filters (Union[None, Sequence[dict[str, Any]], dict[str, Any]]) – filter criteria for datapoints

• of_upper_limits (bool) – return upper limits instead of photo points

Return type:

Optional[list[tuple]]

usage example:

lightcurve.get_ntuples(["fid", "jd", "magpsf"], {'attribute': 'magpsf', 'operator': '<', 'value': 18})

See also

get_values(), get_tuples()

get_photopoints(filters=None)

Get (filtered) photo points

Return type:

Optional[Sequence[DataPoint]]

get_upperlimits(filters=None)

Get (filtered) upper limits

Return type:

Optional[Sequence[DataPoint]]

get_pos(ret='brightest', filters=None)

Calculate the position of the underlying object.

Parameters:

• ret (str) –

  • raw: returns ((ra, dec), (ra, dec), …)

  • mean: returns (<ra>, <dec>)

  • brightest: returns (ra, dec)

  • latest: returns (ra, dec)

• filters (Union[None, Sequence[dict[str, Any]], dict[str, Any]]) – filters to apply to photo points

Return type:

UnionType[None, tuple[Any, Any], Sequence[tuple[Any, Any]]]

Get the position of the brightest PhotoPoint in the ZTF G band:

instance.get_pos(
        "brightest",
        {'attribute': 'alTags', 'operator': 'in', 'value': 'ZTF_G'}
)

Get the position of the latest photopoint with a magnitude brighter than 18 (or an empty array if no photopoint matches this criteria):

instance.get_pos(
        "lastest",
        {'attribute': 'magpsf', 'operator': '<', 'value': 18}
)

class ampel.view.TransientView.TransientView(id, stock=None, origin=None, t0=None, t1=None, t2=None, logs=None, extra=None)

Bases: SnapView

lightcurve: Optional[Sequence[LightCurve]]

get_photopoints()

Return type:

Optional[Sequence[DataPoint]]

get_upperlimits()

Return type:

Optional[Sequence[DataPoint]]

get_lightcurves()

Return type:

Optional[Sequence[LightCurve]]

content_summary()

Return type:

str

extra: Optional[dict[str, Any]]

get_journal_entries(tier=None, process_name=None, filter_func=None)

Get a subset of journal entries.

Parameters:

• tier (Optional[Literal[0, 1, 2, 3]]) – return only journal entries associated with the given tier

• process_name (Optional[str]) – return only journal entries associated with a given process name

• latest – return only the latest entry in the journal (the latest in time)

Return type:

Iterator[JournalRecord]

Returns:

journal entries corresponding to a given tier and/or job, sorted by timestamp.

get_latest_t2_body(unit, link=None, code=None)

Get latest t2 body element from a given unit.

Parameters:

• unit_id – target unit id

• link_id – restrict to a specific link

Return type:

Union[None, str, int, float, bool, bytes, list[Any], dict[str, Any]]

get_raw_t2_body(unit, link=None, code=None)

Parameters:

• link (Union[None, int, bytes, str]) – restrict to a specific link

• code (Optional[int]) – restrict to a specific code

Return type:

Optional[Sequence[Union[None, str, int, float, bool, bytes, list[Any], dict[str, Any]]]]

get_t2_body(unit, ret_type=<class 'dict'>, *, data_slice=-1, link=None, code=None, raise_exc=False)

Get latest t2 body element from a given unit. :type ret_type: type[TypeVar(T)] :param ret_type: expected body element type. If isinstance check is not fullfied None will be returned (unless multiple unit are to be matched and another unit fullfill the criteria) or an exception will be raised if raise_exc is True :type link: Union[None, int, bytes, str] :param link: restrict to a specific link

Return type:

Optional[TypeVar(T)]

get_t2_ntuple(unit, key, rtype, *, no_none=False, require_all_keys=True, code=None)

Examples: get_t2_ntuple(“T2NedTap”, (“ra”, “dec”, “z”, “zunc”), float) get_t2_ntuple((“T2NedSNCosmo”, “T2SNCosmo”), (“fit_result”, “covariance”), dict)

see T2DocView.get_ntuple(…) for more info

Return type:

UnionType[None, tuple[TypeVar(T), ...], tuple[Optional[TypeVar(T)], ...]]

get_t2_value(unit, key, rtype, *, code=None)

Examples: get_t2_value((“T2NedSNCosmo”, “T2SNCosmo”), “fit_result”, dict)

see T2DocView.get_value(…) for more info

Return type:

Optional[TypeVar(T)]

get_t2_views(unit=None, link=None, code=None)

Get a subset of T2 documents.

Parameters:

• unit (Union[None, str, list[str], tuple[str, ...]]) – limits the returned science record(s) to the one with the provided t2 unit id

• link (Union[None, int, bytes, str]) – whether to return the latest science record(s) or not (default: False)

Return type:

Iterator[T2DocView]

get_time_created(output='raw')

Return type:

UnionType[None, float, datetime, str]

get_time_updated(output='raw')

Return type:

UnionType[None, float, datetime, str]

id: Union[int, bytes, str]

logs: Optional[Sequence[LogDocument]]

classmethod of(ab, conf=None, freeze=True)

Return type:

SnapView

origin: Union[None, Sequence[int], int]

serialize()

Return type:

dict

stock: Optional[StockDocument]

t0: Optional[Sequence[DataPoint]]

t1: Optional[Sequence[T1Document]]

t2: Optional[Sequence[T2DocView]]

T1 machinery