common
fptools.preprocess.common
¶
Classes:
-
Pipeline–A pipeline of Processors, and is itself a Processor.
-
Processor–Abstract Processor.
-
ProcessorThatPlots–Abstract Processor which has the ability to plot.
Pipeline
¶
Bases: Processor
A pipeline of Processors, and is itself a Processor.
Methods:
Source code in fptools/preprocess/common.py
class Pipeline(Processor):
"""A pipeline of Processors, and is itself a Processor."""
def __init__(self, steps: Optional[list[Processor]] = None, plot: bool = True, plot_dir: Optional[str] = None):
"""Initialize this pipeline.
Args:
steps: list of Processors to run on a given Session
plot: whether to plot the results of each step
plot_dir: directory to save plots to, if None, will save to current working directory
"""
self.steps: list[Processor]
if steps is None:
self.steps = []
else:
self.steps = steps
self.plot: bool = plot
self.plot_dir: str = plot_dir or os.getcwd()
def __call__(self, session: Session) -> Session:
"""Run this pipeline on a Session.
Args:
session: the session to operate on
Returns:
Session with Processors applied
"""
try:
if self.plot:
# allocate a plot:
# - each step receives one row to plot on
fig, axs = plt.subplots(len(self.steps), 1, figsize=(24, 6 * len(self.steps)))
for i, step in enumerate(self.steps):
session = step(session)
if self.plot:
if hasattr(step, "plot"):
step.plot(session, axs[i])
else:
if hasattr(step, "__class__"):
step_name = step.__class__.__name__ # this handles the case where the step is a class
elif hasattr(step, "__name__"):
step_name = step.__name__ # this handles the case where the step is a function
elif hasattr(step, "func"):
step_name = step.func.__name__ # this handles the case where the step is a partial
else:
step_name = "Unknown" # this handles the case where we cannot figure out a reasonable name for the step
message = f'Step #{i+1} "{step_name}" has no plot method, skipping plotting for this step.'
axs[i].text(0.5, 0.5, message, ha="center", va="center", transform=axs[i].transAxes)
axs[i].axis("off")
return session
except:
raise
finally:
if self.plot:
fig.savefig(os.path.join(self.plot_dir, f"{session.name}.png"), dpi=300)
fig.savefig(os.path.join(self.plot_dir, f"{session.name}.pdf"))
plt.close(fig)
__call__(session)
¶
Run this pipeline on a Session.
Parameters:
-
session(Session) –the session to operate on
Returns:
-
Session–Session with Processors applied
Source code in fptools/preprocess/common.py
def __call__(self, session: Session) -> Session:
"""Run this pipeline on a Session.
Args:
session: the session to operate on
Returns:
Session with Processors applied
"""
try:
if self.plot:
# allocate a plot:
# - each step receives one row to plot on
fig, axs = plt.subplots(len(self.steps), 1, figsize=(24, 6 * len(self.steps)))
for i, step in enumerate(self.steps):
session = step(session)
if self.plot:
if hasattr(step, "plot"):
step.plot(session, axs[i])
else:
if hasattr(step, "__class__"):
step_name = step.__class__.__name__ # this handles the case where the step is a class
elif hasattr(step, "__name__"):
step_name = step.__name__ # this handles the case where the step is a function
elif hasattr(step, "func"):
step_name = step.func.__name__ # this handles the case where the step is a partial
else:
step_name = "Unknown" # this handles the case where we cannot figure out a reasonable name for the step
message = f'Step #{i+1} "{step_name}" has no plot method, skipping plotting for this step.'
axs[i].text(0.5, 0.5, message, ha="center", va="center", transform=axs[i].transAxes)
axs[i].axis("off")
return session
except:
raise
finally:
if self.plot:
fig.savefig(os.path.join(self.plot_dir, f"{session.name}.png"), dpi=300)
fig.savefig(os.path.join(self.plot_dir, f"{session.name}.pdf"))
plt.close(fig)
__init__(steps=None, plot=True, plot_dir=None)
¶
Initialize this pipeline.
Parameters:
-
steps(Optional[list[Processor]], default:None) –list of Processors to run on a given Session
-
plot(bool, default:True) –whether to plot the results of each step
-
plot_dir(Optional[str], default:None) –directory to save plots to, if None, will save to current working directory
Source code in fptools/preprocess/common.py
def __init__(self, steps: Optional[list[Processor]] = None, plot: bool = True, plot_dir: Optional[str] = None):
"""Initialize this pipeline.
Args:
steps: list of Processors to run on a given Session
plot: whether to plot the results of each step
plot_dir: directory to save plots to, if None, will save to current working directory
"""
self.steps: list[Processor]
if steps is None:
self.steps = []
else:
self.steps = steps
self.plot: bool = plot
self.plot_dir: str = plot_dir or os.getcwd()
Processor
¶
Bases: ABC
Abstract Processor.
Implementors should implement the __call__ method.
Methods:
-
__call__–Effect this processing step.
Source code in fptools/preprocess/common.py
class Processor(ABC):
"""Abstract Processor.
Implementors should implement the `__call__` method.
"""
@abstractmethod
def __call__(self, session: Session) -> Session:
"""Effect this processing step.
Args:
session: the session to operate upon
Returns:
Session with the processing step applied
"""
raise NotImplementedError()
ProcessorThatPlots
¶
Bases: Processor
Abstract Processor which has the ability to plot.
Implementors should implement the __call__ and plot methods.
Methods:
-
plot–Plot the effects of this processing step.
Source code in fptools/preprocess/common.py
class ProcessorThatPlots(Processor):
"""Abstract Processor which has the ability to plot.
Implementors should implement the `__call__` and `plot` methods.
"""
@abstractmethod
def plot(self, session: Session, ax: Axes) -> None:
"""Plot the effects of this processing step.
Args:
session: the session being operated upon
ax: matplotlib Axes for plotting onto
"""
raise NotImplementedError()