Skip to content

Commit

Permalink
Fix stage plotting and tests.
Browse files Browse the repository at this point in the history
  • Loading branch information
@cgevans
cgevans committed Aug 28, 2024
1 parent 95e85b0 commit cc19301
Show file tree
Hide file tree
Showing 9 changed files with 172 additions and 122 deletions.
8 changes: 7 additions & 1 deletion src/qslib/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,13 @@
from .experiment import Experiment
from .machine import Machine, MachineStatus, RunStatus
from .plate_setup import PlateSetup, Sample
from .processors import NormRaw, NormToMaxPerWell, NormToMeanPerWell, SmoothEMWMean, SmoothWindowMean
from .processors import (
NormRaw,
NormToMaxPerWell,
NormToMeanPerWell,
SmoothEMWMean,
SmoothWindowMean,
)
from .protocol import CustomStep, Protocol, Stage, Step
from .scpi_commands import AccessLevel

Expand Down
6 changes: 2 additions & 4 deletions src/qslib/cli.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -337,12 +337,10 @@ def error(self, s: str):
click.secho(s, fg="red")


class NoNewAccess(BaseException):
...
class NoNewAccess(BaseException): ...


class NoAccess(BaseException):
...
class NoAccess(BaseException): ...


def set_default_access(m: Machine, p: OutP):
Expand Down
108 changes: 69 additions & 39 deletions src/qslib/experiment.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -487,6 +487,8 @@ def info_html(self) -> str:
summary = self.info(plate="table")

import matplotlib.pyplot as plt
from matplotlib.axes import Axes
from matplotlib.lines import Line2D

fig, ax = plt.subplots(figsize=(21.0 / 2.54, 15.0 / 2.54))
self.protocol.plot_protocol(ax)
Expand Down Expand Up @@ -1152,9 +1154,9 @@ def _new_xml_files(self) -> None:
ET.SubElement(tp, "Name").text = "Custom"
ET.SubElement(tp, "Description").text = "Custom QSLib experiment"
ET.SubElement(tp, "ResultPersisterName").text = "scAnalysisResultPersister"
ET.SubElement(
tp, "ContributedResultPersisterName"
).text = "mcAnalysisResultPersister"
ET.SubElement(tp, "ContributedResultPersisterName").text = (
"mcAnalysisResultPersister"
)
ET.SubElement(e.getroot(), "ChemistryType").text = "Other"
ET.SubElement(e.getroot(), "TCProtocolMode").text = "Standard"
ET.SubElement(e.getroot(), "DNATemplateType").text = "WET_DNA"
Expand Down Expand Up @@ -1274,10 +1276,8 @@ def from_file(cls, file: str | os.PathLike[str] | IO[bytes]) -> Experiment:
except ValueError:
exp.spec_major_version = 1


z.extractall(exp._dir_base)


exp._update_from_files()

return exp
Expand Down Expand Up @@ -1644,9 +1644,11 @@ def _update_from_data(self) -> None:
json.load(f),
self.plate_type,
quant_files_path=(Path(self.root_dir) / "run/quant"),
start_time=self.activestarttime.timestamp()
if self.activestarttime
else None,
start_time=(
self.activestarttime.timestamp()
if self.activestarttime
else None
),
)
mdp = os.path.join(self._dir_base, "primary/multicomponent_data.json")
if os.path.isfile(mdp):
Expand Down Expand Up @@ -2034,11 +2036,11 @@ def plot_anneal_melt(

if ax is None:
ax = plt.figure(
**(
{"constrained_layout": True}
| (({} if figure_kw is None else figure_kw))
)
).add_subplot()
**(
{"constrained_layout": True}
| (({} if figure_kw is None else figure_kw))
)
).add_subplot()

data = self.welldata

Expand Down Expand Up @@ -2067,7 +2069,6 @@ def plot_anneal_melt(
else:
betweendat = None


for sample in samples:
wells = self.plate_setup.get_wells(sample)

Expand Down Expand Up @@ -2183,8 +2184,8 @@ def plot_over_time(
annotate_events: bool = True,
figure_kw: dict[str, Any] | None = None,
line_kw: dict[str, Any] | None = None,
start_time = None,
time_units: Literal["hours","seconds"] = "hours"
start_time=None,
time_units: Literal["hours", "seconds"] = "hours",
) -> "Sequence[Axes]":
"""
Plots fluorescence over time, optionally with temperatures over time.
Expand Down Expand Up @@ -2355,7 +2356,8 @@ def plot_over_time(

lines.append(
ax[0].plot(
filterdat.loc[stages, ("time", time_units)] - start_time_value,
filterdat.loc[stages, ("time", time_units)]
- start_time_value,
filterdat.loc[stages, (well, "fl")],
label=label,
marker=marker,
Expand Down Expand Up @@ -2412,10 +2414,12 @@ def plot_over_time(
t_sl = stage_lines
fl_sl = stage_lines

self._annotate_stages(ax[0], fl_sl, fl_asl, (xlims[1] - xlims[0]) * 3600.0)
self._annotate_stages(
ax[0], fl_sl, fl_asl, (xlims[1] - xlims[0]) * 3600.0, stages=stages
)

if annotate_events:
self._annotate_events(ax[0])
self._annotate_events(ax[0], stages=stages)

if temperatures == "axes":
if len(ax) < 2:
Expand Down Expand Up @@ -2509,6 +2513,7 @@ def plot_temperatures(
"""

import matplotlib.pyplot as plt
from matplotlib.axes import Axes

if not hasattr(self, "temperatures") or self.temperatures is None:
raise ValueError("Experiment has no temperature data.")
Expand Down Expand Up @@ -2538,10 +2543,12 @@ def plot_temperatures(
v = reltemps.loc[:, ("time", "hours")]
totseconds = 3600.0 * (v.iloc[-1] - v.iloc[0])

self._annotate_stages(ax, stage_lines, annotate_stage_lines, totseconds)
self._annotate_stages(
ax, stage_lines, annotate_stage_lines, totseconds, stages=False
)

if annotate_events:
self._annotate_events(ax)
self._annotate_events(ax, stages=False)

ax.set_ylabel("temperature (°C)")
ax.set_xlabel("time (hours)")
Expand All @@ -2552,7 +2559,12 @@ def plot_temperatures(
return ax

def _annotate_stages(
self, ax, stage_lines: bool, annotate_stage_lines: bool | float, totseconds
self,
ax: "Axes",
stage_lines: bool,
annotate_stage_lines: bool | float,
totseconds,
stages: slice | Sequence[int] | bool = slice(None),
):
if stage_lines:
if isinstance(annotate_stage_lines, float):
Expand All @@ -2562,34 +2574,40 @@ def _annotate_stages(
annotate_frac = 0.05

for _, s in self.stages.iterrows():
if s.stage == "PRERUN" or s.stage == "POSTRUN":
continue
if s.stage == "PRERUN" or s.stage == "POSTRUN" or s.stage == "POSTRun":
continue # FIXME: maybe we should include these
#

xlim = ax.get_xlim()
if not (xlim[0] <= s.start_seconds / 3600.0 <= xlim[1]):
if (stages != slice(None)) and (
not (xlim[0] <= s.start_seconds / 3600.0 <= xlim[1])
):
continue

xtrans = ax.get_xaxis_transform()
ax.axvline(
vline = ax.axvline(
s.start_seconds / 3600.0,
linestyle="dotted",
color="black",
linewidth=0.5,
)
durfrac = (s.end_seconds - s.start_seconds) / totseconds
if annotate_stage_lines and (durfrac > annotate_frac):
ax.text(
s.start_seconds / 3600.0 + 0.02,
0.9,
ax.annotate(
f"stage {s.stage}",
transform=xtrans,
xy=(1, 0.9),
xycoords=vline,
xytext=(5, 0),
textcoords="offset points",
rotation=90,
verticalalignment="top",
horizontalalignment="left",
)

def _annotate_events(self, ax):
def _annotate_events(self, ax, stages: slice | Sequence[int] | bool = slice(None)):
first_time = self.stages.iloc[0, :]["start_seconds"] / 3600.0
last_time = self.stages.iloc[-1, :]["end_seconds"] / 3600.0

opi = self.events.index[
(self.events["type"] == "Cover") & (self.events["message"] == "Raising")
]
Expand All @@ -2599,12 +2617,18 @@ def _annotate_events(self, ax):
cli = [cl[cl > x][0] if len(cl[cl > x]) > 0 else None for x in opi]
for x1, x2 in zip(opi, cli):
xlim = ax.get_xlim()
if not (xlim[0] <= self.events.loc[x1, "hours"] <= xlim[1] or
(x2 is not None and xlim[0] <= self.events.loc[x2, "hours"] <= xlim[1])):
continue
if not (
xlim[0] <= self.events.loc[x1, "hours"] <= xlim[1]
or (
x2 is not None
and xlim[0] <= self.events.loc[x2, "hours"] <= xlim[1]
)
):
if stages != slice(None):
continue
ax.axvspan(
self.events.loc[x1, "hours"],
self.events.loc[x2, "hours"] if x2 is not None else None,
self.events.loc[x2, "hours"] if x2 is not None else last_time,
alpha=0.5,
color="yellow",
)
Expand All @@ -2618,12 +2642,18 @@ def _annotate_events(self, ax):
cli = [cl[cl > x][0] if len(cl[cl > x]) > 0 else None for x in opi]
for x1, x2 in zip(opi, cli):
xlim = ax.get_xlim()
if not (xlim[0] <= self.events.loc[x1, "hours"] <= xlim[1] or
(x2 is not None and xlim[0] <= self.events.loc[x2, "hours"] <= xlim[1])):
continue
if not (
xlim[0] <= self.events.loc[x1, "hours"] <= xlim[1]
or (
x2 is not None
and xlim[0] <= self.events.loc[x2, "hours"] <= xlim[1]
)
):
if stages != slice(None):
continue
ax.axvspan(
self.events.loc[x1, "hours"],
self.events.loc[x2, "hours"] if x2 is not None else None,
self.events.loc[x2, "hours"] if x2 is not None else last_time,
alpha=0.5,
color="red",
)
Expand Down
Loading

0 comments on commit cc19301

Please sign in to comment.