Coverage for src/instawell/core/exp_context.py: 93%
71 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-07 15:47 -0600
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-07 15:47 -0600
1from __future__ import annotations
3import logging
4from datetime import datetime, timezone
5from pathlib import Path
7from pydantic import BaseModel, Field, FilePath, computed_field, field_validator, model_validator
9logger = logging.getLogger(__name__)
12class ExperimentContext(BaseModel):
13 experiment_name: str
15 # Where all experiments live (default ./experiments)
16 experiments_root: Path = Field(default_factory=lambda: Path("experiments"))
18 # INSIDE experiment dir
19 raw_data_path: Path
20 layout_data_path: Path
22 # Tracks the original input files
23 raw_data_source: FilePath | None = None
24 layout_data_source: FilePath | None = None
26 # fields (must ensure these are in the correct order)
27 fields: tuple[str, ...] = ("concentration", "ligand", "protein", "buffer")
28 well_col_identifier: str = "Well"
29 empty_condition_placeholder: str = "0"
30 condition_separator: str = "_"
31 temperature_column: str = "Temperature"
32 non_protein_control_marker: str = "NPC"
34 # write the last edited datetime to the metadata
35 # Timestamp (persisted)
36 created_at: datetime = Field(
37 default_factory=lambda: datetime.now(timezone.utc),
38 description="UTC timestamp when this context was created.",
39 )
41 # logging options
42 log_to_file: bool = True
43 log_level: int = logging.INFO
45 @property
46 def experiment_dir(self) -> Path:
47 return (self.experiments_root / self.experiment_name).resolve()
49 @property
50 def log_path(self) -> Path:
51 return self.experiment_dir / "experiment.log"
53 @property
54 def metadata_path(self) -> Path:
55 return self.experiment_dir / "experiment.json"
57 @computed_field # included in model_dump / JSON
58 @property
59 def created_at_iso(self) -> str:
60 """ISO 8601 string version of created_at (nice for logs/UI)."""
61 return self.created_at.isoformat()
63 # validator that separator is a single character, and not a whitespace or comma or other char that might cause issues
64 @field_validator("condition_separator")
65 @classmethod
66 def validate_condition_separator(cls, v: str) -> str:
67 """
68 Ensure the separator is a single, safe, non-whitespace, non-confusing char.
69 """
70 if len(v) != 1:
71 raise ValueError("condition_separator must be exactly one character.")
72 if v.isspace():
73 raise ValueError("condition_separator cannot be whitespace.")
74 if v in {",", ";"}:
75 raise ValueError(
76 f"condition_separator '{v}' is not allowed "
77 "(commas/semicolons conflict with CSV/TSV parsing)."
78 )
79 # optional: discourage alphanumerics, which make parsing ambiguous
80 if v.isalnum():
81 raise ValueError(
82 f"condition_separator '{v}' should not be a letter or digit; "
83 "pick something like '|', ':', or '~'."
84 )
85 return v
87 @computed_field
88 @property
89 def empty_condition_mask(self) -> str:
90 """
91 Mask pattern used to represent an all-empty condition row.
93 Example:
94 fields = ("a", "b", "c", "d")
95 empty_condition_placeholder = "^"
96 condition_separator = "|"
98 -> "^|^|^|^"
99 """
100 n = len(self.fields)
101 if n == 0:
102 return ""
103 return self.condition_separator.join(self.empty_condition_placeholder for _ in range(n))
105 @field_validator("empty_condition_placeholder")
106 @classmethod
107 def validate_empty_placeholder(cls, v: str) -> str:
108 if len(v) != 1:
109 raise ValueError("empty_condition_placeholder must be exactly one character.")
110 if v.isspace():
111 raise ValueError("empty_condition_placeholder cannot be whitespace.")
112 if v in {",", ";"}:
113 raise ValueError(
114 f"empty_condition_placeholder '{v}' is not allowed "
115 "(commas/semicolons conflict with CSV/TSV parsing)."
116 )
117 # if v.isalnum():
118 # raise ValueError(
119 # f"empty_condition_placeholder '{v}' should not be a letter or digit; "
120 # "it should be a symbolic marker unlikely to appear in real values."
121 # )
122 return v
124 @model_validator(mode="after")
125 def validate_separator_and_placeholder(self) -> ExperimentContext:
126 # Must not collide
127 if self.empty_condition_placeholder == self.condition_separator:
128 raise ValueError(
129 "empty_condition_placeholder and condition_separator must be different "
130 f"(both are '{self.condition_separator}' right now)."
131 )
133 # If fields exist, mask must be non-empty (sanity check)
134 if self.fields and not self.empty_condition_mask:
135 raise ValueError(
136 "empty_condition_mask computed as empty; check fields/placeholder config."
137 )
139 return self