orca.metadata.pathsmanagers =========================== .. py:module:: orca.metadata.pathsmanagers .. autoapi-nested-parse:: Path management utilities for OVRO-LWA data access. This module provides PathsManager classes that handle file path resolution for measurement sets, calibration tables, and output directories. Supports both offline batch processing and real-time pipeline modes. Classes: PathsManager: Abstract base class defining the path management interface. OfflinePathsManager: Path manager for offline transient processing. Attributes ---------- .. autoapisummary:: orca.metadata.pathsmanagers.SIDEREAL_DAY Classes ------- .. autoapisummary:: orca.metadata.pathsmanagers.PathsManager orca.metadata.pathsmanagers.OfflinePathsManager Module Contents --------------- .. py:data:: SIDEREAL_DAY .. py:class:: PathsManager Bases: :py:obj:`abc.ABC` Base PathsManager class. It contains functionality to manipulate datetime objects and find dada files. Maybe in the future it will evolve into an interface with abstract methods. .. attribute:: utc_times_mapping An ordered dictionary mapping datetime objects to dada files. .. py:method:: time_filter(start_time: datetime.datetime, end_time: datetime.datetime) -> PathsManager :abstractmethod: .. py:method:: get_bcal_path(bandpass_date: datetime.date, spw: Optional[str]) -> str :abstractmethod: .. py:class:: OfflinePathsManager(utc_times_txt_path: str, dadafile_dir: Optional[str] = None, working_dir: Optional[str] = None, gaintable_dir: str = None, flag_npy_paths: Optional[Union[str, Dict[datetime.datetime, str]]] = None) Bases: :py:obj:`PathsManager` PathsManager for offline transient processing. This could potentially work for processing the buffer too. A config file reader will probably parse a config file into this object. Assumes that the bandpass calibration table is named like bcal_dir/00.bcal' .. py:attribute:: utc_times_txt_path .. py:attribute:: utc_times_mapping .. py:property:: working_dir :type: str .. py:property:: gaintable_dir :type: str .. py:method:: get_bcal_path(bandpass_date: datetime.date, spw: str) -> str Return bandpass calibration path in /gaintable/path/2018-03-02/00.bcal. :param bandpass_date: Date of the bandpass solution requested. :param spw: Spectral window :returns: Bandpass calibration path. .. py:method:: get_gaintable_path(gaintable_date: datetime.date, spw: str, extension: str) -> str Get the path to a certain CASA gaintable. :param gaintable_date: date of the table requested :param spw: spw of the gaintable requested :param extension: extension of the gaintable (bcal etc) :returns: The path to the requested gaintable. .. py:method:: get_ms_path(timestamp: datetime.datetime, spw: str) -> str Generate measurement set paths that look like /path/to/working_dir/msfiles/2018-03-02/hh=02/2018-03-02T02:02:02/00_2018-03-02T02:02:02.ms. :param timestamp: Timestamp of the ms. :param spw: Spectral window of the ms. :returns: Path to the measurement set. .. py:method:: get_ms_parent_path(timestamp: datetime.datetime) -> str Generate measurement set parent paths that look like /path/to/working_dir/msfiles/2018-03-02/hh=02/2018-03-02T02:02:02/. :param timestamp: Timestamp of the ms. :returns: Path to the measurement set. .. py:method:: get_data_product_path(timestamp: datetime.datetime, product: str, file_suffix: str, file_prefix: Optional[str] = None) -> str Generate path for generic data product. Looks like /path/to/working_dir//2018-03-02/hh=02/_2018-03-02T02:02:02. The first underscore is not there is file_prefix=None :param timestamp: Timestamp of observation. :param product: Name of the data product to be used for top-level directory :param file_suffix: Suffix to data file. For example for fits file it'd be '.fits'. You can also have something like '_v2.fits' :param file_prefix: Prefix of the file. Can be spectral window. If none specified then no prefix. Returns: Full path to the data product. .. py:attribute:: dpp .. py:method:: get_flag_npy_path(timestamp: datetime.datetime) -> str Return the a priori npy for the flags column for a given time. :param timestamp: :returns: If only one flag_npy was supplied, the flag_npy; if a Dict[datetime, str] is supplied, the closest one in time to the supplied timestamp. .. py:method:: time_filter(start_time: datetime.datetime, end_time: datetime.datetime) -> OfflinePathsManager Returns another PathsManager object with only utc_times between start_time (inclusive) and end_time (exclusive). :param start_time: :param end_time: :returns: New PathsManager object with time filtered. :rtype: new_paths_manager .. py:method:: chunks_by_integration(chunk_size: int) -> List[List[datetime.datetime]] Chunk the datetime array by number of integrations such that each chunk contains data spanning equal or less than the chunk size. Note that the last chunk may be smaller, if the total number of integrations is not divisible by the chunk size. :param chunk_size: number of integrations per chunk Returns: A list whose elements are the ordered chunks, which are each a list of ordered timestamps. .. py:method:: chunks_by_time(chunk_time: datetime.timedelta) -> List[List[datetime.datetime]] :abstractmethod: Chunk the datetime array by time such that each chunk contains data spanning equal or less than chunk_time amount of time. All of the data will be chunked. Note that 1) The last chunk may be smaller, if the span of the data is not divisible by the chunk time 2) Chunking is based on time and not by the number of integrations. Therefore, some chunks might have more or fewer integrations than some other chunks, if the chunk time is not divisible by the integration time. :param chunk_time: a timedelta object for the chunk time. Returns: A list whose elements are the ordered chunks, which are each a list of ordered timestamps.