Skip to content

Quality control

rationai.resources.qc.QualityControl

Bases: APIResource

Source code in rationai/resources/qc.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class QualityControl(APIResource):
    @retry(
        retry=retry_if_exception(_is_500_error),
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
    )
    def check_slide(
        self,
        wsi_path: PathLike[str] | str,
        output_path: PathLike[str] | str,
        config: SlideCheckConfig | None = None,
        timeout: TimeoutTypes | UseClientDefault = 3600,
    ) -> str:
        """Check quality of a single slide with automatic retry on failure.

        Args:
            wsi_path: Path to the whole slide image.
            output_path: Directory to save output masks.
            config: Optional configuration for the slide quality check.
            timeout: Optional timeout for the request.

        Returns:
            An xOpat link containing the processed WSI, enabling a quick visual
            inspection of the generated masks. The masks are displayed in the same
            format as in the QC report.
        """
        response = self._put(
            "",
            json={
                "wsi_path": str(wsi_path),
                "output_path": str(output_path),
                **asdict(config or SlideCheckConfig()),
            },
            timeout=timeout,
        )
        response.raise_for_status()
        return response.text

    def generate_report(
        self,
        backgrounds: Iterable[PathLike[str] | str],
        mask_dir: PathLike[str] | str,
        save_location: PathLike[str] | str,
        compute_metrics: bool = True,
        timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    ) -> None:
        """Generate a QC report from processed slides.

        Args:
            backgrounds: List of paths to the slide images.
            mask_dir: Directory containing the generated masks.
            save_location: Path where the report HTML will be saved.
            compute_metrics: Whether the generated report should contain aggregated
                metrics about the slide's coverage by the different classes of
                artifacts.
            timeout: Optional timeout for the request.
        """
        response = self._put(
            "report",
            json={
                "backgrounds": [str(bg) for bg in backgrounds],
                "mask_dir": str(mask_dir),
                "save_location": str(save_location),
                "compute_metrics": compute_metrics,
            },
            timeout=timeout,
        )
        response.raise_for_status()

check_slide(wsi_path, output_path, config=None, timeout=3600)

Check quality of a single slide with automatic retry on failure.

Parameters:

Name Type Description Default
wsi_path PathLike[str] | str

Path to the whole slide image.

 required
output_path PathLike[str] | str

Directory to save output masks.

 required
config SlideCheckConfig | None

Optional configuration for the slide quality check.

 None
timeout TimeoutTypes | UseClientDefault

Optional timeout for the request.

 3600

Returns:

Type Description
str

An xOpat link containing the processed WSI, enabling a quick visual
str

inspection of the generated masks. The masks are displayed in the same
str

format as in the QC report.
Source code in rationai/resources/qc.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@retry(
    retry=retry_if_exception(_is_500_error),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
)
def check_slide(
    self,
    wsi_path: PathLike[str] | str,
    output_path: PathLike[str] | str,
    config: SlideCheckConfig | None = None,
    timeout: TimeoutTypes | UseClientDefault = 3600,
) -> str:
    """Check quality of a single slide with automatic retry on failure.

    Args:
        wsi_path: Path to the whole slide image.
        output_path: Directory to save output masks.
        config: Optional configuration for the slide quality check.
        timeout: Optional timeout for the request.

    Returns:
        An xOpat link containing the processed WSI, enabling a quick visual
        inspection of the generated masks. The masks are displayed in the same
        format as in the QC report.
    """
    response = self._put(
        "",
        json={
            "wsi_path": str(wsi_path),
            "output_path": str(output_path),
            **asdict(config or SlideCheckConfig()),
        },
        timeout=timeout,
    )
    response.raise_for_status()
    return response.text

generate_report(backgrounds, mask_dir, save_location, compute_metrics=True, timeout=USE_CLIENT_DEFAULT)

Generate a QC report from processed slides.

Parameters:

Name Type Description Default
backgrounds Iterable[PathLike[str] | str]

List of paths to the slide images.

 required
mask_dir PathLike[str] | str

Directory containing the generated masks.

 required
save_location PathLike[str] | str

Path where the report HTML will be saved.

 required
compute_metrics bool

Whether the generated report should contain aggregated metrics about the slide's coverage by the different classes of artifacts.

 True
timeout TimeoutTypes | UseClientDefault

Optional timeout for the request.

 USE_CLIENT_DEFAULT
Source code in rationai/resources/qc.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def generate_report(
    self,
    backgrounds: Iterable[PathLike[str] | str],
    mask_dir: PathLike[str] | str,
    save_location: PathLike[str] | str,
    compute_metrics: bool = True,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
) -> None:
    """Generate a QC report from processed slides.

    Args:
        backgrounds: List of paths to the slide images.
        mask_dir: Directory containing the generated masks.
        save_location: Path where the report HTML will be saved.
        compute_metrics: Whether the generated report should contain aggregated
            metrics about the slide's coverage by the different classes of
            artifacts.
        timeout: Optional timeout for the request.
    """
    response = self._put(
        "report",
        json={
            "backgrounds": [str(bg) for bg in backgrounds],
            "mask_dir": str(mask_dir),
            "save_location": str(save_location),
            "compute_metrics": compute_metrics,
        },
        timeout=timeout,
    )
    response.raise_for_status()

rationai.resources.qc.AsyncQualityControl

Bases: AsyncAPIResource

Source code in rationai/resources/qc.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
class AsyncQualityControl(AsyncAPIResource):
    @retry(
        retry=retry_if_exception(_is_500_error),
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
    )
    async def check_slide(
        self,
        wsi_path: PathLike[str] | str,
        output_path: PathLike[str] | str,
        config: SlideCheckConfig | None = None,
        timeout: TimeoutTypes | UseClientDefault = 3600,
    ) -> str:
        """Check quality of a single slide with automatic retry on failure.

        Args:
            wsi_path: Path to the whole slide image.
            output_path: Directory to save output masks.
            config: Optional configuration for the slide quality check.
            timeout: Optional timeout for the request.

        Returns:
            An xOpat link containing the processed WSI, enabling a quick visual
            inspection of the generated masks. The masks are displayed in the same
            format as in the QC report.
        """
        response = await self._put(
            "",
            json={
                "wsi_path": str(wsi_path),
                "output_path": str(output_path),
                **asdict(config or SlideCheckConfig()),
            },
            timeout=timeout,
        )
        response.raise_for_status()
        return response.text

    async def check_slides(
        self,
        wsi_paths: Iterable[PathLike[str] | str],
        output_path: PathLike[str] | str,
        config: SlideCheckConfig | None = None,
        timeout: TimeoutTypes | UseClientDefault = 3600,
        max_concurrent: int = 4,
    ) -> AsyncIterator[SlideCheckResult]:
        """Check quality of multiple slides.

        Args:
            wsi_paths: List of paths to the whole slide images.
            output_path: Directory to save output masks.
            config: Configuration for the slide quality check.
            timeout: Optional timeout for the request.
            max_concurrent: Maximum number of concurrent slide checks.

        Yields:
            An asynchronous generator yielding SlideCheckResult for each slide.
        """

        async def safe_check(path: PathLike[str] | str) -> SlideCheckResult:
            try:
                url = await self.check_slide(path, output_path, config, timeout)
                return SlideCheckResult(path, xopat_url=url, success=True)
            except Exception as e:
                return SlideCheckResult(path, error=str(e), success=False)

        pending: set[asyncio.Task[SlideCheckResult]] = set()
        for path in wsi_paths:
            if len(pending) >= max_concurrent:
                done, pending = await asyncio.wait(
                    pending, return_when=asyncio.FIRST_COMPLETED
                )
                for d in done:
                    yield await d

            pending.add(asyncio.create_task(safe_check(path)))

        for task in asyncio.as_completed(pending):
            yield await task

    async def generate_report(
        self,
        backgrounds: Iterable[PathLike[str] | str],
        mask_dir: PathLike[str] | str,
        save_location: PathLike[str] | str,
        compute_metrics: bool = True,
        timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
    ) -> None:
        """Generate a QC report from processed slides.

        Args:
            backgrounds: List of paths to the slide images.
            mask_dir: Directory containing the generated masks.
            save_location: Path where the report HTML will be saved.
            compute_metrics: Whether the generated report should contain aggregated
                metrics about the slide's coverage by the different classes of
                artifacts.
            timeout: Optional timeout for the request.
        """
        data = {
            "backgrounds": [str(bg) for bg in backgrounds],
            "mask_dir": str(mask_dir),
            "save_location": str(save_location),
            "compute_metrics": compute_metrics,
        }

        response = await self._put("report", json=data, timeout=timeout)
        response.raise_for_status()

check_slide(wsi_path, output_path, config=None, timeout=3600) async

Check quality of a single slide with automatic retry on failure.

Parameters:

Name Type Description Default
wsi_path PathLike[str] | str

Path to the whole slide image.

 required
output_path PathLike[str] | str

Directory to save output masks.

 required
config SlideCheckConfig | None

Optional configuration for the slide quality check.

 None
timeout TimeoutTypes | UseClientDefault

Optional timeout for the request.

 3600

Returns:

Type Description
str

An xOpat link containing the processed WSI, enabling a quick visual
str

inspection of the generated masks. The masks are displayed in the same
str

format as in the QC report.
Source code in rationai/resources/qc.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@retry(
    retry=retry_if_exception(_is_500_error),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
)
async def check_slide(
    self,
    wsi_path: PathLike[str] | str,
    output_path: PathLike[str] | str,
    config: SlideCheckConfig | None = None,
    timeout: TimeoutTypes | UseClientDefault = 3600,
) -> str:
    """Check quality of a single slide with automatic retry on failure.

    Args:
        wsi_path: Path to the whole slide image.
        output_path: Directory to save output masks.
        config: Optional configuration for the slide quality check.
        timeout: Optional timeout for the request.

    Returns:
        An xOpat link containing the processed WSI, enabling a quick visual
        inspection of the generated masks. The masks are displayed in the same
        format as in the QC report.
    """
    response = await self._put(
        "",
        json={
            "wsi_path": str(wsi_path),
            "output_path": str(output_path),
            **asdict(config or SlideCheckConfig()),
        },
        timeout=timeout,
    )
    response.raise_for_status()
    return response.text

check_slides(wsi_paths, output_path, config=None, timeout=3600, max_concurrent=4) async

Check quality of multiple slides.

Parameters:

Name Type Description Default
wsi_paths Iterable[PathLike[str] | str]

List of paths to the whole slide images.

 required
output_path PathLike[str] | str

Directory to save output masks.

 required
config SlideCheckConfig | None

Configuration for the slide quality check.

 None
timeout TimeoutTypes | UseClientDefault

Optional timeout for the request.

 3600
max_concurrent int

Maximum number of concurrent slide checks.

 4

Yields:

Type Description
AsyncIterator[SlideCheckResult]

An asynchronous generator yielding SlideCheckResult for each slide.
Source code in rationai/resources/qc.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
async def check_slides(
    self,
    wsi_paths: Iterable[PathLike[str] | str],
    output_path: PathLike[str] | str,
    config: SlideCheckConfig | None = None,
    timeout: TimeoutTypes | UseClientDefault = 3600,
    max_concurrent: int = 4,
) -> AsyncIterator[SlideCheckResult]:
    """Check quality of multiple slides.

    Args:
        wsi_paths: List of paths to the whole slide images.
        output_path: Directory to save output masks.
        config: Configuration for the slide quality check.
        timeout: Optional timeout for the request.
        max_concurrent: Maximum number of concurrent slide checks.

    Yields:
        An asynchronous generator yielding SlideCheckResult for each slide.
    """

    async def safe_check(path: PathLike[str] | str) -> SlideCheckResult:
        try:
            url = await self.check_slide(path, output_path, config, timeout)
            return SlideCheckResult(path, xopat_url=url, success=True)
        except Exception as e:
            return SlideCheckResult(path, error=str(e), success=False)

    pending: set[asyncio.Task[SlideCheckResult]] = set()
    for path in wsi_paths:
        if len(pending) >= max_concurrent:
            done, pending = await asyncio.wait(
                pending, return_when=asyncio.FIRST_COMPLETED
            )
            for d in done:
                yield await d

        pending.add(asyncio.create_task(safe_check(path)))

    for task in asyncio.as_completed(pending):
        yield await task

generate_report(backgrounds, mask_dir, save_location, compute_metrics=True, timeout=USE_CLIENT_DEFAULT) async

Generate a QC report from processed slides.

Parameters:

Name Type Description Default
backgrounds Iterable[PathLike[str] | str]

List of paths to the slide images.

 required
mask_dir PathLike[str] | str

Directory containing the generated masks.

 required
save_location PathLike[str] | str

Path where the report HTML will be saved.

 required
compute_metrics bool

Whether the generated report should contain aggregated metrics about the slide's coverage by the different classes of artifacts.

 True
timeout TimeoutTypes | UseClientDefault

Optional timeout for the request.

 USE_CLIENT_DEFAULT
Source code in rationai/resources/qc.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
async def generate_report(
    self,
    backgrounds: Iterable[PathLike[str] | str],
    mask_dir: PathLike[str] | str,
    save_location: PathLike[str] | str,
    compute_metrics: bool = True,
    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
) -> None:
    """Generate a QC report from processed slides.

    Args:
        backgrounds: List of paths to the slide images.
        mask_dir: Directory containing the generated masks.
        save_location: Path where the report HTML will be saved.
        compute_metrics: Whether the generated report should contain aggregated
            metrics about the slide's coverage by the different classes of
            artifacts.
        timeout: Optional timeout for the request.
    """
    data = {
        "backgrounds": [str(bg) for bg in backgrounds],
        "mask_dir": str(mask_dir),
        "save_location": str(save_location),
        "compute_metrics": compute_metrics,
    }

    response = await self._put("report", json=data, timeout=timeout)
    response.raise_for_status()