io

trap.io.is_fits(file: Path)[source]

Check if the supplied file exists and is a .fits image.

Parameters:

file (Path) – The path to the file of which we want to know if it is a valid fits file

Returns:

Whether the file is a fits file or not

Return type:

bool

trap.io.find_fits(path: str | Path) → List[Path][source]

Find the .fits files in the supplied directories. Only .fits images are supported. Both absolute and relative paths are allowed. Supported input style:

  • File (/data/im0.fits)

  • Directory (/data/im0.fits)

  • Glob (/data//im_)

In the case of a directory all fits files in the given directory are used. Any non-fits files are ignored. Same with the glob pattern.

Parameters:

fits_path (list) – A list of directories, glob patterns or file locations.

Returns:

fits_path – A list of file paths for each .fits file found

Return type:

numpy.ndarray

trap.io.read_fits_headers(fits_paths: List[Path]) → List[Header][source]

Read the fits header for the given fits paths.

Parameters:

fits_path (list) – A list of .fits files.

Returns:

The headers in the order of the given fits paths

Return type:

fits.header.Header

trap.io.max_image_size(headers: List[Header]) → int[source]

Determine the size of the largest image given a list of fits headers.

Parameters:

headers (list) – A list of .fits headers

Returns:

The size of the largest image in bytes

Return type:

Int

trap.io.headers_by_time(headers: List[Header]) → ndarray[source]

Order a list of headers of fits images by their acquisition time.

Parameters:

headers (list) – A list of .fits headers.

Returns:

An array of integers representing the order of the header datetimes

Return type:

numpy.ndarray

trap.io.init_db(db_backend: Literal['sqlite', 'postgres'], db_name: str, db_user: str | None = None, db_password: str | None = None, db_host: str | None = None, db_port: str | None = None, db_overwrite: bool = False) → Engine[source]

Create a new database. An error is raised if the database already exists, except if db_overwrite is provided. In that case the original database will be removed before the new database is created.

Parameters:

  • db_backend (Literal["sqlite", "postgres"] (default: "sqlite")) – The database implementation to use.

  • db_name (str) – If db_backend is “sqlite” this is the path to the database. If db_backend is “postgres” this is the name of database.

  • db_user (str) – The username with which to log in to the database. Only used if db_backend is “postgres”.

  • db_password (str) – The password to go along with db_user. Only used if db_backend is “postgres”.

  • db_host (str) – The name of the host where the database is located. Only used if db_backend is “postgres”.

  • db_port (str) – The port at which to connect to the host. Goes along wht db_host. Only used if db_backend is “postgres”.

  • db_overwrite (bool (default: False)) – Allow overwriting of an existing database. If db_overwrite is False, an error is raised if the database db_name already exists. If db_overwrite is True, the database db_name will be removed if it already exists.

Return type:

A sqlalchemy database engine.

trap.io.open_db(db_backend: Literal['sqlite', 'postgres'], db_name: str, db_user: str | None = None, db_password: str | None = None, db_host: str | None = None, db_port: str | None = None) → Engine[source]

Open a handle to a database for reading or appending to.

Parameters:

  • db_backend (Literal["sqlite", "postgres"] (default: "sqlite")) – The database implementation to use.

  • db_name (str) – If db_backend is “sqlite” this is the path to the database. If db_backend is “postgres” this is the name of database.

  • db_user (str) – The username with which to log in to the database. Only used if db_backend is “postgres”.

  • db_password (str) – The password to go along with db_user. Only used if db_backend is “postgres”.

  • db_host (str) – The name of the host where the database is located. Only used if db_backend is “postgres”.

  • db_port (str) – The port at which to connect to the host. Goes along wht db_host. Only used if db_backend is “postgres”.

Return type:

A sqlalchemy database engine.

trap.io.export_to_database(db_kwargs, im_id, im_meta, varmetric, extracted_sources)[source]

Write the data for time step im_id to the database.

Parameters:

  • db_kwargs (dict) – A dictionary containing the arguments for trap.io.open_db()

  • im_id (int) – The index corresponding to the image

  • im_meta (dict) – A dictionary containing the metadata corresponding to the image that is to be written to the database. # TODO: point to table in database description (yet to be documented)

  • extracted_sources (dict) – A dictionary containing the information of each source that was fitted in the image (corresponding to the supplied im_id and im_meta). These sources will be written to the database. # TODO: point to table in database description (yet to be documented)

Returns:

db_kwargs – Return the same database arguments, unaltered. This is exported such that this can be fed into the next call to export_database to enable the creation of a sequential Dask chain, which would prevent simultaneous execution of multiple export_to_database calls at once. # TODO: investigate if we can have parallel writes to the same table (I believe postgres supports this but sqlite less so)

Return type:

dict

trap.io.source_list_from_db(db_backend: Literal['sqlite', 'postgres'], db_name: str, db_user: str | None = None, db_password: str | None = None, db_host: str | None = None, db_port: str | None = None) → DataFrame[source]

Restore a source list from a TraP database as required for association.

Parameters:

  • db_backend (Literal["sqlite", "postgres"]) – The database implementation to use.

  • db_name (str) – If db_backend is “sqlite”, this is the filesystem path to the SQLite database file. If db_backend is “postgres”, this is the name of the PostgreSQL database.

  • db_user (str, optional) – The username used to authenticate with a PostgreSQL database. Ignored for SQLite.

  • db_password (str, optional) – The password associated with db_user. Ignored for SQLite.

  • db_host (str, optional) – The hostname of the PostgreSQL database server. Ignored for SQLite.

  • db_port (str, optional) – The port on which the PostgreSQL server is listening. Ignored for SQLite.

Returns:

A DataFrame indexed by src_id containing the following columns:

  • rafloat

    The most recently measured right ascension for the source.

  • decfloat

    The most recently measured declination for the source.

  • uncertainty_nsfloat

    North–south positional uncertainty of the latest detection in degrees.

  • uncertainty_ewfloat

    East–west positional uncertainty of the latest detection in degrees.

  • nr_consecutive_force_fitsint

    Number of consecutive null-detections where the source was force-fit.

  • latest_extracted_source_idint

    The ID of the most recent measurement in extracted_sources corresponding to this source.

Return type:

pandas.DataFrame

Notes

The function expects two tables in the database: extracted_sources and images. The grouping and update logic assume the schema used by the TraP pipeline as documented here: