Source code for

import os
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
from matplotlib.figure import Figure
from typing import List, Dict, Union, ClassVar

from resistics.common.base import ResisticsBase
from resistics.common.print import arrayToString, listToString
from import (
from import ConfigData
from import SiteData

[docs]class ProjectData(ResisticsBase): """Project class for storing project attributes Attributes ---------- projectFile : str Path to project file timePath : str Path to time series data specPath : str Path to spectra data statPath : str Path to statistics transPath : str Path to transfer function data calPath : str Path to calibration data imagePath : str Path to project images refTime : datetime.datetime The project reference time projStart : datetime.datetime The first time sample in the data projEnd : datetime.datetime The last time sample in the data sites : List[str] List of sites in the project siteData : Dict[str, SiteData] Site data for each site calFiles : List A list of calibration files config : ConfigData Configuration data Methods ------- __init__() Initialise project with default values refresh() Refresh the project checkDirectories() Check the project directories to make sure they exist getCalInfo() Get calibration file information getSiteInfo() Get information about all the sites and store site data information getSites(fs=None) Get a list of all sites, or those with data at a sampling frequency of fs Hz getNumSites() Get the total number of sites getSampleFreqs() Get a list of all the unique sample frequencies in the project getSiteConcurrent(site) Get a list of sites that were recording concurrently to site getSiteData(site) Get site data for a single site checkSite(site) Check a site exists view() View the project timeline printList() Class status returned as list of strings """ def __init__( self, projectFile: str, refTime: datetime, calPath: str, timePath: str, specPath: str, statPath: str, maskPath: str, transFuncPath: str, imagePath: str, config=None, ): # project file and reference time self.projectFile: str = projectFile self.refTime: datetime = refTime # paths to various parts of the project self.calPath: str = calPath self.timePath: str = timePath self.specPath: str = specPath self.statPath: str = statPath self.maskPath: str = maskPath self.transFuncPath: str = transFuncPath self.imagePath: str = imagePath self.projStart: datetime = self.projEnd: datetime = # site information self.sites: List[str] = [] self.siteData: Dict[str, SiteData] = {} # cal files list self.calFiles: List = [] # configuration options self.config = config if self.config is None: self.config = ConfigData() # refresh self.refresh()
[docs] def refresh(self) -> None: """Refresh the project If new time, spectra or other data directories have been added to the project, these will not show up in the project information until the project has been refreshed (i.e. all the information read again from the disc) """ self.checkDirectories() self.getCalInfo() self.getSiteInfo() # update start and end dates for the project startDates = [] endDates = [] for site in self.sites: startDates.append(self.getSiteData(site).siteStart) endDates.append(self.getSiteData(site).siteEnd) if len(self.sites) > 0: self.projStart = min(startDates) self.projEnd = max(endDates)
[docs] def checkDirectories(self) -> None: """Check to see if project directories exist and if not, make them""" # data directories checkAndMakeDir(self.timePath) checkAndMakeDir(self.specPath) checkAndMakeDir(self.statPath) checkAndMakeDir(self.maskPath) checkAndMakeDir(self.transFuncPath) # calibration directories checkAndMakeDir(self.calPath) # image directory checkAndMakeDir(self.imagePath)
[docs] def getCalInfo(self): """Get a list of all the calibration files in the project""" self.calFiles = getFilesInDirectory(self.calPath)
[docs] def getSiteInfo(self) -> None: """Get information about sites""" self.sites = getDirsInDirectory(self.timePath) self.sites = sorted(self.sites) for site in self.sites: self.siteData[site] = SiteData( site, self.timePath, self.specPath, self.statPath, self.maskPath, self.transFuncPath, )
[docs] def getSites(self, sampleFreq: Union[None, float, int] = None): """Get list of sites Optionally filtered to those with measurements recorded at fs Hz Parameters ---------- sampleFreq : float, int, optional Sampling frequency Hz Returns ------- List[str] List of site names """ if sampleFreq is None: return self.sites # fs is supplied filteredSites = [] for site in self.sites: if sampleFreq in self.getSiteData(site).getSampleFreqs(): filteredSites.append(site) return filteredSites
[docs] def getNumSites(self) -> int: """Number of sites in the project Returns ------- int Number of sites """ return len(self.sites)
[docs] def getSampleFreqs(self) -> List[float]: """List of all the sampling frequencies found in the project Returns ------- List[float] List of sampling frequencies in Hz """ sampleFreq = set() for site in self.sites: sampleFreq.update(self.getSiteData(site).getSampleFreqs()) return sorted(list(sampleFreq))
[docs] def getSiteConcurrent(self, site: str) -> List[str]: """Find sites which were occupied at the same time as a chosen site This lists all sites that were occupied at the same time as the user supplied site. It can be useful in cases where multi-site processing will be performed Parameters ---------- site : str The name of the site Returns ------- List[str] List of sites which were recording concurrently """ siteStart = self.getSiteData(site).siteStart siteEnd = self.getSiteData(site).siteEnd concSites = [] for s in self.sites: if s == site: continue start = self.getSiteData(s).siteStart end = self.getSiteData(s).siteEnd # check that start time is before end time of site and that end time is after start time of site if (start < siteEnd) and (end > siteStart): concSites.append(s) return concSites
[docs] def getSiteData(self, site) -> Union[SiteData, bool]: """Get site data for site Parameters ---------- site : str Site name Returns ------- SiteData Site data for site or False if the site is not found """ if site not in self.sites: self.printWarning("Site {} does not exist in the project".format(site)) return False return self.siteData[site]
[docs] def checkSite(self, site: str) -> bool: """Check if site is part of the project Parameters ---------- site : str Name of the site """ if site in self.sites: return True else: return False
[docs] def createSite(self, site: str) -> bool: """Creates a site folder in the time data path Parameters ---------- site : str Name of the site """ sitePath = os.path.join(self.timePath, site) chk = checkDirExistence(sitePath) if chk: self.printWarning( "Site {} already exists in project time data folder".format(site) ) return False checkAndMakeDir(sitePath) return True
[docs] def view(self, **kwargs) -> Figure: """Plot a timeline of the project Parameters ---------- figsize : Tuple (width, height), optional The figure size plotfonts : Dict, optional Fonts to use for plotting show : bool, optional Boolean flag fow showing plot. Default is True. """ import matplotlib.dates as mdates from matplotlib.patches import Rectangle import numpy as np from resistics.common.plot import getPlotFonts figsize = kwargs["figsize"] if "figsize" in kwargs else (15, 8) plotFonts = kwargs["plotFonts"] if "plotFonts" in kwargs else getPlotFonts() show = kwargs["show"] if "show" in kwargs else True fig = plt.figure(figsize=figsize) ax = plt.subplot(1, 1, 1) # get xlimits xStart = self.refTime - timedelta(days=2) xEnd = self.projEnd + timedelta(days=2) numSites = len(self.sites) height = 0.4 for idx, site in enumerate(self.sites): idx = numSites - idx siteStart = mdates.date2num(self.siteData[site].siteStart) siteEnd = mdates.date2num(self.siteData[site].siteEnd) siteWidth = siteEnd - siteStart yOffset = 0.5 + height rect = Rectangle((siteStart, idx - yOffset), siteWidth, 2 * height) ax.add_patch(rect) # plot the reference line plt.plot( [self.refTime, self.refTime], [0, numSites], color="lightcoral", ls="-" ) plt.text( self.refTime, 0.02, " Reference Time", fontsize=plotFonts["axisTicks"] ) # x axis formatting weeks = mdates.WeekdayLocator() weeksFmt = mdates.DateFormatter("%Y-%m-%d") days = mdates.DayLocator() ax.xaxis.set_major_locator(weeks) ax.xaxis.set_major_formatter(weeksFmt) ax.xaxis.set_minor_locator(days) plt.xlim([xStart, xEnd]) ax.xaxis.grid(True, which="minor", ls="--", color="gainsboro") plt.xlabel("Date", fontsize=plotFonts["axisLabel"]) fig.autofmt_xdate(rotation=30) for label in ax.get_xticklabels(): label.set_fontsize(plotFonts["axisTicks"]) # y axis formatting plt.ylim([0, numSites]) yticks = np.arange(1, numSites + 1) ax.set_yticks(yticks) ax.set_yticklabels([]) ax.set_yticks(yticks - 0.5, minor=True) ax.set_yticklabels( self.sites[::-1], minor=True, fontsize=plotFonts["axisTicks"] ) ax.yaxis.grid(True, which="major", ls="--", color="gainsboro") plt.ylabel("Sites", fontsize=plotFonts["axisLabel"], labelpad=20) # title plt.title("Project Timeline", fontsize=plotFonts["title"]) fig.tight_layout() if show: return fig
[docs] def printList(self) -> List[str]: """Class information as a list of strings Returns ------- out : List[str] List of strings with information """ textLst: List[str] = [] textLst.append("Time data path = {}".format(self.timePath)) textLst.append("Spectra data path = {}".format(self.specPath)) textLst.append("Statistics data path = {}".format(self.statPath)) textLst.append("Mask data path = {}".format(self.maskPath)) textLst.append("TransFunc data path = {}".format(self.transFuncPath)) textLst.append("Calibration data path = {}".format(self.calPath)) textLst.append("Images data path = {}".format(self.imagePath)) textLst.append( "Reference time = {}".format(self.refTime.strftime("%Y-%m-%d %H:%M:%S")) ) textLst.append( "Project start time = {}".format( self.projStart.strftime("%Y-%m-%d %H:%M:%S.%f") ) ) textLst.append( "Project stop time = {}".format( self.projEnd.strftime("%Y-%m-%d %H:%M:%S.%f") ) ) textLst.append("Project found {} sites:".format(self.getNumSites())) for site in self.sites: textLst.append( "{}\t\tstart: {}\tend: {}".format( site, self.getSiteData(site).siteStart, self.getSiteData(site).siteEnd, ) ) textLst.append( "Sampling frequencies found in project (Hz): {}".format( listToString(self.getSampleFreqs()) ) ) return textLst