Compare commits

..

10 Commits

Author SHA1 Message Date
yxz321
f14e1bc4a0 plan: add modify reset plan 2026-05-21 19:53:58 +08:00
yxz321
247a0ee4c6 add and expose sync_from_external
从奔曜同步本地资源树+publish到deck
2026-05-21 19:37:02 +08:00
yxz321
a084031af0 improve: 使用调度状态接口做健康检查
- 将 Bioyond 启动健康检查从物料类型列表切换为 scheduler_status()。
- 避免 debug 日志输出完整物料类型列表。
- 保持连接监控响应更轻量、更易读。
2026-05-21 15:59:17 +08:00
yxz321
212f9ec448 plan: modify add_two_node plan for material unload 2026-05-20 19:19:02 +08:00
hanhua@dp.tech
2fd8f0d3f1 add plan 2026-05-20 11:55:08 +08:00
hanhua@dp.tech
a4678b7aa8 udpate reset 2026-05-18 18:26:16 +08:00
hanhua@dp.tech
72495bfc74 fix bug 2026-05-18 14:24:38 +08:00
hanhua@dp.tech
97ccc38c7f execute plan 2026-05-18 11:10:20 +08:00
yxz321
1df8fbd173 chore: cleanup peptide implementation to remove stale browser based pathway and other legacy helper functions. 2026-05-16 22:57:05 +08:00
yxz321
26155b8343 feat: PEP add Bioyond peptide station runtime
- Add the Bioyond peptide station package with the station-facing Day2 submission flow inlined into BioyondPeptideStation.
- Add LIMS sample Excel upload, Day2/Day3 order creation helpers, scheduler/reset controls, and manual-confirm start/reset actions.
- Register peptide material PLR resource classes and default peptide material type mappings for runtime resource synchronization.
- Add the Bioyond peptide deck definition and warehouse axis/key-axis metadata needed for peptide layout conversion.
- Update shared Bioyond warehouse/resource conversion helpers so peptide deck coordinates round-trip correctly.
- Include shared Bioyond raw-call debug logging support used by station actions, with a generic local debug output default.
- Register the peptide deck in PLR additional resources for deserialization/import visibility.
- Exclude private temp_benyao docs, HAR/API inputs, live diagnostics, and siRNA-only station/material files from this handoff commit.
2026-05-13 19:43:57 +08:00
18 changed files with 3683 additions and 5595 deletions

View File

@@ -1,914 +0,0 @@
# Draft: Resource And Material Sync Guidance For Sirna And Similar Bioyond Systems
Status: draft for discussion, not an implementation mandate.
This note triangulates across five source categories with different authority:
1. `docs/developer_guide/examples/workstation_architecture.md`: desired shape
and vocabulary. It is a design target, not proof that the current code has
every behavior.
2. `BioyondWorkstation`, `BioyondResourceSynchronizer`, and shared graphio code:
the current compatibility anchor. New guidance should mostly preserve this
lifecycle and extend it deliberately.
3. Existing non-Sirna Bioyond stations: practical examples of how the shared
base is used, including shortcuts that should not become policy.
4. Sirna implementation, plans, findings, and guide notes: stress-test evidence.
They expose real missing cases, but the current Sirna code is not the
architecture authority because parts were added without fully aligning to the
shared base.
5. UniLabOS resource framework behavior around PLR resources,
`ResourceTreeSet`, serialize/deserialize, and `update_resource(resources=...)`.
The short recommendation is:
Keep the shared Bioyond workstation lifecycle as the center of gravity. Evolve
the shared synchronizer with small project hooks for classification, ID-based
resolution, and non-slot material handling, rather than letting Sirna become a
parallel synchronization model. For Sirna, those hooks should resolve placement
by Bioyond IDs, distinguish physical slot labware from reagent liquid contents,
preserve Bioyond IDs in `unilabos_extra`, mutate the local PLR deck, and publish
the full deck through `update_resource(resources=[deck])`.
Do not treat every Bioyond stock row as a deck resource.
## Evidence Weighting
This task is not a Sirna implementation retrospective. It is a best-practice
alignment pass across the architecture target, the shared base class, observed
station behavior, and Sirna's newly exposed edge cases.
Use the sources this way:
- Architecture doc: ask "what shape should this system eventually have?"
- Shared Bioyond base: ask "what behavior must remain compatible today?"
- Other Bioyond stations: ask "what patterns are already working in practice?"
- Sirna current code and notes: ask "what did the base model fail to handle, and
which Sirna fixes conflict with the shared lifecycle?"
- Live API/schema evidence: ask "what is true for this deployment's material,
warehouse, and coordinate data?"
The Sirna AGENT_GUIDE is useful for finding caveats and prior investigations,
but it should not be cited as an independent source of truth when source code,
framework behavior, live API evidence, or architecture docs can answer the same
question.
## Mental Model
There are three resource worlds. Confusing them is the main source of bugs.
| World | Owner | Purpose | Recommended truth |
| --- | --- | --- | --- |
| Bioyond/LIMS | Bioyond APIs through `BioyondV1RPC` | External stock, material IDs, warehouse IDs, location IDs, inbound/outbound side effects | External material truth |
| PLR deck | Workstation driver | Runtime workstation material layout, warehouse occupancy, liquid contents | Local mutation surface |
| UniLabOS resource tree | `ResourceTreeSet` / ROS node / host resource APIs | Canonical UniLabOS/cloud representation | Framework/cloud truth |
The architecture guide describes `Deck` as the local material system and
`ResourceSynchronizer` as the optional external-system bridge
(`docs/developer_guide/examples/workstation_architecture.md:221`). The broader
framework serializes PLR objects into `ResourceTreeSet`, whose resource dicts
carry `id`, `uuid`, `parent_uuid`, `type`, `class`, `pose`, `config`, `data`,
and `extra` (`unilabos/resources/resource_tracker.py:107`).
Therefore the correct path is:
```text
Bioyond rows
-> normalized material records
-> preserve Bioyond materialTypeMode: Sample / Consumables / Reagent
-> resolve material/location/warehouse IDs
-> choose UniLabOS handling for that mode
-> mutate PLR deck
-> ResourceTreeSet.from_plr_resources([deck])
-> ROS update_resource(resources=[deck])
-> host/cloud resource-tree update
```
Avoid direct cloud JSON patches and avoid treating Bioyond records as already
being UniLabOS resource nodes.
## Target Architecture
The ideal architecture in `workstation_architecture.md` is still the right
direction:
1. `WorkstationBase` owns the local `deck`, workflow state, and hardware
interface.
2. `ResourceSynchronizer` owns external material synchronization.
3. `BioyondResourceSynchronizer` owns the Bioyond-specific use of
`BioyondV1RPC`.
4. `ROS2WorkstationNode.update_resource(resources)` owns the UniLabOS/cloud
resource-tree update.
5. Optional HTTP report handlers may trigger local deck mutation, external sync,
and cloud publication.
The documented startup sequence is:
1. Create workstation.
2. Initialize PLR deck and warehouses.
3. Create Bioyond RPC hardware interface.
4. Create resource synchronizer.
5. `sync_from_external()`.
6. Initialize ROS node and children.
7. `post_init(ros_node)`.
8. Upload `resources=[deck]`.
See `docs/developer_guide/examples/workstation_architecture.md:308`,
`docs/developer_guide/examples/workstation_architecture.md:497`, and
`docs/developer_guide/examples/workstation_architecture.md:737`.
Important caveat: that document is the hope. It is still valuable because it
names the desired responsibilities, but current Bioyond stations implement a
more limited, best-effort side-effect sync. When the doc and current code
diverge, use the doc to choose direction and the shared base code to choose the
next compatible step.
## Current Practical Behavior
The shared `BioyondWorkstation` implementation is the current behavioral
baseline. It does this today:
1. Requires a deck.
2. Reconstructs `deck.warehouses` from deck children/config when needed.
3. Creates `BioyondV1RPC`.
4. Installs `BioyondResourceSynchronizer`.
5. Immediately calls `sync_from_external()`.
6. Later, in `post_init`, publishes the whole deck with
`ROS2DeviceNode.run_async_func(self._ros_node.update_resource, True,
resources=[self.deck])`.
Relevant code:
- `BioyondResourceSynchronizer` is defined in
`unilabos/devices/workstation/bioyond_studio/station.py:117`.
- `sync_from_external()` fetches stock `typeMode` 0, 1, and 2, then passes all
rows to `resource_bioyond_to_plr(...)`
(`unilabos/devices/workstation/bioyond_studio/station.py:147`).
- `BioyondWorkstation.__init__` installs the synchronizer and syncs immediately
(`unilabos/devices/workstation/bioyond_studio/station.py:856`).
- `post_init()` uploads the deck (`unilabos/devices/workstation/bioyond_studio/station.py:893`).
- `resource_tree_add()` performs Bioyond create/inbound side effects
(`unilabos/devices/workstation/bioyond_studio/station.py:958`).
This practical behavior works for simple physical-resource stock import, but it
does not provide full continuous two-way sync, conflict resolution, or reliable
stale-state cleanup. In particular:
- Re-fetching stock does not clearly clear stale local deck state first.
- Local-to-external update no-ops unless `unilabos_extra["update_resource_site"]`
is present.
- `process_material_change_report()` is mostly TODO-level in the base station.
- Some station-specific code bypasses the shared synchronizer with direct LIMS
calls.
- Some existing stations contain shortcuts that should not be copied, such as
duplicate initialization, stale globals, or hardcoded warehouse/axis cases.
Treat current Bioyond behavior as operationally useful and compatibility
important, not as proof that the ideal architecture is already achieved. The
goal is to tighten this base path, not replace it with a Sirna-only path.
## Recommended Shared Pipeline
For Sirna and similar Bioyond systems, the base synchronizer should evolve from
the current shared path into one shared pipeline with small project hooks:
```text
sync_from_external()
fetch stock rows from Bioyond
update RPC material cache
normalize rows into a common internal shape
preserve materialTypeMode: Sample / Consumables / Reagent
resolve warehouse/location by Bioyond IDs when available
choose handling for the row's mode
apply Sample / Consumables rows as mapped slot labware by default
apply Reagent rows as physical reagent labware or liquid content by evidence
report unknown modes or unmapped handling loudly
publish deck if deck changed and ROS node is available
```
Suggested extension points:
```python
class BioyondResourceSynchronizer(ResourceSynchronizer):
def external_material_mode(self, row: dict) -> str:
return row["materialTypeMode"] # Sample | Consumables | Reagent
def resolve_external_row(self, row: dict) -> dict:
...
def apply_external_row(self, row: dict, mode: str, resolved: dict) -> None:
...
```
The base class should continue to own:
- stock fetches for `typeMode` 0/1/2;
- material-cache refresh;
- common normalization;
- mode validation for `Sample` / `Consumables` / `Reagent`;
- publication orchestration;
- error aggregation;
- stale-state policy once agreed.
Project code should own only the parts that are truly deployment-specific:
- material type/mode to PLR class mapping;
- project-local handling inside `Sample` / `Consumables` / `Reagent`;
- project-local warehouse ID/name mapping;
- project-local coordinate conventions;
- special row handling such as Sirna reagent-as-liquid.
This keeps existing Bioyond stations close to the same lifecycle while still
absorbing the Sirna lessons that the earlier base did not model. The default
hook behavior can remain "Sample/Consumables/Reagent rows become mapped slot
labware" for simpler stations; Sirna should override Reagent handling where
evidence requires liquid-content behavior.
## Sirna Findings To Feed Back Into Shared Design
Treat Sirna as a stress test for the shared base, not as a replacement design.
It raises real questions that earlier stations did not need to answer:
- within the `Reagent` mode, some external rows may be liquid contents rather
than slot-occupying reagent labware;
- placement should be ID-first where Bioyond supplies material/location IDs;
- warehouse and axis metadata must survive serialize/deserialize;
project-specific mode handling is available.
For Sirna-like deployments, the old implicit rule "stock row equals PLR
resource" is too coarse. For simpler deployments, it can remain the default
mode-handling behavior.
### IDs Win
Placement identity should prefer:
```text
materialId
locationId
materialTypeId
warehouseId / whid
```
Display/debug fields are not identity:
```text
materialName
materialCode
locationCode
locationShowName
warehouseName / whName
```
`locationCode` such as `1-1` is not globally unique. It can exist in multiple
warehouses. Code-only resolution may be kept as a diagnostic fallback, but it
must raise on ambiguity.
The current Sirna ID-first resolver is useful source evidence for this direction
(`unilabos/devices/workstation/bioyond_studio/sirna_station/sirna_station.py:3659`).
The Sirna mega plan captures the same concern, but the implementation and live
Bioyond IDs are the stronger evidence.
### Bioyond Modes And UniLabOS Handling Are Different
Bioyond has three primary material modes in this context:
```text
Sample
Consumables
Reagent
```
Those modes should be preserved as the external taxonomy. UniLabOS still needs a
handling decision inside the mode: should the row create/place a physical PLR
resource, or update contents on an already-placed parent resource?
Physical slot resources include things like:
- tip racks;
- plates;
- cell culture plates;
- empty trough/bottle labware;
- other objects that occupy a warehouse slot.
Reagents are contents of a parent labware when the row describes a liquid rather
than a physical holder. For Sirna, Bioyond `stock_material(typeMode=2)` returns
`materialTypeMode="Reagent"` rows; those rows still need evidence-based handling
as either physical reagent labware or reagent liquid content. The finding in
`temp_benyao/sirna/_findings/2026-05-07_reagents_vs_resources.md:6` records the
bug: the generic path can fall back to `RegularContainer`, fail registry lookup,
and drop reagent rows such as `试剂槽裂解液/Betame`.
Recommended behavior:
1. Validate `materialTypeMode` as `Sample`, `Consumables`, or `Reagent`.
2. For `Sample` and `Consumables`, default to mapped slot labware: instantiate
the mapped PLR class and place it into the resolved warehouse slot.
3. For `Reagent`, decide whether the row represents physical reagent labware or
reagent liquid content from material type evidence, row shape, and live data.
4. For physical reagent labware, place the mapped PLR resource in the resolved
slot.
5. For reagent liquid content, find the parent trough/bottle and attach liquid
metadata to its tracker.
6. Preserve Bioyond IDs in `parent.unilabos_extra["reagent_bioyond_ids"]`.
7. Make liquid attachment idempotent by Bioyond material ID.
8. If the parent labware is missing or the mode/handling cannot be resolved,
defer or log with IDs; do not guess.
The existing Sirna helper `_attach_liquid_to_parent()` already follows the
idempotent metadata direction
(`unilabos/devices/workstation/bioyond_studio/sirna_station/sirna_station.py:4074`).
### The `0003` Question Must Stay Evidence-Based
Do not blindly map `materialTypeCode 0003` to `BioyondSirna_ReagentTrough`.
If live/schema evidence says a `0003` row is physical trough labware, map it to
a PLR resource class. If evidence says it is liquid content, attach it to the
parent trough. If evidence is contradictory, mark it unsupported and log the
Bioyond IDs.
Treat this as an open design decision until source/schema/live evidence settles
the row shape. The Sirna plan records the question, but it should not decide the
mapping by itself.
### Sirna Warehouse And Axis Rules Are Project-Local
Sirna warehouse layout and axis conventions must be verified from Sirna source,
Sirna schema, current deck behavior, and live/read-only Sirna APIs when
available. Do not import Peptide, reaction, dispensing, or cell station layout
truth.
For Sirna specifically, the current integrated deck display should be treated as
the good baseline. The prior col-row slot-key fix is already present in the
deck/graphio path, and any remaining x/y or y-reverse correction should be
handled through shared warehouse metadata and shared graphio/display mapping
rather than by reshaping the Sirna station display ad hoc.
### Display Geometry: Evidence Before Axis Values
Peptide resource sync should not be treated as a validated model, but its
UniLabOS display work is useful and should influence this guidance.
Confirmed Peptide behavior:
- Peptide live warehouse evidence found `自动化堆栈` with 170 locations where
`code="10-17"` appears at `x=17, y=10`
(`../Uni-Lab-OS-Peptide/temp_benyao/peptide/_findings/2026-05-13_1404_peptide_col_row_deck.md:6`).
- Peptide's current display convention is Peptide-specific evidence:
`bioyond_axis="xy_col_row"` and `bioyond_key_axis="col_row"` in the current
Peptide station implementation. Do not copy that pair into Sirna or any other
project without live warehouse evidence for that target system. With this
Peptide combination, graphio does not apply the legacy x/y swap
(`../Uni-Lab-OS-Peptide/unilabos/resources/graphio.py:868`).
- Peptide models the main automation stack as 17 visual rows by 10 visual
columns while preserving labels such as `10-17`
(`../Uni-Lab-OS-Peptide/unilabos/resources/bioyond/decks.py:308`).
- Peptide warehouses and deck child positions use `frontend_y_flip=True` /
`_frontend_y_flipped_coordinate(...)` so stored PLR coordinates compensate for
the frontend y-axis inversion
(`../Uni-Lab-OS-Peptide/unilabos/resources/bioyond/decks.py:299`;
`temp_benyao/peptide/_findings/2026-05-13_1514_frontend_y_flip_layout.md:6`).
- The older Peptide graph-layout note is stale for the Sirna discussion. The
current Sirna integrated station/deck display is a good baseline; the reusable
Peptide lesson is axis metadata and frontend y-reverse compatibility.
- Peptide tests encode the intended behavior: `10-17` lands at row 17 /
column 10, and after frontend y-flip the displayed site positions match the
expected top-to-bottom layout
(`../Uni-Lab-OS-Peptide/temp_benyao/peptide/tests/test_peptide_deck_layout.py:60`).
Guidance for Sirna and similar systems:
1. Treat `bioyond_axis` and `bioyond_key_axis` as two separate concepts:
- `bioyond_axis` describes how Bioyond numeric `x/y` map to visual axes.
- `bioyond_key_axis` describes how slot labels such as `10-17` are generated
or preserved.
2. Do not infer display orientation from label strings alone. Both `row_col`
and `col_row` can preserve the same final label text while producing
different visual layouts and graphio swap behavior.
3. Use live/read-only `code/x/y` evidence for each project and each warehouse
family before choosing axis metadata.
4. Keep visual orientation fixes separate from material identity. IDs and slot
labels determine registration; display dimensions and frontend y-flip only
determine how the deck appears.
5. Author intended display coordinates first, then convert stored y coordinates
for the active frontend convention:
- deck child stored y = `deck_height - display_y - child_height`;
- warehouse site stored y = `warehouse_height - display_y - site_height`;
- graph-level y values should be transformed only when the active frontend
convention requires it.
6. Preserve the current Sirna display as the baseline unless concrete frontend
evidence shows a y-reverse problem. Do not change station/deck graph
semantics just to match stale Peptide layout notes.
7. Add layout tests that assert:
- warehouse `num_items_x`, `num_items_y`, capacity, first key, and last key;
- representative `code/x/y` examples land on the intended site key;
- frontend y-flip produces expected displayed positions;
- generated deck children do not overlap in displayed coordinates.
8. When borrowing from Peptide, borrow the evidence pattern: live discovery,
axis/key-axis metadata, y-flip tests, and display fixtures. Do not borrow a
literal axis pair or Peptide resource-sync behavior as a validated path.
Concrete live discovery workflow:
1. Prefer the reusable read-only probe pattern before reading old findings:
```bash
python3 temp_benyao/sirna/tests/probe_readonly_storage_inventory.py \
--base-url <api_host> \
--api-key <api_key> \
--output temp_benyao/<project>/_logs/<timestamp>_readonly_storage_inventory_probe.json
```
This probe checks swagger candidates, project storage/location endpoints,
material type endpoints, `warehouse-info-by-mat-type-id`, and `stock-material`,
then writes a redacted evidence file and a merged warehouse summary.
2. For Sirna quick checks, `temp_benyao/sirna/tests/discover_sirna_warehouses.py`
is the narrower historical helper. Treat it as a template unless it has been
parameterized for the target config.
3. Query `/api/storage/location/locations-by-type?type=0&typeMode=0&materialType=0`
first for stack names, warehouse IDs, full slot lists, `code`, `x`, `y`, `z`,
and display mode. This endpoint is the topology source when available.
4. Cross-reference with `/api/lims/storage/material-types` and
`/api/lims/storage/warehouse-info-by-mat-type-id` to learn material-type
placement constraints. Use `stock-material` only for occupied-slot evidence;
it cannot reveal empty topology.
5. Infer `bioyond_key_axis` from how slot labels must be generated or preserved,
and infer `bioyond_axis` from how raw Bioyond `x/y` must map to PLR holder
indices. A label such as `10-17` alone is ambiguous; compare it to the same
record's `x/y`, and use boundary examples from non-square stacks.
6. Encode the discovered convention on the warehouse resource, not in a one-off
station branch. The metadata must serialize/deserialize with the warehouse
because graph load and cloud sync rebuild resources.
7. Add or update layout tests before changing Sirna display behavior. For Sirna,
test against the current good display first, then only change shared x/y or
y-reverse mapping if the fixture demonstrates a real mismatch.
### Deserialize Must Be Idempotent
This is confirmed framework behavior, not a possible risk. Resource publication
builds `ResourceTreeSet` from live PLR resources by calling
`resource.serialize()` and `resource.serialize_all_state()`
(`unilabos/resources/resource_tracker.py:547`). Readback/reconstruction goes
through `ResourceTreeSet.to_plr_resources()`, finds the PLR subclass, calls
`sub_cls.deserialize(...)`, then restores PLR state, UUIDs, and `unilabos_extra`
(`unilabos/resources/resource_tracker.py:637`). `ResourceTreeSet.dump()` also
serializes resource nodes while excluding `children` from each individual node
record (`unilabos/resources/resource_tracker.py:914`), so parent/child
relationships and object state must survive the framework tree shape rather than
only an in-memory PLR object graph.
Sirna resource/deck classes therefore must survive:
1. registry-time construction;
2. `Resource.deserialize()` round trips;
3. cloud-synced deck state with serialized children.
If a synced deck already has serialized `children`, default setup must not
create duplicate/stale child resources. Sirna noticed this problem, but the
reason is framework-level: cloud/resource-tree sync reconstructs PLR objects
from serialized state, so constructors and setup logic must be idempotent.
Existing Sirna deck code already acknowledges the issue: `BIOYOND_Sirna_Deck`
turns `setup=False` during deserialize when serialized `children` are present
(`unilabos/resources/bioyond/decks.py:153`). Sirna material classes also use
registry-visible `@resource(...)` decorators and tolerant constructors with
`*args, **kwargs`, defaults, and `ordering` fallbacks
(`unilabos/resources/bioyond/sirna_materials.py:14`). Treat those as required
patterns for any new synced resource class.
Guidance:
- Do not add a deck/resource class until it round-trips through
`Resource.serialize()` / `Resource.deserialize()` and
`ResourceTreeSet.from_plr_resources(...).to_plr_resources()`.
- Preserve `unilabos_uuid`, parentage, `unilabos_extra`, and liquid state across
the round trip.
- Make default setup conditional: if serialized children exist, do not recreate
default warehouses or default child resources on top of them.
- Itemized PLR subclasses must provide `ordered_items` or `ordering`; otherwise
deserialization can fail or rebuild a different structure.
## Startup And Resync Recommendation
The current Sirna code installs `SirnaResourceSynchronizer` after
`super().post_init(ros_node)` (`sirna_station.py:363`). That is useful as a
phase patch and as evidence that classification hooks are needed, but it should
not become the long-term lifecycle. Base initialization may already have run
generic stock sync and base post-init may already have published the deck before
the Sirna synchronizer is installed.
Recommended pathway:
1. Keep the `BioyondWorkstation` lifecycle as the base path.
2. Add a synchronizer factory or hook registration point before eager sync, for
example `create_resource_synchronizer()`.
3. Let Sirna provide classification/resolution hooks through that shared
synchronizer shape.
4. Make startup sync use the installed hook-aware synchronizer before the first
deck publication.
5. Make manual resync, reset resync, report-triggered resync, and future
periodic sync all call `self.resource_synchronizer.sync_from_external()`.
6. Do not instantiate fresh base synchronizers inside Sirna actions, because
that bypasses the installed project logic.
This directly addresses the overlap finding in
`temp_benyao/sirna/_findings/2026-05-12_sirna_synchronizer_overlap.md:6`.
## Material Registration From Create-Order Results
Create-order allocation registration should remain separate from stock sync, but
it should use the same classification and apply logic.
Recommended create-order pipeline:
1. Normalize allocation records into:
```python
{
"materialId": "...",
"materialCode": "...",
"materialName": "...",
"materialTypeId": "...",
"materialTypeCode": "...",
"materialTypeMode": "Sample|Consumables|Reagent",
"materialTypeName": "...",
"locationId": "...",
"locationCode": "...",
"locationShowName": "...",
}
```
2. Resolve warehouse and slot by:
```text
material-info(materialId)
-> warehouse-info-by-mat-type-id(materialTypeId) matched by locationId
-> code-only diagnostic fallback only if unambiguous
```
3. Classify as `slot_labware`, `liquid_content`, or `unsupported`.
4. Apply mutation to the PLR deck.
5. Publish the full deck once after the batch.
The current Sirna `_register_materials_to_tree()` already documents this flow
(`sirna_station.py:3659`) and should be the local reference implementation until
the shared hook is designed.
## Cloud / UniLabOS Publication Rules
Always mutate real tracked PLR resources first. Then publish through the
framework.
Correct call shape:
```python
ROS2DeviceNode.run_async_func(
self._ros_node.update_resource,
True,
**{"resources": [self.deck]},
)
```
The real method is:
```python
async def update_resource(self, resources: List["ResourcePLR"])
```
See `unilabos/ros/nodes/base_device_node.py:727`.
Do not call:
```python
update_resource(resource_name=..., resource_data=...)
```
Do not manually serialize the deck for this path. `update_resource()` creates a
`ResourceTreeSet`, sends it to `/c2s_update_resource_tree`, and applies returned
UUID mappings. Missing root parent UUIDs are auto-mounted to the current device,
so parentage should be preserved on PLR objects before publication.
## Metadata Contract
Every Bioyond-originated slot resource should carry enough metadata for unload,
audit, and later sync:
```python
plr_resource.unilabos_extra = {
"material_bioyond_id": mat["materialId"],
"material_bioyond_code": mat["materialCode"],
"material_bioyond_name": mat["materialName"],
"material_bioyond_type_id": mat["materialTypeId"],
"material_bioyond_type_code": mat["materialTypeCode"],
"material_bioyond_type_mode": mat["materialTypeMode"],
"location_bioyond_id": mat["locationId"],
"location_code": resolved["location_code"],
"warehouse_bioyond_id": resolved["warehouse_id"],
"warehouse_bioyond_name": resolved["warehouse_name"],
"location_resolution_source": resolved["source"],
}
```
Every Bioyond-originated liquid content should preserve equivalent metadata on
the parent labware:
```python
parent.unilabos_extra.setdefault("reagent_bioyond_ids", []).append({
"material_bioyond_id": mat["materialId"],
"material_bioyond_code": mat["materialCode"],
"material_bioyond_name": mat["materialName"],
"material_bioyond_type_id": mat["materialTypeId"],
"material_bioyond_type_code": mat["materialTypeCode"],
"location_bioyond_id": mat["locationId"],
"quantity": mat.get("quantity"),
"location_resolution_source": resolved["source"],
})
```
This metadata is not decorative. It is required for correct unload, audit,
duplicate prevention, and future round-trip sync.
## Stale State And Conflict Policy
This is confirmed current behavior, not just a theoretical risk.
Current stock import places resources into empty warehouse slots and skips
occupied slots. That prevents overwrites, but it can leave stale local resources
after external moves/deletes.
Evidence:
- `BioyondResourceSynchronizer.sync_from_external()` fetches Bioyond stock and
delegates to `resource_bioyond_to_plr(...)`, but it does not compute a
before/after diff or remove local resources absent from the snapshot
(`unilabos/devices/workstation/bioyond_studio/station.py:147`).
- `resource_bioyond_to_plr(...)` only assigns when the target warehouse position
is empty or a placeholder; when a real resource already occupies the slot, it
logs "跳过放置" and leaves the existing object in place
(`unilabos/resources/graphio.py:936`).
- Bioyond outbound/removal logic exists in separate local-to-external hooks such
as `resource_tree_remove(...)`, but that is not invoked by stock refresh for
Bioyond rows that disappeared externally
(`unilabos/devices/workstation/bioyond_studio/station.py:987`).
Recommended direction:
1. Define Bioyond as the source of truth for external stock snapshots unless a
local operation is in progress.
2. Before applying a stock snapshot, compute a diff by Bioyond material ID.
3. Remove or mark local resources whose Bioyond IDs disappeared from the
snapshot, subject to workflow safety checks.
4. Move local resources whose Bioyond location ID changed.
5. Attach/detach liquid contents idempotently by Bioyond material ID.
6. Publish once after applying the batch.
7. In ambiguous or active-operation cases, log and require manual confirmation.
Until this policy exists, call the current implementation "refresh/import" rather
than "authoritative synchronization."
## External-To-Local And Local-To-External Boundaries
External-to-local:
- Bioyond stock and create-order allocation rows mutate the local PLR deck.
- Then UniLabOS publication derives resource trees from the PLR deck.
Local-to-external:
- `resource_tree_add` / update / remove may call Bioyond add, inbound, outbound,
or move APIs.
- These paths should require Bioyond IDs or explicit creation parameters.
- Movement should use `unilabos_extra["update_resource_site"]` only as an
explicit request, not as hidden ambient state.
- After successful Bioyond side effects, refresh or update the PLR deck and
publish the deck.
Avoid station-level direct LIMS calls that bypass the synchronizer unless the
action explicitly reconciles the deck afterward.
## Error Handling Rules
Fail loudly on:
- unknown material type code/name;
- unresolved warehouse ID/name;
- location ID not found in `warehouse-info-by-mat-type-id`;
- code-only location ambiguity;
- missing parent labware for liquid content;
- invalid PLR resource class or `Resource.deserialize()` failure;
- Bioyond RPC returning empty/fallback values where an ID is required.
Do not silently create `RegularContainer` to hide mapping failures. The Sirna
finding at `temp_benyao/sirna/_findings/2026-05-07_reagents_vs_resources.md:52`
calls this out as a reusable trap.
## Tests And Validation
Minimum offline tests:
1. `update_resource` is called only with `resources=[deck]`.
2. `ResourceTreeSet.from_plr_resources([deck])` preserves UUIDs, parentage,
`data`, and `unilabos_extra`.
3. Sirna allocation records resolve by `material-info` first.
4. `warehouse-info-by-mat-type-id` resolves by `locationId`.
5. Code-only fallback raises on ambiguous slots.
6. `Sample` and `Consumables` rows become slot labware.
7. Reagent content rows become parent liquid metadata.
8. Re-running registration does not duplicate liquid entries.
9. Missing parent labware is deferred/logged, not guessed.
10. Unknown material type emits Bioyond IDs.
11. Deck deserialize does not recreate duplicate default children.
12. Warehouse coordinate mapping is tested per warehouse, not globally.
13. A synced Sirna deck round-trips through PLR deserialize and
`ResourceTreeSet` conversion without duplicate default children.
14. Stock refresh with a missing Bioyond material ID proves current add/skip
behavior first, then verifies the chosen diff/delete policy once implemented.
15. Display geometry tests cover project-local live discovery fixtures,
`bioyond_axis`, `bioyond_key_axis`, representative `code/x/y` mappings,
frontend y-flip, and no-overlap deck layout.
Focused Sirna command:
```bash
pytest temp_benyao/sirna/tests/test_sirna_resource_system.py -q
```
Shared conversion command, when changing graphio or Bioyond converters:
```bash
pytest tests/resources/test_converter_bioyond.py -q
```
Live/read-only validation should capture fixtures for:
- `stock_material(typeMode=0/1/2, includeDetail=true)`;
- `material-info(materialId)`;
- `warehouse-info-by-mat-type-id(materialTypeId)`;
- create-order allocation records;
- frontend/cloud resource-tree readback after publication.
## Discrepancies To Discuss
### 1. Architecture Is Direction; Base Code Is The Compatibility Anchor
The architecture guide presents `ResourceSynchronizer` as bidirectional. Current
Bioyond code is mostly eager external import plus selected add/remove/update
side effects. It has no complete continuous conflict-resolution loop.
Sirna current code should be read in this context: it exposes missing
capabilities, but parts of it conflict with the existing base lifecycle because
it was layered on after the base sync behavior already existed.
Recommendation: keep the bidirectional architecture as the target, preserve
`BioyondWorkstation` as the implementation anchor, and introduce shared hooks
for the missing Sirna-class problems. Describe the current implementation as
best-effort refresh plus Bioyond side effects until a diff/conflict policy
exists.
### 2. The Doc Centers `Deck`; The Framework Centers `ResourceTreeSet`
The architecture guide uses PLR `Deck` as the material system. UniLabOS resource
tracking uses `ResourceTreeSet` as the canonical serialized representation.
Recommendation: phrase the model as "Deck is the local PLR mutation surface;
ResourceTreeSet is the UniLabOS/cloud representation derived from it."
### 3. Base Bioyond Sync Treats Every Stock Row As PLR Resource
`BioyondResourceSynchronizer.sync_from_external()` currently feeds typeMode
0/1/2 rows into `resource_bioyond_to_plr(...)`.
Sirna evidence shows this is wrong for reagent/content-like rows.
Recommendation: add classification hooks before conversion. Rows classified as
liquid content must not go through standalone PLR resource conversion.
### 4. Sirna Fix Exists As A Fork, Not A Shared Hook
Current `SirnaResourceSynchronizer` captures the important reagent-as-liquid
question, but it duplicates or bypasses shared sync mechanics. Startup can run
base sync before the Sirna synchronizer is installed, and manual/reset paths
have used fresh base synchronizers.
Recommendation: keep the discovered behavior where evidence supports it, but
refactor the shape. A factory/hook in the shared synchronizer is preferable to a
long-lived parallel sync class.
### 5. Sirna Create-Order Registration Is Stronger Than Stock Sync
Create-order registration now has ID-first resolution and a clearer classifier.
That is valuable evidence, not automatically the canonical sync design. The
stock sync path still resolves external liquid rows by warehouse name/code and
stores weaker metadata.
Recommendation: extract the stronger resolver/classifier ideas into shared
hooks, then converge stock sync and create-order registration on the same
functions.
### 6. Graphio Has Shared Fragility
`resource_bioyond_to_plr()` contains hardcoded warehouse/coordinate cases and a
fallback to `RegularContainer`. Hardening it affects multiple Bioyond projects.
Recommendation: after Sirna is green, discuss shared graphio hardening in a
separate change: no silent `RegularContainer`, explicit unknown-type diagnostics,
and project-specific axis metadata instead of hardcoded names.
### 7. Serialize/Deserialize Is A Gating Contract
This is not optional documentation neatness. Any new resource/deck class or sync
metadata must pass the framework serialize/deserialize path before it can be
trusted in cloud-synced operation.
Recommendation: make round-trip tests part of acceptance for resource-system
changes, especially for Sirna deck setup, warehouse metadata, `reagent_bioyond_ids`,
liquid tracker state, and itemized labware ordering.
### 8. Existing Stock Sync Is Add/Skip-Biased, Not Delete-Aware
Current `sync_from_external()` imports and places observed Bioyond resources,
but it does not delete local resources missing from the latest stock snapshot.
When a target slot is occupied by a real resource, graphio skips placement rather
than replacing it.
Recommendation: describe the current behavior as "stock refresh/import" until a
Bioyond-ID diff policy is implemented. Deletion/removal must be a deliberate
phase, with workflow-safety checks and tests, not implied by the current stock
sync name.
### 9. Peptide Display Is Useful; Peptide Sync Is Not Yet Authority
Peptide's latest display work is useful evidence for Peptide's own warehouse
axis/key-axis metadata and for the shared need to account for frontend y-axis
inversion. It does not define Sirna's axis values. Its resource sync remains
untested/problematic and should not be treated as the model for Sirna sync
behavior.
Recommendation: fold the Peptide display lesson into shared guidance as a
discovery-and-test pattern, not as a literal axis pair. Keep Sirna display
behavior grounded in Sirna live evidence and the current good deck baseline; keep
resource synchronization recommendations grounded in the Sirna ID-first
registration path and the shared Bioyond synchronizer refactor.
### 10. Some Existing Stations Should Be Examples Only, Not Templates
Non-Sirna stations show useful patterns, but also shortcuts:
- direct LIMS calls bypassing the synchronizer;
- duplicate `super().__init__()` in reaction station;
- stale/undefined `WAREHOUSE_MAPPING` references;
- RPC methods collapsing failures into empty values;
- no robust stale deck cleanup on stock refresh.
Recommendation: borrow the shared deck publication and Bioyond ID metadata
patterns, not the incidental shortcuts.
## Recommended Discussion Path
For the next implementation discussion, decide these in order:
1. What is the smallest shared-base extension that lets project hooks exist
before eager sync without breaking existing Bioyond stations?
2. Should `BioyondWorkstation` gain a synchronizer factory, a hook registry, or
a delayed eager-sync point?
3. Is `materialTypeCode 0003` in current Sirna live data physical labware,
liquid content, or both depending on row shape?
4. Where should Sirna warehouse Bioyond IDs live: station config, deck metadata,
or both?
5. Should stock refresh clear/diff deck state now, or remain append/skip until
after create-order registration is stable?
6. Should `update_resource` remain async/non-blocking for material registration,
or should selected user-facing actions fail if publication fails?
7. Which parts of the Sirna reagent-as-liquid rule are project-local, and which
should become shared Bioyond behavior after Peptide/cell validation?
8. Should graphio fallback hardening land before or after Sirna stock sync is
fully validated?
9. Should any remaining Sirna x/y or y-reverse issue be fixed in shared
graphio/display mapping after a live fixture proves it, or should the current
Sirna deck remain untouched?
## Recommended Pathway
The pragmatic path is:
1. Preserve and test the current `BioyondWorkstation` lifecycle as the shared
compatibility baseline.
2. Treat Sirna create-order registration as a local proof of the missing
classification/ID-resolution behavior, not as the architecture shape to copy.
3. Validate live/read-only Sirna fixtures for `0003`, warehouse IDs, and stock
reagent rows.
4. Refactor the shared Bioyond synchronizer to expose classification,
resolution, and non-slot row hooks with default behavior matching existing
stations.
5. Make Sirna install hooks before the first external sync and first deck
publication through the shared lifecycle.
6. Route startup, manual resync, reset resync, and report-triggered resync
through the installed synchronizer.
7. Converge stock sync and create-order registration on one
resolver/classifier/apply implementation.
8. Add stale-state diffing only after the basic ID-first path is stable.
9. Harden graphio fallback as a shared follow-up.
This gives Sirna the necessary behavior without locking in a second Bioyond sync
system that future projects will have to debug around.

View File

@@ -0,0 +1,576 @@
# Peptide Station 新增三个节点:等待订单完成 + 下料确认 + take-out 同步
> 日期: 2026-05-20
> 目标文件: [peptide_station.py](../unilabos/devices/workstation/bioyond_studio/peptide_station/peptide_station.py)
> 参考实现:
> - [bioyond_cell_workstation.py](/Users/dp/python/yxz/unilabos/devices/workstation/bioyond_studio/bioyond_cell/bioyond_cell_workstation.py)`wait_for_order_finish`、`get_material_info`
> - [bioyond_rpc.py L782-824](../unilabos/devices/workstation/bioyond_studio/bioyond_rpc.py)`take_out`
> - [workstation_architecture.md](../docs/developer_guide/examples/workstation_architecture.md)HTTP 报送进入 workstation运行态记录保存在 workstation 内存)
> 状态: 仅需求草稿,不写代码
---
## 一、需求背景
`BioyondPeptideStation` 当前实验流程在 `start_experiment`manual_confirm 启动调度器)之后即结束,缺少:
1. **等待奔耀回报实验完成**:调度器跑完后,奔耀通过 LIMS 推送 `POST /report/order_finish` 回调;目前 peptide_station 没有把这条推送封装成 action 节点,下游无法在工作流图上等待结果,也拿不到 `usedMaterials` 等下游所需信息。
2. **下料引导**:实验完成后操作员需要把样品/产物从仓位里取出,下料前需要看到每个物料对应的 **仓库 / 位置 / 物料名称 / 数量**;下料完成后还需要回写奔耀(调用 `take-out` 接口),让奔耀清空相应库位状态。
本轮新增三个 action 节点,串在 `start_experiment` 之后:
```text
submit_experiment_dayN
-> start_experiment(manual_confirm 上料)
-> wait_for_order_finish (等回调 + 生成 unloadTable)
-> confirm_unload_materials (manual_confirm 下料确认)
-> take_out_materials (调用 take-out 同步奔耀)
```
---
## 二、关键设计决策
### D1. 本轮只支持单订单,`order_ids` 只做占位
当前不实现多订单等待、乱序回调缓存、并发 wait 隔离。节点输入以 `order_id` 为主,`order_code` 可作为调试兜底。
`order_ids` 可在 handle/返回值中保留为占位字段,但实现只处理第一笔或直接忽略多订单列表。多订单、乱序回调、跨节点重跑复用缓存放到后续迭代。
### D2. `start_experiment` 需要显式透传订单字段
当前 `start_experiment` 只有输入 handles缺少输出 handles如果下游 wait 节点要接在 `start_experiment` 之后,必须让 `start_experiment` 透传:
- `order_id`
- `order_ids`(占位)
- `order_code`
- `resultTable`
实现时给 `start_experiment` 增加对应 `ActionOutputHandle`,并在返回值里保留这些字段。`submit_experiment_dayN``start_experiment` 嵌套字典也应包含 `order_code`,便于工作流编辑器连线。
### D3. `unloadTable` 必须与 `resultTable` 字段一致
`unloadTable` 不新增 `posX/posY/posZ/unit` 列,直接复用现有 `RESULT_TABLE_COLUMNS` 的四列:
```python
RESULT_TABLE_COLUMNS = [
{"name": "设备", "key": "whName"},
{"name": "位置", "key": "locationCode"},
{"name": "物料名称", "key": "materialName"},
{"name": "数量", "key": "quantity"},
]
```
`submit_experiment_dayN` 现有上料确认表 `resultTable` 形状如下;`unloadTable` 也必须保持同一 shape只改 `tableName` 和行数据来源:
```json
{
"data": [
{
"whName": "自动化堆栈",
"locationCode": "A1",
"materialName": "96孔板",
"quantity": "1"
}
],
"columns": [
{"name": "设备", "key": "whName"},
{"name": "位置", "key": "locationCode"},
{"name": "物料名称", "key": "materialName"},
{"name": "数量", "key": "quantity"}
],
"tableName": "resultTable"
}
```
`material-info` 官方 schema 中位置坐标字段是 `locations[].x/y/z`,不是 `posX/posY/posZ`。本轮下料表不展示坐标。
### D4. manual_confirm 只做人确认take-out 放到普通 action
`confirm_unload_materials` 只负责展示下料表、等待操作员确认并透传数据;真正的 `take-out` 调用放到后续普通 action `take_out_materials`
这样更接近 UniLab manual_confirm 的推荐模式manual_confirm 是人机确认检查点,副作用由独立普通 action 执行。
### D5. 本轮不做 unload context 缓存
虽然 workstation architecture 文档支持在 workstation 内存保存 HTTP 报送记录,但本轮暂不实现 `unload_context_cache` 或 order report 缓存。
因此本轮限制如下:
- `wait_for_order_finish` 只等待本次进入节点之后到达的 `/report/order_finish`
- 如果 push 早于 wait 节点到达,本轮不自动补救。
- 如果用户在 `confirm_unload_materials` approve 时忘记勾选,节点失败;当前架构不支持原地重新弹出同一个 manual_confirm也不在本轮实现失败节点重跑。
- 后续若要支持重跑复用,应在 `BioyondPeptideStation` 实例上新增 station runtime 的 `unload_context_cache`,按 `orderCode` 缓存 `unloadTable/material_ids/order_id` 等上下文。
---
## 三、节点 1`wait_for_order_finish`(等推送 + 生成 unloadTable
### 行为
1. 解析单订单目标:
- 首选 `order_id`
- 如果没有 `order_code`,通过 `self.hardware_interface.order_report(order_id)` 取返回数据中的 `code` 作为 `orderCode`
- `order_ids` 仅占位,本轮不实现多订单循环。
2. 设置 `self.last_order_code = order_code``self.last_order_report = None`,并 `self.order_finish_event.clear()`
3. 阻塞在 `self.order_finish_event.wait(timeout=timeout_seconds)` 等 LIMS 推送。
4. peptide_station override `process_order_finish_report(report_request, used_materials)`
- 先调用 `super().process_order_finish_report(...)` 保留父类行为(状态发布、物料同步等)。
-`report_request.data.orderCode == self.last_order_code` 时,把 `report_request.data` 存入 `self.last_order_report`,并 `set()` event。
- 非当前订单推送只记录日志,本轮不缓存。
5. 解除阻塞后解析 `status`
- `"30"` -> `success`
- `"-11"` -> `abnormal_stop`
- `"-12"` -> `manual_stop`
- 其它 -> `unknown_<status>`
- 超时 -> `timeout`
6.`report.usedMaterials[].materialId` 调用 `self.hardware_interface.material_info(material_id)`,带本地函数级 `material_info_cache` 避免重复请求。
7. 组装 `unloadTable``material_ids``preintake_ids``unload_summary` 并作为输出 handles 暴露。
### 入参goal_default
| 参数 | 类型 | 说明 |
|------|------|------|
| `order_id` | `str` | 来自 `start_experiment` 透传输出,必填优先 |
| `order_code` | `str` | 调试兜底;若已知 orderCode 可跳过 `order_report` 反查 |
| `order_ids` | `List[str]` | 占位字段;本轮不实现多订单 |
| `timeout_seconds` | `int` | 默认 `36000`10h |
| `poll_mode` | `bool` | 默认 `False`;如需要可沿用 bioyond_cell 的轮询等待风格 |
### 输出 handles
| key | data_type | 说明 |
|-----|-----------|------|
| `order_finish_status` | `str` | `success` / `abnormal_stop` / `manual_stop` / `timeout` / `unknown_*` |
| `order_finish_report` | `json` | 完整 `report_request.data` |
| `used_materials` | `json` | JSON 化后的 `usedMaterials` 列表 |
| `material_ids` | `json` | 从 `used_materials` 抽出的 `materialId` 列表,可为空 |
| `preintake_ids` | `json` | 本轮默认 `[]`,保留扩展点 |
| `unloadTable` | `table` | 下料表,字段与 `resultTable` 一致 |
| `unload_summary` | `json` | `{ "order_code": ..., "total_items": N, "missing_material_info": [...] }` |
| `order_id` | `str` | 透传给后续节点 |
| `order_code` | `str` | 透传给后续节点 |
| `order_ids` | `json` | 占位透传 |
### `unloadTable` 组装规则
返回结构:
```json
{
"data": [
{
"whName": "自动化堆栈",
"locationCode": "A1",
"materialName": "多肽产物",
"quantity": "10 mg"
}
],
"columns": [
{"name": "设备", "key": "whName"},
{"name": "位置", "key": "locationCode"},
{"name": "物料名称", "key": "materialName"},
{"name": "数量", "key": "quantity"}
],
"tableName": "unloadTable"
}
```
每行字段:
| key | 数据来源 |
|-----|----------|
| `whName` | `material_info.locations` 中匹配 `usedMaterials.locationId` 的 location 的 `whName`;匹配不到取第一条 location 的 `whName`;失败为空串 |
| `locationCode` | 匹配 location 的 `code`;匹配不到取第一条 location 的 `code`;再兜底 `usedMaterials.locationId` |
| `materialName` | `material_info.name`;失败为空串 |
| `quantity` | `usedMaterials.usedQuantity`,若 `material_info.unit` 存在则拼成字符串(如 `"10 mg"` |
`material-info` 失败时不抛异常,对应行尽量保留 `locationCode` / `quantity``whName``materialName` 用空串,并把 `materialId` 放入 `unload_summary.missing_material_info`
### 接口依赖
| 接口 | 调用方式 | 用途 |
|------|----------|------|
| `process_order_finish_report` 钩子 | 基类 HTTP 服务已注册 | 接 LIMS push |
| `POST /api/lims/order/order-report` | `self.hardware_interface.order_report(order_id)` | 从 `order_id` 反查 `orderCode` |
| `POST /api/lims/storage/material-info` | `self.hardware_interface.material_info(material_id)` | 查 `whName/locationCode/materialName/unit` |
#### `order_report(order_id)` API 形式
请求:
```json
{
"apiKey": "B10B5995",
"requestTime": "2026-05-20T10:50:00.123Z",
"data": "<orderId UUID>"
}
```
响应中本节点只依赖 `data.code`
```json
{
"code": 1,
"message": null,
"timestamp": 1779255000000,
"data": {
"id": "<orderId UUID>",
"name": "实验260520-103000",
"code": "EXP260520-103000",
"status": 30,
"statusName": "完成"
}
}
```
`BioyondV1RPC.order_report(order_id)` 已经返回响应中的 `data`,所以实现中应读取 `raw.get("code")`,不是 `raw["data"]["code"]`
#### `material_info(material_id)` API 形式
请求:
```json
{
"apiKey": "B10B5995",
"requestTime": "2026-05-20T10:50:00.123Z",
"data": "<materialId UUID>"
}
```
响应中本节点依赖:
```json
{
"id": "<materialId UUID>",
"name": "多肽产物",
"unit": "mg",
"locations": [
{
"id": "<locationId UUID>",
"whid": "<warehouse UUID>",
"whName": "自动化堆栈",
"code": "A1",
"x": 1,
"y": 1,
"z": 1,
"quantity": 10
}
]
}
```
`BioyondV1RPC.material_info(material_id)` 已经返回响应中的 `data`。字段映射:
| unloadTable key | 来源 |
|-----------------|------|
| `whName` | 匹配 `locationId``locations[].whName` |
| `locationCode` | 匹配 `locationId``locations[].code` |
| `materialName` | `name` |
| `quantity` | `usedMaterials[].usedQuantity` + `unit` |
### 实现要点
- `BioyondPeptideStation.__init__` 末尾追加 `self.order_finish_event = threading.Event()``self.last_order_code = None``self.last_order_report = None`
- 新增 `process_order_finish_report` override`super()`,再做单订单匹配。
- `used_materials` 参数是 `MaterialUsage` dataclass 列表;输出前必须转成 JSON dict。
- `unloadTable` 复用 `RESULT_TABLE_COLUMNS`,不新增 `UNLOAD_TABLE_COLUMNS`
---
## 四、节点 2`confirm_unload_materials`(人工下料确认)
### 行为
1. 接收节点 1 输出的 `order_id` / `order_code` / `material_ids` / `preintake_ids` / `unloadTable`
2. 进入 `NodeType.MANUAL_CONFIRM` 阻塞,操作员根据 `unloadTable` 物理下料。
3. 操作员勾选 `materials_unloaded=True` 并 approve 后,节点函数体继续。
4. 校验 `materials_unloaded == True`
- 为 True返回确认结果并透传 `order_id/material_ids/preintake_ids/unloadTable` 给节点 3。
- 为 False`RuntimeError("下料未确认,拒绝继续 take-out")`
### 入参goal_default
| 参数 | 类型 | 说明 |
|------|------|------|
| `order_id` | `str` | 来自节点 1必填 |
| `order_code` | `str` | 来自节点 1日志/排错用 |
| `material_ids` | `List[str]` | 来自节点 1可为空 |
| `preintake_ids` | `List[str]` | 来自节点 1默认 `[]` |
| `unloadTable` | `table` | 来自节点 1供人工确认展示 |
| `materials_unloaded` | `bool` | manual_confirm 勾选字段,默认 `False` |
| `timeout_seconds` | `int` | 默认 `3600` |
| `assignee_user_ids` | `List[str]` | `placeholder_keys={"assignee_user_ids": "unilabos_manual_confirm"}` |
### 输出 handles
| key | data_type | 说明 |
|-----|-----------|------|
| `unload_confirmed` | `bool` | 是否已人工确认下料 |
| `order_id` | `str` | 透传 |
| `order_code` | `str` | 透传 |
| `material_ids` | `json` | 透传 |
| `preintake_ids` | `json` | 透传 |
| `unloadTable` | `table` | 透传 |
### 实现要点
- 装饰器使用 `node_type=NodeType.MANUAL_CONFIRM`
- `always_free=True``placeholder_keys``feedback_interval=300` 与现有 `start_experiment` 保持一致。
- 本节点不调用 `take_out`,只做确认与透传。
- 忘记勾选后不会自动重新显示下料指引;本轮不实现缓存或失败节点原地重跑。
---
## 五、节点 3`take_out_materials`(调用 take-out 同步奔耀)
### 行为
1. 接收节点 2 透传的 `order_id` / `material_ids` / `preintake_ids`
2. 校验 `order_id` 非空。
3. 调用 `self.hardware_interface.take_out(order_id, preintake_ids=preintake_ids, material_ids=material_ids)`
4. 返回 `take_out_result``unloaded_count``success`
### 入参
| 参数 | 类型 | 说明 |
|------|------|------|
| `order_id` | `str` | 必填 |
| `material_ids` | `List[str]` | 可为空;为空时由奔耀按 `orderId` 处理的能力以后现场确认 |
| `preintake_ids` | `List[str]` | 可为空,默认 `[]` |
| `order_code` | `str` | 日志/排错用 |
### 输出 handles
| key | data_type | 说明 |
|-----|-----------|------|
| `take_out_result` | `json` | `take-out` 原始响应 `{code, message, timestamp, data}` |
| `unloaded_count` | `int` | `len(material_ids)` |
| `success` | `bool` | `take_out_result.code == 1``data` 不为 False |
### 接口依赖
| 接口 | 调用方式 | 用途 |
|------|----------|------|
| `POST /api/lims/order/take-out` | `self.hardware_interface.take_out(order_id, preintake_ids, material_ids)` | 通知奔耀同步取出 |
请求体 schemahelper script 已核对):
```json
{
"apiKey": "B10B5995",
"requestTime": "2026-05-20T10:50:00.123Z",
"data": {
"orderId": "<UUID>",
"preintakeIds": [],
"materialIds": ["<UUID-1>", "<UUID-2>"]
}
}
```
响应 schema
```json
{
"code": 1,
"message": null,
"timestamp": 1779255000000,
"data": true
}
```
`BioyondV1RPC.take_out(...)` 返回完整响应包,因此 `take_out_materials` 应保留原始包到 `take_out_result`
### 实现要点
- 本轮不修改 `sample_waste_removal`,保持 backward compatibility。
- 新节点只调用现有完整能力的 `take_out(...)`
- `preintake_ids` / `material_ids` 都按可选列表处理,默认 `[]`
- `take-out` 返回 `code != 1` 时返回 `success=False` 并记录 warning是否抛异常留作开放问题。
---
## 六、端到端工作流连线
```mermaid
flowchart LR
submit["submit_experiment_dayN"] --> start["start_experiment<br/>manual_confirm: 上料"]
start -->|order_id, order_code| wait["wait_for_order_finish<br/>等 order_finish + 生成 unloadTable"]
wait -->|order_id, material_ids,<br/>preintake_ids, unloadTable| confirm["confirm_unload_materials<br/>manual_confirm: 操作员下料确认"]
confirm -->|order_id, material_ids,<br/>preintake_ids| takeout["take_out_materials<br/>调用 take-out 同步奔耀"]
bioyond["奔耀 LIMS"] -.HTTP POST /report/order_finish.-> wait
wait -.material-info.-> bioyond
takeout -.take-out.-> bioyond
```
---
## 七、影响面与兼容性
- **`peptide_station.py`**
- 修改 `start_experiment`:增加 `order_id/order_code/order_ids/resultTable` 输出 handles并在返回值透传。
- 增加 `wait_for_order_finish``confirm_unload_materials``take_out_materials` 三个 action。
- 增加 `process_order_finish_report` override。
- 增加 `_build_unload_table(...)` 等私有辅助方法。
- **`bioyond_rpc.py` 不动**
- `take_out` 已有完整 schema 能力。
- `sample_waste_removal` 本轮不改,保持兼容。
- **基类 `station.py` 不动**
- override 中保留 `super().process_order_finish_report(...)` 调用。
- **HTTP 服务不动**
- `WorkstationHTTPService` 已支持 `/report/order_finish`
- **本轮不做缓存**
- 不新增 `unload_context_cache`
- 不支持 push 早于 wait 的自动补救。
- 不支持失败 manual_confirm 原地重跑。
- **测试**:补在现有路径 `unilabos/devices/workstation/bioyond_studio/peptide_station/tests/test_peptide_station_contracts.py`
1. `start_experiment` 输出 handles/返回值透传 `order_id/order_code/order_ids/resultTable`
2. `process_order_finish_report` orderCode 匹配 / 不匹配时 event 是否触发。
3. `wait_for_order_finish` 单订单成功、超时、状态映射、`used_materials` JSON 化。
4. `_build_unload_table` 列顺序严格等于 `RESULT_TABLE_COLUMNS`,且无 `posX/posY/posZ/unit` 列。
5. `material-info` 失败时不抛异常,`missing_material_info` 正确记录。
6. `confirm_unload_materials` 未勾选时报错,勾选后透传下游字段且不调用 `take_out`
7. `take_out_materials` 调用 `hardware_interface.take_out(order_id, preintake_ids, material_ids)`,不调用 `sample_waste_removal`
---
## 八、待人类确认的开放问题
1. **过滤产物 vs 全量**`usedMaterials` 同时包含试剂、耗材、样品(`typeMode` 区分),下料表是否需要默认排除试剂/耗材?当前默认全量列出。
2. **take-out 失败是否阻塞工作流**:本计划暂定返回 `success=False` 并 warning不抛异常如果希望奔耀仓位状态必须一致可改为抛 `RuntimeError`
3. **后续缓存/重跑能力**:如果要支持 push 早到、忘勾选后重跑复用 `unloadTable`,后续应在 `BioyondPeptideStation` station runtime 上实现 `unload_context_cache`,但本轮不做。
4. **多订单**:本轮只保留 `order_ids` 占位,不实现多订单等待、乱序回调或并发 wait。
---
## 附录 AAPI schema 核对摘要
使用 `temp_benyao/scripts/api_helper.py --root temp_benyao/peptide` 核对:
### A.1 `POST /api/lims/storage/material-info`
请求:
```json
{
"apiKey": "B10B5995",
"requestTime": "2026-05-20T10:50:00.123Z",
"data": "<materialId UUID>"
}
```
响应关键字段:
```json
{
"id": "<materialId UUID>",
"typeName": "样品",
"code": "MAT-001",
"barCode": "BC-001",
"name": "多肽产物",
"quantity": 10,
"lockQuantity": 0,
"unit": "mg",
"status": 1,
"isUse": true,
"locations": [
{
"id": "<locationId UUID>",
"whid": "<warehouse UUID>",
"whName": "自动化堆栈",
"code": "A1",
"x": 1,
"y": 1,
"z": 1,
"quantity": 10
}
],
"detail": []
}
```
注意schema 没有 `posX/posY/posZ`,本轮也不展示坐标。
### A.2 `POST /api/lims/order/take-out`
请求:
```json
{
"apiKey": "B10B5995",
"requestTime": "2026-05-20T10:50:00.123Z",
"data": {
"orderId": "<orderId UUID>",
"preintakeIds": [],
"materialIds": ["<materialId UUID>"]
}
}
```
响应:
```json
{
"code": 1,
"message": null,
"timestamp": 1779255000000,
"data": true
}
```
源码中已有 `BioyondV1RPC.take_out(order_id, preintake_ids=None, material_ids=None)`,本轮复用它。
### A.3 `/report/order_finish`
`/report/order_finish` 不在 Peptide JSON OpenAPI specs 中schema 依据:
- `unilabos/devices/workstation/workstation_http_service.py`
- `temp_benyao/peptide/docs/reference/api_manual.md`
关键字段:
```json
{
"token": "token-from-lims",
"request_time": "2026-05-20 10:50:00.123",
"data": {
"orderCode": "EXP260520-103000",
"orderName": "实验260520-103000",
"startTime": "2026-05-20 09:00:00",
"endTime": "2026-05-20 10:50:00",
"status": "30",
"usedMaterials": [
{
"materialId": "<materialId UUID>",
"locationId": "<locationId UUID>",
"typeMode": "1",
"usedQuantity": 10
}
]
}
}
```
`WorkstationHTTPService` 会把 `usedMaterials[]` 转成 `MaterialUsage` dataclass 列表传给 `process_order_finish_report(report_request, used_materials)`peptide 输出 `used_materials` handle 前需要转回 JSON dict
```json
[
{
"materialId": "<materialId UUID>",
"locationId": "<locationId UUID>",
"typeMode": "1",
"usedQuantity": 10
}
]
```
---
## 附录 B本轮不实现的内容
- 不做 station runtime 的 `unload_context_cache`
- 不做多订单。
- 不做 push 早到后的补救。
- 不做 failed manual_confirm 原地重跑。
- 不改前端。
- 不改 `sample_waste_removal`

View File

@@ -0,0 +1,461 @@
# Peptide Four-Checkbox Reset Plan
Date: 2026-05-21 16:30
Status: Proposal only / not executed
## Scope
This plan replaces `2026-05-21_1556_peptide_reset_sirna_reference_plan.md` for Peptide reset work.
User direction captured here:
- `take_out` is unnecessary for Peptide reset.
- Do not add a material-cache refresh checkbox.
- Change reset to four checkbox-controlled operations:
- 调度器复位
- 订单状态复位
- 库位复位
- 仪器复位
- The first three checkboxes default to checked.
- The fourth checkbox, 仪器复位 / `reset_devices`, defaults to unchecked.
- Replace the current public `reset` action with:
- `reset_auto`: normal ILab action node. This is the renamed/replaced version of the current reset implementation.
- `reset_manual`: manual-confirm action node with a physical cleanup confirmation message.
## Evidence Summary
Current Peptide source:
- Reset action code is currently in `unilabos/devices/workstation/bioyond_studio/peptide_station/peptide_station.py`.
- Current Peptide reset selects `scheduler_reset`, `reset_order_status`, and `reset_location`, and passes ids to order/location resets.
- `BioyondV1RPC.reset_devices()` already calls `/api/lims/device/reset-devices` with only `apiKey` and `requestTime`.
- `BioyondV1RPC.scheduler_reset()` already calls `/api/lims/scheduler/reset` with only `apiKey` and `requestTime`.
- `BioyondV1RPC.reset_order_status(order_id)` and `reset_location(location_id)` currently send `data`, but live probes showed that omitted `data` succeeds.
Live Peptide no-data reset probes using `temp_benyao/peptide/peptide_station_config.example.json`:
- `POST /api/lims/order/reset-order-status` with request keys `["apiKey", "requestTime"]` returned HTTP 200 and `code=1`.
- `POST /api/lims/scheduler/reset` with request keys `["apiKey", "requestTime"]` returned HTTP 200 and `code=1`.
- `POST /api/lims/storage/reset-location` with request keys `["apiKey", "requestTime"]` returned HTTP 200 and `code=1`.
- `reset-devices` was not live-probed in this session, but the current RPC wrapper already sends no `data`.
Raw findings:
- `temp_benyao/peptide/_findings/2026-05-21_1613_reset_order_status_no_data_live.md`
- `temp_benyao/peptide/_findings/2026-05-21_1615_remaining_resets_no_data_live.md`
## Proposed Public Actions
### `reset_auto`
Normal action node. This is the auto/no-manual-confirm path. It replaces the current public `reset` action; do not leave a second public `reset` action unless a later compatibility request explicitly asks for an alias.
Checkbox schema rule:
- Use plain `bool` annotations in the action signature.
- Do not use `Annotated[bool, Field(...)]` for these checkbox params in this implementation plan.
- The current AST registry schema path does not unwrap `Annotated[...]`; plain `bool` is required so generated JSON Schema marks the fields as boolean and the renderer can show checkboxes.
- Put human-facing labels/descriptions in the method docstring or action description. If field-level `Field(description=...)` metadata is required later, add registry `Annotated` support and a schema test as a separate change.
Decorator shape:
```python
@action(
always_free=True,
goal_default={
"reset_scheduler": True,
"reset_order_status": True,
"reset_location": True,
"reset_devices": False,
},
description="自动复位调度器/订单状态/库位,可选仪器复位",
)
def reset_auto(
self,
reset_scheduler: bool = True,
reset_order_status: bool = True,
reset_location: bool = True,
reset_devices: bool = False,
**kwargs: Any,
) -> Dict[str, Any]:
"""自动复位调度器/订单状态/库位,可选仪器复位。
Args:
reset_scheduler[调度器复位]: 调用 /api/lims/scheduler/reset默认勾选。
reset_order_status[订单状态复位]: 调用 /api/lims/order/reset-order-status默认勾选。
reset_location[库位复位]: 调用 /api/lims/storage/reset-location默认勾选。
reset_devices[仪器复位]: 调用 /api/lims/device/reset-devices默认不勾选。
"""
...
```
Implementation notes:
- Use real plain-`bool` parameters, not hidden `**kwargs` and not `Annotated`, so the action renderer can expose four checkboxes.
- Rename/replace the existing `reset` action as `reset_auto`; the implementation should not keep the old id-shaped `reset` action as another public path by default.
- Keep the three routine reset defaults checked.
- Keep `reset_devices` unchecked because it can be broader and more disruptive.
- Do not require or resolve order ids or location ids.
- Do not call `take_out`.
- Do not call `refresh_material_cache`.
### `reset_manual`
Manual-confirm node. It should show the operator a physical cleanup warning, then execute the same reset helper as `reset_auto` after the operator confirms.
Actual manual-confirm decorator pattern in this repo:
- Use `@action(node_type=NodeType.MANUAL_CONFIRM)`.
- Set `always_free=True`.
- Add `placeholder_keys={"assignee_user_ids": "unilabos_manual_confirm"}`.
- Include `timeout_seconds: int` and `assignee_user_ids: list[str]`.
- Add `goal_default` for `timeout_seconds` and `assignee_user_ids`.
- Manual-confirm actions are normally side-effect-light, but existing Peptide `start_experiment` is already a `MANUAL_CONFIRM` action that performs scheduler start after the operator gate, so a reset-after-confirm pattern is compatible with current Peptide style.
Proposed confirmation text:
```text
请确认G3、CEM、Tecan、撕膜机、封膜机、打标机、旋转堆栈上下料位、3个转台等位置的物料已清理完毕
请开门检查冰箱、IDOT、酶标仪、离心机、LCMS内部没有遗留物料。
```
Decorator/function shape:
```python
RESET_MANUAL_CONFIRM_MESSAGE = (
"请确认G3、CEM、Tecan、撕膜机、封膜机、打标机、旋转堆栈上下料位、3个转台等位置的物料已清理完毕\n"
"请开门检查冰箱、IDOT、酶标仪、离心机、LCMS内部没有遗留物料。"
)
@action(
always_free=True,
node_type=NodeType.MANUAL_CONFIRM,
placeholder_keys={"assignee_user_ids": "unilabos_manual_confirm"},
goal_default={
"reset_scheduler": True,
"reset_order_status": True,
"reset_location": True,
"reset_devices": False,
"physical_cleanup_confirmed": False,
"timeout_seconds": 3600,
"assignee_user_ids": [],
},
feedback_interval=300,
description=RESET_MANUAL_CONFIRM_MESSAGE,
)
def reset_manual(
self,
reset_scheduler: bool = True,
reset_order_status: bool = True,
reset_location: bool = True,
reset_devices: bool = False,
physical_cleanup_confirmed: bool = False,
timeout_seconds: int = 3600,
assignee_user_ids: Optional[List[str]] = None,
**kwargs: Any,
) -> Dict[str, Any]:
"""人工确认物理清理后执行复位。
Args:
reset_scheduler[调度器复位]: 调用 /api/lims/scheduler/reset默认勾选。
reset_order_status[订单状态复位]: 调用 /api/lims/order/reset-order-status默认勾选。
reset_location[库位复位]: 调用 /api/lims/storage/reset-location默认勾选。
reset_devices[仪器复位]: 调用 /api/lims/device/reset-devices默认不勾选。
physical_cleanup_confirmed[物理清理确认]: 确认清理提示中的物料检查已经完成,默认不勾选。
"""
...
```
Execution rule:
- If `physical_cleanup_confirmed` is false, return a blocked result and do not call any reset API.
- If it is true, call the same internal helper as `reset_auto`.
- Return `confirmation_message` in the result payload so call logs preserve the exact operator instruction text.
Renderer caveat:
- `description` should carry the warning in generated action metadata.
- `physical_cleanup_confirmed` must remain a plain `bool` so it renders as a checkbox.
- The cleanup warning should be carried by the action `description` and the docstring param description. Do not rely on `Field(description=...)` unless registry `Annotated` support has been implemented and tested.
- If the current frontend does not show action descriptions or docstring field descriptions reliably, add a read-only string parameter such as `confirmation_message: str = RESET_MANUAL_CONFIRM_MESSAGE` with `goal_default`, or use a handle-based display only after renderer behavior is verified.
## Shared Internal Helper
Both public actions should delegate to one helper, for example:
```python
def _execute_reset_operations(
self,
*,
reset_scheduler: bool,
reset_order_status: bool,
reset_location: bool,
reset_devices: bool,
) -> Dict[str, Any]:
...
```
Call order:
1. `scheduler_reset`
2. `reset_order_status`
3. `reset_location`
4. `reset_devices`
Result shape:
```python
{
"selected_operations": [
{"key": "reset_scheduler", "label": "调度器复位", "selected": True},
{"key": "reset_order_status", "label": "订单状态复位", "selected": True},
{"key": "reset_location", "label": "库位复位", "selected": True},
{"key": "reset_devices", "label": "仪器复位", "selected": False},
],
"executed_calls": [
{"operation": "scheduler_reset", "endpoint": "/api/lims/scheduler/reset", "result": {"code": 1}},
],
"skipped_operations": [
{"operation": "reset_devices", "reason": "checkbox_disabled"},
],
"warnings": [],
}
```
Failure handling:
- Execute selected operations sequentially and record each result.
- If an operation returns non-`1` code, add a warning and continue unless the caller later requests fail-fast.
- If an RPC method raises, catch it, record an error entry, and continue to the next selected operation unless fail-fast is introduced.
## RPC Wrapper Adjustment
Adjust the two id-shaped wrappers to no-data calls:
- `BioyondV1RPC.reset_order_status()` should no longer require `order_id`.
- `BioyondV1RPC.reset_location()` should no longer require `location_id`.
Current no-data wrappers already exist:
- `scheduler_reset()`
- `reset_devices()`
Suggested RPC signatures:
```python
def scheduler_reset(self) -> int: ...
def reset_order_status(self) -> int: ...
def reset_location(self) -> int: ...
def reset_devices(self) -> int: ...
```
Compatibility option:
```python
def reset_order_status(self, order_id: Optional[str] = None) -> int:
del order_id
...
def reset_location(self, location_id: Optional[str] = None) -> int:
del location_id
...
```
This keeps older code from crashing while making the actual wire request no-data.
## Adjusted Runtime API Schemas
These are the schemas Peptide reset code should target at runtime after the live no-data probes. They intentionally omit `data`, even though OpenAPI models nullable `data` for these endpoints.
All four requests use:
```json
{
"apiKey": "string",
"requestTime": "date-time"
}
```
No `data` field should be sent by default.
All four responses use:
```json
{
"code": 1,
"message": "",
"timestamp": 0
}
```
### 调度器复位
Endpoint:
```text
POST /api/lims/scheduler/reset
```
Adjusted request:
```json
{
"apiKey": "B10B5995",
"requestTime": "2026-05-21T08:15:16.494Z"
}
```
Live response:
```json
{
"code": 1,
"message": "",
"timestamp": 1779351316072
}
```
Notes:
- OpenAPI says `data` is nullable int32.
- Live Peptide accepted omitted `data`.
### 订单状态复位
Endpoint:
```text
POST /api/lims/order/reset-order-status
```
Adjusted request:
```json
{
"apiKey": "B10B5995",
"requestTime": "2026-05-21T08:13:34.750Z"
}
```
Live response:
```json
{
"code": 1,
"message": "",
"timestamp": 1779351214422
}
```
Notes:
- OpenAPI says `data` is nullable string.
- Live Peptide accepted omitted `data`.
- Do not model this as order-id scoped unless Bioyond confirms backend behavior.
### 库位复位
Endpoint:
```text
POST /api/lims/storage/reset-location
```
Adjusted request:
```json
{
"apiKey": "B10B5995",
"requestTime": "2026-05-21T08:15:18.924Z"
}
```
Live response:
```json
{
"code": 1,
"message": "",
"timestamp": 1779351318565
}
```
Notes:
- OpenAPI says `data` is nullable string.
- Live Peptide accepted omitted `data`.
- Do not model this as location-id scoped unless Bioyond confirms backend behavior.
### 仪器复位
Endpoint:
```text
POST /api/lims/device/reset-devices
```
Adjusted request:
```json
{
"apiKey": "B10B5995",
"requestTime": "date-time"
}
```
Expected response shape:
```json
{
"code": 1,
"message": "",
"timestamp": 0
}
```
Notes:
- OpenAPI says `data` is nullable string.
- Current `BioyondV1RPC.reset_devices()` already sends no `data`.
- This endpoint was not live-probed in the no-data reset session.
- Keep checkbox default unchecked.
## Tests To Add Before Implementation
1. `reset_auto` is not `NodeType.MANUAL_CONFIRM`.
2. `reset_manual` has `node_type=NodeType.MANUAL_CONFIRM`.
3. `reset_manual` metadata includes:
- `always_free=True`
- `placeholder_keys={"assignee_user_ids": "unilabos_manual_confirm"}`
- `timeout_seconds=3600`
- `assignee_user_ids=[]`
- `physical_cleanup_confirmed=False`
4. Both reset actions expose four real boolean params:
- `reset_scheduler`
- `reset_order_status`
- `reset_location`
- `reset_devices`
5. The generated registry schema marks those reset params as JSON Schema `type: boolean`, not `object` or `string`, so the frontend can render checkboxes.
6. `reset_auto` replaces the current public `reset` action. Unless a later compatibility request adds an alias, no old id-shaped public `reset` action remains.
7. Goal defaults are:
- first three reset checkboxes `True`
- `reset_devices=False`
8. `reset_manual(..., physical_cleanup_confirmed=False)` does not call any RPC reset method.
9. `reset_auto()` with defaults calls:
- `scheduler_reset()`
- `reset_order_status()`
- `reset_location()`
- not `reset_devices()`
10. `reset_auto(reset_devices=True)` also calls `reset_devices()`.
11. `reset_order_status()` and `reset_location()` RPC wrappers send no `data` key.
12. No reset path calls `take_out`.
13. No reset path calls `refresh_material_cache`.
## Non-Goals
- Do not implement `take_out` in reset.
- Do not refresh `material_cache` from reset.
- Do not resolve order ids or location ids for reset.
- Do not add Project/cache/browser cleanup routes.
- Do not make `reset_devices` default-on.
- Do not execute this plan during planning.

View File

@@ -779,6 +779,49 @@ class BioyondV1RPC(BaseRequest):
return response.get("data", {})
def take_out(
self,
order_id: str,
preintake_ids: list[str] | None = None,
material_ids: list[str] | None = None,
) -> dict:
"""取出订单关联通量/物料
参数:
order_id: 订单ID
preintake_ids: 通量ID列表可为空
material_ids: 物料ID列表可为空
返回值:
dict: 服务端响应包,失败返回空字典
"""
if not order_id:
self._logger.error("取出订单关联通量/物料错误: 缺少订单ID")
return {}
params = {
"orderId": order_id,
"preintakeIds": list(preintake_ids or []),
"materialIds": list(material_ids or []),
}
response = self.post(
url=f'{self.host}/api/lims/order/take-out',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": params,
})
if not response:
return {}
if response['code'] != 1:
self._logger.error(f"取出订单关联通量/物料错误: {response.get('message', '')}")
return response
return response
def cancel_order(self, json_str: str) -> bool:
"""取消指定任务

View File

@@ -8,7 +8,7 @@ returned. Outside of an active session the wrapped method delegates to the
original (unwrapped) implementation, leaving non-debug behavior intact.
The session writes a Markdown file under ``out_dir`` mirroring the format of
``temp_benyao/peptide/_logs/2026-04-30_160316_day3_samplefile_only_raw_calls.md``
``bioyond_debug_records/2026-04-30_160316_day3_samplefile_only_raw_calls.md``
minus the "Raw Payload Argument" section.
This module has no dependency on ``BioyondV1RPC`` itself; the only contract is

View File

@@ -0,0 +1,3 @@
from .peptide_station import BioyondPeptideStation, fetch_workflow_list, load_peptide_config
__all__ = ["BioyondPeptideStation", "fetch_workflow_list", "load_peptide_config"]

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,681 @@
"""多肽站 AST/参数/结果表 离线契约测试。"""
from __future__ import annotations
import importlib
import inspect
import json
import sys
from pathlib import Path
from typing import Any, Dict, List
from unittest.mock import MagicMock
import pytest
REPO_ROOT = Path(__file__).resolve().parents[6]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
MODULE_PATH = "unilabos.devices.workstation.bioyond_studio.peptide_station.peptide_station"
CLASS_NAME = "BioyondPeptideStation"
ORDER_GUID = "3a20eabe-bad5-ef95-49bd-7ffbd5df189d"
CREATE_ALLOCATION = {
ORDER_GUID: [
{
"materialId": "mat-tip",
"materialName": "200μL枪头盒",
"materialCode": "0008-00105",
"quantity": "1个",
"materialTypeMode": "Consumables",
"locationCode": "1-01",
"locationShowName": "1-01",
},
{
"materialId": "mat-plate",
"materialName": "96孔板",
"materialCode": "PLATE-96",
"quantity": "1",
"materialTypeMode": "Sample",
"locationCode": "A1",
"locationShowName": "A1-show",
},
{
"materialId": "mat-extra",
"materialName": "未知耗材",
"materialCode": "X-1",
"quantity": "2",
"materialTypeMode": "Future",
"locationCode": "Z9",
"locationShowName": "",
},
]
}
FLATTENED_LIVE = [
{"step": "39c78d4b-b5d3-f721-2001-9d52000084c3", "step_name": "S1", "Key": "SampleFile", "m": 0, "n": 0, "Value": "", "DisplayValue": "", "TaskDisplayable": 1},
{"step": "39c78d4b-b5d3-f721-2001-9d52000084c3", "step_name": "S1", "Key": "Example", "m": 0, "n": 0, "Value": "x", "DisplayValue": "x", "TaskDisplayable": 1},
{"step": "39c78d4b-b5d3-f721-2001-9d52000084c4", "step_name": "S2", "Key": "protocol", "m": 14, "n": 28, "Value": "", "DisplayValue": "", "TaskDisplayable": 1},
{"step": "39c78d4b-b5d3-f721-2001-9d52000084c5", "step_name": "S3", "Key": "CEMMethodFileName", "m": 0, "n": 0, "Value": "", "DisplayValue": "", "TaskDisplayable": 1},
]
def _import_module() -> Any:
return importlib.import_module(MODULE_PATH)
def _make_station() -> Any:
module = _import_module()
cls = getattr(module, CLASS_NAME)
station = object.__new__(cls)
station.bioyond_config = {"api_host": "http://test", "api_key": "k", "warehouse_mapping": {}}
rpc = MagicMock()
rpc.host = "http://test"
rpc.api_key = "k"
rpc.material_info.return_value = {"locations": [{"whName": "自动化堆栈", "code": "1-01"}]}
station.hardware_interface = rpc
return station
# ---------------------------------------------------------------------------
# 1. AST/导入面
# ---------------------------------------------------------------------------
def test_required_actions_exposed() -> None:
cls = getattr(_import_module(), CLASS_NAME)
required = {
"upload_sample_excel",
"list_sample_excels",
"get_step_parameters",
"submit_experiment",
"submit_experiment_day1",
"submit_experiment_day2",
"submit_experiment_day3",
"submit_experiment_day4",
"submit_experiment_day4_LCMS",
"start_experiment",
"reset",
"scheduler_start",
"scheduler_stop",
"scheduler_pause",
"scheduler_continue",
"get_order_list",
"get_order_report",
"get_aggregated_order_report",
}
have = {name for name, _ in inspect.getmembers(cls, inspect.isfunction)}
missing = sorted(required - have)
assert not missing, f"缺少动作: {missing}"
def test_manual_confirm_node_types() -> None:
module = _import_module()
cls = getattr(module, CLASS_NAME)
manual = {"submit_experiment_day1", "start_experiment"}
normal = {
"submit_experiment",
"submit_experiment_day2",
"submit_experiment_day3",
"submit_experiment_day4",
"submit_experiment_day4_LCMS",
"reset",
"scheduler_start",
"list_sample_excels",
"get_step_parameters",
"get_order_list",
"get_order_report",
}
for name in manual:
meta = getattr(getattr(cls, name), "_action_registry_meta", {})
assert meta.get("node_type") == module.NodeType.MANUAL_CONFIRM, name
for name in normal:
meta = getattr(getattr(cls, name), "_action_registry_meta", {})
assert meta.get("node_type") != module.NodeType.MANUAL_CONFIRM, name
def test_submit_and_reset_signatures_exclude_legacy_manual_confirm() -> None:
cls = getattr(_import_module(), CLASS_NAME)
for name in (
"submit_experiment",
"submit_experiment_day2",
"submit_experiment_day3",
"submit_experiment_day4",
"submit_experiment_day4_LCMS",
"reset",
):
params = inspect.signature(getattr(cls, name)).parameters
assert "timeout_seconds" not in params, name
assert "assignee_user_ids" not in params, name
def test_day1_submit_accepts_manual_confirm_kwargs() -> None:
"""plan: Day1 是 MANUAL_CONFIRM框架会注入 timeout_seconds/assignee_user_ids函数必须能接收。"""
cls = getattr(_import_module(), CLASS_NAME)
sig = inspect.signature(cls.submit_experiment_day1)
has_kwargs = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values())
assert has_kwargs, "submit_experiment_day1 必须有 **kwargs 以容纳人工确认框架字段"
def test_typed_dicts_present() -> None:
module = _import_module()
for cls_name in (
"PeptideGenericSubmitRequiredParams",
"PeptideGenericSubmitOptionalParams",
"PeptideDay1RequiredParams",
"PeptideDay1OptionalParams",
"PeptideDay2RequiredParams",
"PeptideDay2OptionalParams",
"PeptideDay3RequiredParams",
"PeptideDay3OptionalParams",
"PeptideDay4RequiredParams",
"PeptideDay4OptionalParams",
"PeptideDay4LCMSRequiredParams",
"PeptideDay4LCMSOptionalParams",
):
assert hasattr(module, cls_name), cls_name
def test_workflow_constants_split() -> None:
module = _import_module()
assert module.DAY4_PEPTIDE_WORKFLOW_NAME == "Day4环肽酰化-酶标"
assert module.DAY4_LCMS_PEPTIDE_WORKFLOW_NAME == "Day4环肽酰化-酶标+LCMS"
assert module.DAY_WORKFLOW_BINDINGS["day4_lcms"]["sub_name"] == "Day4环肽酰化-酶标LCMS"
assert module.DAY1_CEM_METHOD_DEFAULT == "5microdouble-20250911.MPM"
# ---------------------------------------------------------------------------
# 2. Sample Excel
# ---------------------------------------------------------------------------
def test_list_sample_excels_modes() -> None:
station = _make_station()
records = [
{"fileName": "DPR019-a.xlsx", "relativePath": "upload\\sample\\DPR019-a.xlsx"},
{"fileName": "DPR019-b.xlsx", "relativePath": "upload\\sample\\DPR019-b.xlsx"},
]
station._list_sample_excels = MagicMock(return_value=records) # type: ignore[method-assign]
info = station.list_sample_excels(sample_excel_pattern="DPR019-a", deterministic_resolve=False)
assert "sample_excel_data" in info
assert "sample_excel_relative_path" not in info
resolved = station.list_sample_excels(sample_excel_pattern="DPR019-a", deterministic_resolve=True)
assert resolved["sample_excel_relative_path"] == "upload\\sample\\DPR019-a.xlsx"
with pytest.raises(Exception):
station.list_sample_excels(sample_excel_pattern="DPR019", deterministic_resolve=True)
def test_resolve_submit_sample_file_direct_path() -> None:
station = _make_station()
relative, selected = station._resolve_submit_sample_file({}, {}, "upload/sample/x.xlsx")
assert relative == "upload\\sample\\x.xlsx"
assert selected["fileName"] == "x.xlsx"
def test_filename_matches_pattern_substring_and_glob() -> None:
station = _make_station()
assert station._filename_matches_pattern("DPR019-20260421-thrombin-5.xlsx", "DPR019")
assert station._filename_matches_pattern("a.xlsx", "*.xlsx")
assert not station._filename_matches_pattern("a.xlsx", "*.docx")
assert station._filename_matches_pattern("a.xlsx", "")
# ---------------------------------------------------------------------------
# 3. Step parameter helper
# ---------------------------------------------------------------------------
def test_filter_step_parameters_preserves_zero_and_skips_unknown() -> None:
station = _make_station()
records = [
{"TaskDisplayable": 1, "Value": 0, "DisplayValue": ""},
{"TaskDisplayable": 1, "Value": "", "DisplayValue": ""},
{"TaskDisplayable": 0, "Value": "", "DisplayValue": ""},
{"TaskDisplayable": None, "Value": "", "DisplayValue": ""},
]
filtered = station._filter_step_parameter_records(records, True, True, True)
assert {(r.get("Value"), r.get("TaskDisplayable")) for r in filtered} == {(0, 1), ("", 1), ("", 0)}
def test_get_step_parameters_zero_match_returns_status() -> None:
station = _make_station()
station._query_workflow_records = MagicMock(return_value=[]) # type: ignore[method-assign]
out = station.get_step_parameters(workflow_name_filter="不存在")
status = out["step_parameters_raw_json"]
assert status.get("code") == -1
assert out["filtered_subworkflows"] == []
def test_get_step_parameters_multi_match_returns_status() -> None:
station = _make_station()
station._query_workflow_records = MagicMock(return_value=[ # type: ignore[method-assign]
{"workflowId": "w1", "workflowName": "A", "subworkflowId": "s1", "subworkflowName": "A1"},
{"workflowId": "w1", "workflowName": "A", "subworkflowId": "s2", "subworkflowName": "A2"},
])
out = station.get_step_parameters(workflow_name_filter="A")
assert out["step_parameters_raw_json"].get("code") == 0
assert len(out["filtered_subworkflows"]) == 2
def test_get_step_parameters_direct_sub_workflow_id() -> None:
station = _make_station()
station._query_step_parameters = MagicMock(return_value={ # type: ignore[method-assign]
"39c78d4b-b5d3-f721-2001-9d52000084c3": [
{"name": "S1", "m": 0, "n": 0, "parameterList": [
{"Key": "SampleFile", "TaskDisplayable": 1, "Value": "", "DisplayValue": ""},
]},
]
})
out = station.get_step_parameters(sub_workflow_id="39c78d4b-b5d3-f721-2001-9d52000084c3")
augmented = out["step_parameters_raw_json"]
assert augmented["code"] == 1
assert any(p["Key"] == "SampleFile" for p in augmented["data"]["filteredParameters"])
# ---------------------------------------------------------------------------
# 4. Partial parameter entries + live resolution
# ---------------------------------------------------------------------------
def test_partial_entries_inject_samplefile_and_overrides() -> None:
station = _make_station()
entries, warnings = station._build_partial_parameter_entries(
sample_excel_relative_path="upload\\sample\\f.xlsx",
day_key="day2",
parameter_overrides=[{"Key": "Example", "Value": 0}],
)
assert entries[0] == {"Key": "SampleFile", "Value": "upload\\sample\\f.xlsx"}
assert any(e["Key"] == "Example" and e["Value"] == 0 for e in entries)
assert warnings == []
def test_day1_partial_entries_inject_cem_default() -> None:
station = _make_station()
entries, _ = station._build_partial_parameter_entries(
sample_excel_relative_path="upload\\sample\\f.xlsx",
day_key="day1",
extra_autofill=[{"Key": "CEMMethodFileName", "Value": "5microdouble-20250911.MPM"}],
)
assert any(e["Key"] == "CEMMethodFileName" and e["Value"] == "5microdouble-20250911.MPM" for e in entries)
def test_overrides_duplicate_last_write_wins_warning() -> None:
station = _make_station()
entries, warnings = station._build_partial_parameter_entries(
sample_excel_relative_path="x",
day_key="day2",
parameter_overrides=[
{"Key": "Example", "m": 0, "n": 0, "Value": "first"},
{"Key": "Example", "m": 0, "n": 0, "Value": "second"},
],
)
example_entries = [e for e in entries if e["Key"] == "Example"]
assert len(example_entries) == 1
assert example_entries[0]["Value"] == "second"
assert any("重复" in w for w in warnings)
def test_resolve_against_live_unique_match_and_failure() -> None:
station = _make_station()
resolved = station._resolve_parameter_entries_against_live_steps(
[{"Key": "SampleFile", "Value": "upload\\sample\\f.xlsx"}], FLATTENED_LIVE
)
assert resolved[0]["step"] == "39c78d4b-b5d3-f721-2001-9d52000084c3"
assert resolved[0]["m"] == 0 and resolved[0]["n"] == 0
# 没有 protocol 在 m/n=0/0 处 → 0 匹配
with pytest.raises(Exception) as exc:
station._resolve_parameter_entries_against_live_steps(
[{"Key": "protocol", "m": 0, "n": 0, "Value": "v"}], FLATTENED_LIVE
)
assert "0 条" in str(exc.value)
def test_group_resolved_entries_uses_lowercase_keys() -> None:
station = _make_station()
grouped = station._group_resolved_entries_to_param_values([
{"step": "39c78d4b-b5d3-f721-2001-9d52000084c3", "Key": "SampleFile", "m": 0, "n": 0, "Value": "x"},
])
step_entries = grouped["39c78d4b-b5d3-f721-2001-9d52000084c3"]
assert step_entries[0] == {"key": "SampleFile", "value": "x", "m": 0, "n": 0}
def test_create_order_payload_shape() -> None:
station = _make_station()
payload = station._create_order_payload(
order_code="EXP260518-103000",
order_name="实验260518-103000",
sub_workflow_id="3a1d35f9-63ce-67d6-1784-9f6abcca4eda",
param_values={"39c78d4b-b5d3-f721-2001-9d52000084c3": [{"key": "SampleFile", "value": "x", "m": 0, "n": 0}]},
border_number=1,
extend_properties=None,
)
assert isinstance(payload, list) and len(payload) == 1
item = payload[0]
assert item["workFlowId"] == "3a1d35f9-63ce-67d6-1784-9f6abcca4eda"
assert item["paramValues"]
assert item["extendProperties"] == ""
assert item["borderNumber"] == 1
def test_order_identity_format() -> None:
station = _make_station()
code, name = station._build_order_identity("day2")
assert code.startswith("EXP") and len(code) == 16 # EXP + YYMMDD-HHmmss
assert name.startswith("实验")
code2, name2 = station._build_order_identity("day2", "自定义")
assert name2 == "自定义"
# ---------------------------------------------------------------------------
# 5. Generic submit / day wrappers (含会抦住 BUG 1 的用例)
# ---------------------------------------------------------------------------
def _wire_submit_pipeline(station: Any) -> None:
station._resolve_workflow_binding_from_names = MagicMock(return_value={ # type: ignore[method-assign]
"workflow_name": "DAY2多肽定量",
"root_workflow_id": "3a1d35f0-9436-895b-2eda-039a5465275e",
"sub_workflow_id": "3a1d35f0-9f7e-c2c1-0bc0-8d94b81d90ca",
"sub_workflow_name": "DAY2多肽定量",
"raw": {},
})
station._resolve_workflow_binding = MagicMock(side_effect=lambda day_key: station._resolve_workflow_binding_from_names("DAY2多肽定量")) # type: ignore[method-assign]
station._query_step_parameters = MagicMock(return_value={}) # type: ignore[method-assign]
station._flatten_step_parameters = MagicMock(return_value=FLATTENED_LIVE) # type: ignore[method-assign]
station._create_order = MagicMock(return_value=json.dumps(CREATE_ALLOCATION)) # type: ignore[method-assign]
def test_submit_experiment_generic_succeeds() -> None:
"""plan §「Generic And Day 1 Submit」line 919-924这条同时抦住 BUG 1binding= 关键字)。"""
station = _make_station()
_wire_submit_pipeline(station)
result = station.submit_experiment(
{"workflow_name": "DAY2多肽定量", "sample_excel_pattern": ""},
{"parameter_overrides": []},
sample_excel_relative_path="upload/sample/f.xlsx",
)
assert result["success"] is True
assert result["order_id"] == ORDER_GUID
assert result["resultTable"]["tableName"] == "resultTable"
def test_submit_experiment_rejects_day1_alias() -> None:
station = _make_station()
with pytest.raises(Exception):
station.submit_experiment(
{"workflow_name": "Day1线肽合成", "sample_excel_pattern": "x"},
{},
sample_excel_relative_path="upload/sample/f.xlsx",
)
def test_submit_experiment_day2_calls_pipeline() -> None:
station = _make_station()
_wire_submit_pipeline(station)
result = station.submit_experiment_day2(
{"sample_excel_pattern": ""},
{"parameter_overrides": []},
sample_excel_relative_path="upload/sample/f.xlsx",
)
assert result["success"] is True
assert result["order_ids"] == [ORDER_GUID]
assert result["auto_register_materials"] is True
assert result["material_registration"]["status"] == "not_implemented"
def test_day1_placeholder_does_not_call_create_order() -> None:
station = _make_station()
station._resolve_workflow_binding = MagicMock(return_value={ # type: ignore[method-assign]
"workflow_name": "Day1线肽合成",
"root_workflow_id": "rid",
"sub_workflow_id": "sid",
"sub_workflow_name": "Day1线肽合成",
"raw": {},
})
station._create_order = MagicMock(side_effect=AssertionError("Day1 不应触达 create_order")) # type: ignore[method-assign]
out = station.submit_experiment_day1(
{"sample_excel_pattern": "", "cem_method_file_name": ""},
{},
sample_excel_relative_path="upload/sample/f.xlsx",
# 模拟人工确认框架注入的字段(这条会抦住 BUG 3
timeout_seconds=3600,
assignee_user_ids=[],
materials_loaded=False,
)
assert out["status"] == "manual_confirm_placeholder"
assert out["cem_method_file_name"] == "5microdouble-20250911.MPM"
assert isinstance(out["partial_parameter_entries"], list)
# ---------------------------------------------------------------------------
# 6. Allocation map parsing + resultTable
# ---------------------------------------------------------------------------
def test_parse_allocation_map_extracts_order_id_and_groups() -> None:
station = _make_station()
parsed = station._parse_create_order_allocation_map(json.dumps(CREATE_ALLOCATION))
assert parsed["order_ids"] == [ORDER_GUID]
assert len(parsed["allocation_rows"]) == 3
assert set(parsed["materials_by_type"].keys()) == {"Consumables", "Sample", "Future"}
def test_parse_allocation_map_handles_python_str_repr() -> None:
"""RPC.create_order 返回的是 str(dict),含单引号。"""
station = _make_station()
parsed = station._parse_create_order_allocation_map(str(CREATE_ALLOCATION))
assert parsed["order_ids"] == [ORDER_GUID]
def test_parse_allocation_map_empty() -> None:
station = _make_station()
parsed = station._parse_create_order_allocation_map("{}")
assert parsed["allocation_rows"] == []
assert parsed["order_ids"] == []
def test_build_result_table_order_and_columns() -> None:
station = _make_station()
parsed = station._parse_create_order_allocation_map(json.dumps(CREATE_ALLOCATION))
table = station._build_result_table(parsed["materials_by_type"])
assert table["tableName"] == "resultTable"
assert [c["key"] for c in table["columns"]] == ["whName", "locationCode", "materialName", "quantity"]
# 顺序Sample → Consumables → Future未知 mode 保留在末尾)
names = [row["materialName"] for row in table["data"]]
assert names == ["96孔板", "200μL枪头盒", "未知耗材"]
# locationShowName 优先 locationCode
assert table["data"][0]["locationCode"] == "A1-show"
assert table["data"][1]["locationCode"] == "1-01"
def test_build_result_table_empty_returns_empty_data() -> None:
station = _make_station()
table = station._build_result_table({})
assert table["data"] == []
assert [c["key"] for c in table["columns"]] == ["whName", "locationCode", "materialName", "quantity"]
def test_resolve_wh_name_handles_material_info_failure() -> None:
station = _make_station()
station.hardware_interface.material_info.side_effect = RuntimeError("HTTP 500")
cache: Dict[str, Dict[str, Any]] = {}
assert station._resolve_wh_name_by_material_id("mat-1", cache) == ""
def test_submit_returns_warning_when_allocation_empty() -> None:
station = _make_station()
_wire_submit_pipeline(station)
station._create_order = MagicMock(return_value="{}") # type: ignore[method-assign]
result = station.submit_experiment_day2(
{"sample_excel_pattern": ""},
{},
sample_excel_relative_path="upload/sample/f.xlsx",
)
assert "create_order_allocation_unavailable_for_result_table" in result["warnings"]
# ---------------------------------------------------------------------------
# 7. Reports + workflow records
# ---------------------------------------------------------------------------
def test_get_order_list_passes_json_string() -> None:
station = _make_station()
station.hardware_interface.order_query.return_value = {"items": [], "totalCount": 0}
station.get_order_list(filter_text="abc", page_count=10)
args, kwargs = station.hardware_interface.order_query.call_args
payload = json.loads(args[0])
assert payload["filter"] == "abc"
assert payload["pageCount"] == 10
def test_get_order_report_calls_typed_rpc() -> None:
station = _make_station()
station.hardware_interface.order_report.return_value = {"id": ORDER_GUID, "name": "x", "preIntakes": [], "resultList": []}
out = station.get_order_report(ORDER_GUID)
station.hardware_interface.order_report.assert_called_once_with(ORDER_GUID)
assert out["success"] is True
assert out["summary"]["id"] == ORDER_GUID
def test_get_aggregated_order_report_is_todo_placeholder() -> None:
station = _make_station()
out = station.get_aggregated_order_report(ORDER_GUID)
assert out["status"] == "not_implemented"
def test_query_workflow_records_filters_unsaved_subworkflows() -> None:
station = _make_station()
station.hardware_interface.query_workflow.return_value = {
"items": [
{
"id": "rid",
"name": "Day3线肽环化",
"subWorkflows": [
{"id": "saved-id", "name": "Day3线肽环化", "isSaved": True},
{"id": "draft-id", "name": "Day3线肽环化-草稿", "isSaved": False},
],
}
]
}
records = station._query_workflow_records("Day3线肽环化")
assert [r["subworkflowId"] for r in records] == ["saved-id"]
# ---------------------------------------------------------------------------
# 8. Debug / fetch_workflow_list 守护
# ---------------------------------------------------------------------------
def test_module_fetch_workflow_list_is_debug_guarded() -> None:
module = _import_module()
assert module.DEBUG_CLI_ENABLED is False
with pytest.raises(AssertionError):
module.fetch_workflow_list(config={"api_host": "http://x", "api_key": "k"})
def test_station_fetch_workflow_list_uses_rpc() -> None:
station = _make_station()
station.hardware_interface.query_workflow.return_value = {"items": [], "totalCount": 0}
station.fetch_workflow_list(filter_text="Day2")
args, _ = station.hardware_interface.query_workflow.call_args
payload = json.loads(args[0])
assert payload["filter"] == "Day2"
assert payload["includeDetail"] is True
# ---------------------------------------------------------------------------
# 9. start_experiment 装载闸门
# ---------------------------------------------------------------------------
def test_start_experiment_blocks_when_materials_not_loaded() -> None:
station = _make_station()
station.hardware_interface.scheduler_start.return_value = 1
with pytest.raises(RuntimeError):
station.start_experiment(
order_id=ORDER_GUID,
resultTable={"data": [{"materialName": "x"}]},
materials_loaded=False,
)
def test_start_experiment_starts_when_table_empty() -> None:
station = _make_station()
station.hardware_interface.scheduler_start.return_value = 1
result = station.start_experiment(order_id=ORDER_GUID, resultTable={"data": []})
assert result["success"] is True
assert result["order_ids"] == [ORDER_GUID]
# ---------------------------------------------------------------------------
# 10. Reset
# ---------------------------------------------------------------------------
def test_reset_signature_drops_legacy_params_and_uses_literal() -> None:
"""plan 调整:删除 dry_run/order_id/location_idreset_operations 用 Literal 注解。"""
cls = getattr(_import_module(), CLASS_NAME)
sig = inspect.signature(cls.reset)
params = sig.parameters
for legacy in ("dry_run", "order_id", "location_id"):
assert legacy not in params, f"reset 不应再有 {legacy} 入参"
assert "reset_operations" in params
assert any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values()), \
"reset 必须保留 **kwargs 以兜底 reset_order_id/reset_location_id"
annotation = params["reset_operations"].annotation
rendered = annotation if isinstance(annotation, str) else repr(annotation)
for op in ("scheduler_reset", "reset_order_status", "reset_location"):
assert op in rendered, f"reset_operations 的 Literal 必须包含 {op}"
def test_reset_goal_default_contains_all_operations() -> None:
"""像 sirna 一样goal_default 默认勾选全部三个 reset 操作。"""
cls = getattr(_import_module(), CLASS_NAME)
meta = getattr(cls.reset, "_action_registry_meta", {})
goal_default = meta.get("goal_default") or {}
assert goal_default.get("reset_operations") == [
"scheduler_reset",
"reset_order_status",
"reset_location",
]
def test_reset_executes_typed_rpc_calls() -> None:
station = _make_station()
station.hardware_interface.scheduler_reset.return_value = 1
station.hardware_interface.reset_order_status.return_value = 1
station.hardware_interface.reset_location.return_value = 1
out = station.reset(
reset_operations=["scheduler_reset", "reset_order_status", "reset_location"],
reset_order_id=ORDER_GUID,
reset_location_id="loc-1",
)
station.hardware_interface.scheduler_reset.assert_called_once_with()
station.hardware_interface.reset_order_status.assert_called_once_with(ORDER_GUID)
station.hardware_interface.reset_location.assert_called_once_with("loc-1")
assert out["selected_operations"] == [
"scheduler_reset",
"reset_order_status",
"reset_location",
]
assert len(out["executed_calls"]) == 3
assert out["skipped_operations"] == []
def test_reset_skips_when_ids_missing() -> None:
"""没有 order_id / location_id 时应该 skip 而不是抛错。"""
station = _make_station()
station.hardware_interface.scheduler_reset.return_value = 1
out = station.reset(
reset_operations=["scheduler_reset", "reset_order_status", "reset_location"],
)
station.hardware_interface.scheduler_reset.assert_called_once_with()
station.hardware_interface.reset_order_status.assert_not_called()
station.hardware_interface.reset_location.assert_not_called()
skipped_ops = {item["operation"] for item in out["skipped_operations"]}
assert skipped_ops == {"reset_order_status", "reset_location"}

View File

@@ -1,3 +0,0 @@
from .sirna_station import BioyondSirnaStation, fetch_workflow_list, load_sirna_config
__all__ = ["BioyondSirnaStation", "fetch_workflow_list", "load_sirna_config"]

View File

@@ -56,13 +56,17 @@ class ConnectionMonitor:
def _monitor_loop(self):
while self._running:
try:
# 使用 lightweight API 检查连接
# query_matial_type_list 是比较快的查询
start_time = time.time()
result = self.workstation.hardware_interface.material_type_list()
# 使用轻量级调度状态接口检查连接,避免启动时打印完整物料类型列表。
result = self.workstation.hardware_interface.scheduler_status()
status = "online" if result else "offline"
msg = "Connection established" if status == "online" else "Failed to get material type list"
if status == "online":
msg = (
f"Scheduler status={result.get('status')}, "
f"hasTask={result.get('hasTask')}"
)
else:
msg = "Failed to get scheduler status"
if status != self._last_status:
logger.info(f"Bioyond连接状态变更: {self._last_status} -> {status}")
@@ -730,7 +734,7 @@ class BioyondWorkstation(WorkstationBase):
"""解析 ``debug_log_dir`` 为绝对路径。"""
configured = (getattr(self, "bioyond_config", {}) or {}).get("debug_log_dir")
default_dir = getattr(self, "_DEBUG_LOG_DEFAULT_DIR", None)
candidate = configured or default_dir or "temp_benyao/_logs/bioyond_debug"
candidate = configured or default_dir or "bioyond_debug_records"
path = Path(candidate)
if not path.is_absolute():
repo_root = Path(__file__).resolve().parents[4]

View File

@@ -1 +1,9 @@
from . import sirna_materials # noqa: F401 ensure @resource classes are importable for PLR deserialize
try:
from . import peptide_materials # noqa: F401 ensure @resource classes are importable for PLR deserialize
except Exception: # pragma: no cover - 允许轻量环境导入非资源辅助函数
peptide_materials = None # type: ignore[assignment]
try:
from . import sirna_materials # noqa: F401 ensure @resource classes are importable for PLR deserialize
except Exception: # pragma: no cover - 允许轻量环境导入非资源辅助函数
sirna_materials = None # type: ignore[assignment]

View File

@@ -258,13 +258,13 @@ class BIOYOND_PeptideStation_Deck(Deck):
],
"xy_col_row",
)
WAREHOUSE_BIOYOND_KEY_AXIS = dict.fromkeys(WAREHOUSE_BIOYOND_AXIS, "row_col")
WAREHOUSE_BIOYOND_KEY_AXIS = dict.fromkeys(WAREHOUSE_BIOYOND_AXIS, "col_row")
def __init__(
self,
name: str = "PeptideStation_Deck",
size_x: float = 3500.0,
size_y: float = 1800.0,
size_x: float = 2700.0,
size_y: float = 2000.0,
size_z: float = 1500.0,
category: str = "deck",
setup: bool = False
@@ -276,6 +276,8 @@ class BIOYOND_PeptideStation_Deck(Deck):
@classmethod
def deserialize(cls, data: dict, allow_marshal: bool = False):
if data.get("children") and data.get("setup") is True:
data = data.copy()
data["setup"] = False
# 已有序列化子资源,跳过 setup 避免重复创建
result = super(BIOYOND_PeptideStation_Deck, cls).deserialize(data, allow_marshal=allow_marshal)
else:
@@ -293,81 +295,140 @@ class BIOYOND_PeptideStation_Deck(Deck):
if key_axis and not hasattr(child, "bioyond_key_axis"):
child.bioyond_key_axis = key_axis
def _frontend_y_flipped_coordinate(self, display_x: float, display_y: float, child) -> Coordinate:
"""把期望显示坐标转换为兼容前端 y 轴翻转的存储坐标。"""
return Coordinate(display_x, self.get_size_y() - display_y - child.get_size_y(), 0.0)
def setup(self) -> None:
# 多肽工作站仓库配置
# 基于 2026-05-09 live API probe 发现的实际仓库拓扑 (12个仓库)
# 数据来源: temp_benyao/peptide/_logs/warehouse_discovery_raw_live_2026-05-09.json
# 数据来源: Bioyond 现场仓库发现结果。
self.warehouses = {
# 主自动化堆栈 - live API: code 10-17 -> x=17, y=10显示为 10×17
# 主自动化堆栈 - live API: code 10-17 -> x=17, y=10显示为 17×10
"自动化堆栈": bioyond_warehouse_numeric_stack(
"自动化堆栈", rows=10, columns=17, bioyond_axis="xy_col_row", bioyond_key_axis="row_col"
"自动化堆栈",
rows=17,
columns=10,
bioyond_axis="xy_col_row",
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
# 低温存储
"低温冰箱仓库": bioyond_warehouse_live_grid(
"低温冰箱仓库", rows=2, columns=3, slot_keys=["1", "2", "3", "4", "5", "6"]
"低温冰箱仓库",
rows=3,
columns=2,
slot_keys=["1", "2", "3", "4", "5", "6"],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
# 移液站库位
"Tecan移液站库": bioyond_warehouse_live_grid(
"Tecan移液站库", rows=1, columns=18, slot_keys=[str(index) for index in range(1, 19)]
"Tecan移液站库",
rows=18,
columns=1,
slot_keys=[str(index) for index in range(1, 19)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"G3移液站库": bioyond_warehouse_live_grid(
"G3移液站库",
rows=1,
columns=18,
rows=18,
columns=1,
slot_keys=["1", "2", "3", "4", "垃圾桶", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18"],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"IDOT移液站库": bioyond_warehouse_live_grid(
"IDOT移液站库",
rows=1,
columns=12,
rows=12,
columns=1,
slot_keys=[f"0009-{index:04d}" for index in range(1, 13)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
# 缓冲库位
"G3缓冲库": bioyond_warehouse_live_grid(
"G3缓冲库", rows=1, columns=5, slot_keys=[str(index) for index in range(1, 6)]
"G3缓冲库",
rows=5,
columns=1,
slot_keys=[str(index) for index in range(1, 6)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"盖板缓冲库": bioyond_warehouse_live_grid(
"盖板缓冲库", rows=1, columns=7, slot_keys=[str(index) for index in range(1, 8)]
"盖板缓冲库",
rows=7,
columns=1,
slot_keys=[str(index) for index in range(1, 8)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"配平板缓冲库": bioyond_warehouse_live_grid(
"配平板缓冲库", rows=1, columns=3, slot_keys=[str(index) for index in range(1, 4)]
"配平板缓冲库",
rows=3,
columns=1,
slot_keys=[str(index) for index in range(1, 4)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"IDOT缓冲库": bioyond_warehouse_live_grid(
"IDOT缓冲库", rows=1, columns=2, slot_keys=["1", "1"]
"IDOT缓冲库",
rows=2,
columns=1,
slot_keys=["1", "1"],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"固相合成板底座缓冲位": bioyond_warehouse_live_grid(
"固相合成板底座缓冲位",
rows=1,
columns=4,
rows=4,
columns=1,
slot_keys=[f"0015-{index:04d}" for index in range(1, 5)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
# 设备库位
"离心机库位": bioyond_warehouse_live_grid(
"离心机库位", rows=1, columns=4, slot_keys=[f"0017-{index:04d}" for index in range(1, 5)]
"离心机库位",
rows=4,
columns=1,
slot_keys=[f"0017-{index:04d}" for index in range(1, 5)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"热封膜机位": bioyond_warehouse_live_grid(
"热封膜机位", rows=1, columns=2, slot_keys=[f"0016-{index:04d}" for index in range(1, 3)]
"热封膜机位",
rows=2,
columns=1,
slot_keys=[f"0016-{index:04d}" for index in range(1, 3)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
}
# 仓库位置布局 (需根据实际硬件布局调整)
# 仓库显示布局:紧凑排列;存储 y 坐标按前端兼容翻转预先反向。
display_layout = {
"自动化堆栈": (0.0, 0.0),
"Tecan移液站库": (1520.0, 0.0),
"G3移液站库": (1710.0, 0.0),
"IDOT移液站库": (1900.0, 0.0),
"G3缓冲库": (2090.0, 0.0),
"盖板缓冲库": (2090.0, 580.0),
"低温冰箱仓库": (2280.0, 0.0),
"配平板缓冲库": (2280.0, 370.0),
"IDOT缓冲库": (2470.0, 370.0),
"固相合成板底座缓冲位": (2280.0, 740.0),
"离心机库位": (2470.0, 740.0),
"热封膜机位": (2280.0, 1210.0),
}
self.warehouse_locations = {
"自动化堆栈": Coordinate(0.0, 0.0, 0.0),
"Tecan移液站库": Coordinate(0.0, 1150.0, 0.0),
"G3移液站库": Coordinate(0.0, 1300.0, 0.0),
"IDOT移液站库": Coordinate(0.0, 1450.0, 0.0),
"G3缓冲库": Coordinate(0.0, 1600.0, 0.0),
"盖板缓冲库": Coordinate(850.0, 1600.0, 0.0),
"低温冰箱仓库": Coordinate(2700.0, 0.0, 0.0),
"配平板缓冲库": Coordinate(2700.0, 300.0, 0.0),
"IDOT缓冲库": Coordinate(2700.0, 450.0, 0.0),
"固相合成板底座缓冲位": Coordinate(2700.0, 600.0, 0.0),
"离心机库位": Coordinate(2700.0, 750.0, 0.0),
"热封膜机位": Coordinate(2700.0, 900.0, 0.0),
name: self._frontend_y_flipped_coordinate(x, y, self.warehouses[name])
for name, (x, y) in display_layout.items()
}
for warehouse_name, warehouse in self.warehouses.items():

View File

@@ -0,0 +1,247 @@
"""Peptide Station Material Resource Definitions."""
from __future__ import annotations
from collections import OrderedDict
try:
from pylabrobot.resources import Container, Plate, TipRack
except Exception: # pragma: no cover - 允许无 pylabrobot 的轻量动作测试导入
class _FallbackResource:
def __init__(self, *args, **kwargs):
self.args = args
self.kwargs = kwargs
class Container(_FallbackResource): # type: ignore[no-redef]
pass
class Plate(_FallbackResource): # type: ignore[no-redef]
pass
class TipRack(_FallbackResource): # type: ignore[no-redef]
pass
try:
from unilabos.registry.decorators import resource
except Exception: # pragma: no cover - 允许无完整 registry 依赖时导入常量
def resource(*args, **kwargs):
def decorator(cls):
return cls
return decorator
def _ensure_itemized_ordering(kwargs: dict) -> None:
if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
kwargs["ordering"] = OrderedDict()
class _PeptideTipRack(TipRack):
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 64.0)
kwargs.setdefault("with_tips", True)
_ensure_itemized_ordering(kwargs)
super().__init__(*args, **kwargs)
class _PeptidePlate(Plate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 14.35)
kwargs.setdefault("plate_type", "skirted")
_ensure_itemized_ordering(kwargs)
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_1000ul_tip_rack",
category=["labware", "tip_rack"],
description="1000uL tip rack for Bioyond peptide station",
)
class BioyondPeptide_1000ul_TipRack(_PeptideTipRack):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_1000ul_tip_rack")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_200ul_tip_rack",
category=["labware", "tip_rack"],
description="200uL tip rack for Bioyond peptide station",
)
class BioyondPeptide_200ul_TipRack(_PeptideTipRack):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_200ul_tip_rack")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_50ul_tip_rack",
category=["labware", "tip_rack"],
description="50uL tip rack for Bioyond peptide station",
)
class BioyondPeptide_50ul_TipRack(_PeptideTipRack):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_50ul_tip_rack")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_96_well_deep_well_plate",
category=["labware", "plate"],
description="96 well deep well plate for Bioyond peptide station",
)
class BioyondPeptide_96WellDeepWellPlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_96_well_deep_well_plate")
kwargs.setdefault("size_z", 44.0)
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_96_well_synthesis_plate",
category=["labware", "plate"],
description="96 well solid-phase synthesis plate for Bioyond peptide station",
)
class BioyondPeptide_96WellSynthesisPlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_96_well_synthesis_plate")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_96_well_synthesis_plate_base",
category=["labware", "adapter"],
description="96 well solid-phase synthesis plate base for Bioyond peptide station",
)
class BioyondPeptide_96WellSynthesisPlateBase(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_96_well_synthesis_plate_base")
kwargs.setdefault("size_z", 20.0)
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_96_well_balance_plate",
category=["labware", "plate"],
description="96 well balance plate for Bioyond peptide station",
)
class BioyondPeptide_96WellBalancePlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_96_well_balance_plate")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_384_well_plate",
category=["labware", "plate"],
description="384 well plate for Bioyond peptide station",
)
class BioyondPeptide_384WellPlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_384_well_plate")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_384_lcms_plate",
category=["labware", "plate"],
description="384 well LCMS plate for Bioyond peptide station",
)
class BioyondPeptide_384LCMSPlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_384_lcms_plate")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_384_balance_plate",
category=["labware", "plate"],
description="384 well balance plate for Bioyond peptide station",
)
class BioyondPeptide_384BalancePlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_384_balance_plate")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_cover_plate",
category=["labware", "cover"],
description="Cover plate for Bioyond peptide station",
)
class BioyondPeptide_CoverPlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_cover_plate")
kwargs.setdefault("size_z", 8.0)
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_sealing_base",
category=["labware", "adapter"],
description="Sealing base for Bioyond peptide station",
)
class BioyondPeptide_SealingBase(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_sealing_base")
kwargs.setdefault("size_z", 20.0)
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_reagent_trough",
category=["labware", "trough"],
description="Reagent trough for Bioyond peptide station",
)
class BioyondPeptide_ReagentTrough(Container):
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 44.0)
kwargs.setdefault("max_volume", 300000.0)
kwargs.setdefault("model", "bioyond_peptide_reagent_trough")
super().__init__(*args, **kwargs)
DEFAULT_PEPTIDE_MATERIAL_TYPE_MAPPINGS = {
"bioyond_peptide_1000ul_tip_rack": ["1000μL枪头盒", "3a1890bb-736e-cfdd-3213-eb314e8a60f9"],
"bioyond_peptide_200ul_tip_rack": ["200μL枪头盒", "3a1890bb-36d1-964a-18bd-0bf0f2877a7b"],
"bioyond_peptide_50ul_tip_rack": ["50μL枪头盒", "3a1890bc-5fae-361c-cc09-e6f2f6dcd71d"],
"bioyond_peptide_96_well_deep_well_plate": ["96孔深孔板", "3a1890bc-1fa8-fe39-9faa-12279ed4569b"],
"bioyond_peptide_96_well_synthesis_plate": ["96孔固相合成板", "3a1871cb-99f3-f01d-23e2-08dbbd0045b5"],
"bioyond_peptide_96_well_synthesis_plate_base": ["96孔固相合成板底座", "3a1b997e-241b-64f0-80d1-47bca08799d1"],
"bioyond_peptide_96_well_balance_plate": ["96孔配平板", "3a187661-2378-1e20-fa5c-a27d49fdc15d"],
"bioyond_peptide_384_well_plate": ["384孔酶标板", "3a1890bf-2148-ed20-92bd-d85869947d9a"],
"bioyond_peptide_384_lcms_plate": ["384孔LCMS板", "3a1e6a8b-cb61-74da-a089-8e6f197f80f0"],
"bioyond_peptide_384_balance_plate": ["384孔配平板", "3a18be7e-47cc-888c-fc68-055753286826"],
"bioyond_peptide_cover_plate": ["防挥发盖板", "3a19d5a6-b0e2-b486-e5eb-bcabc632f4de"],
"bioyond_peptide_sealing_base": ["封膜底座", "3a1d1d7b-e33b-6975-165d-c56cba5ed345"],
"bioyond_peptide_reagent_trough": ["12道试剂槽", "3a18b431-ac58-ca2e-9680-2a4f5880ea45"],
}
MATERIAL_TYPE_CODE_TO_CLASS = {
"0001": BioyondPeptide_96WellSynthesisPlate,
"0002": BioyondPeptide_96WellBalancePlate,
"0008": BioyondPeptide_200ul_TipRack,
"0009": BioyondPeptide_1000ul_TipRack,
"0011": BioyondPeptide_96WellDeepWellPlate,
"0012": BioyondPeptide_50ul_TipRack,
"0016": BioyondPeptide_384WellPlate,
"0018": BioyondPeptide_384WellPlate,
"0024": BioyondPeptide_ReagentTrough,
"0026": BioyondPeptide_384BalancePlate,
"0035": BioyondPeptide_CoverPlate,
"0039": BioyondPeptide_96WellSynthesisPlateBase,
"0041": BioyondPeptide_SealingBase,
"0049": BioyondPeptide_384LCMSPlate,
}
def get_material_class_by_type_code(type_code: str):
"""Return a peptide material class by Bioyond material type code."""
return MATERIAL_TYPE_CODE_TO_CLASS.get(type_code)

View File

@@ -1,126 +0,0 @@
"""Sirna Station Material Resource Definitions
Defines PyLabRobot resource classes for Bioyond Sirna station materials.
Each class is decorated with @resource for AST-based registry discovery.
"""
from collections import OrderedDict
from pylabrobot.resources import Plate, TipRack, Container
from unilabos.registry.decorators import resource
@resource(
id="bioyond_sirna_g3_200ul_tip_rack",
category=["labware", "tip_rack"],
description="G3-200ul枪头盒 for Sirna station",
)
class BioyondSirna_G3_200ul_TipRack(TipRack):
"""G3-200ul tip rack for Sirna liquid handling."""
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 64.0)
kwargs.setdefault("model", "bioyond_sirna_g3_200ul_tip_rack")
kwargs.setdefault("with_tips", True)
if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
kwargs["ordering"] = OrderedDict()
super().__init__(*args, **kwargs)
@resource(
id="bioyond_sirna_g3_50ul_tip_rack",
category=["labware", "tip_rack"],
description="G3-50ul枪头盒 for Sirna station",
)
class BioyondSirna_G3_50ul_TipRack(TipRack):
"""G3-50ul tip rack for Sirna liquid handling."""
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 64.0)
kwargs.setdefault("model", "bioyond_sirna_g3_50ul_tip_rack")
kwargs.setdefault("with_tips", True)
if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
kwargs["ordering"] = OrderedDict()
super().__init__(*args, **kwargs)
@resource(
id="bioyond_sirna_384_well_plate",
category=["labware", "plate"],
description="384孔板 for Sirna assays",
)
class BioyondSirna_384WellPlate(Plate):
"""384-well plate for Sirna reporter gene detection."""
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 14.35)
kwargs.setdefault("model", "bioyond_sirna_384_well_plate")
kwargs.setdefault("plate_type", "skirted")
if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
kwargs["ordering"] = OrderedDict()
super().__init__(*args, **kwargs)
@resource(
id="bioyond_sirna_cell_culture_plate",
category=["labware", "plate"],
description="细胞培养板 for Sirna cell culture",
)
class BioyondSirna_CellCulturePlate(Plate):
"""Cell culture plate for Sirna experiments."""
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 14.35)
kwargs.setdefault("model", "bioyond_sirna_cell_culture_plate")
kwargs.setdefault("plate_type", "skirted")
if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
kwargs["ordering"] = OrderedDict()
super().__init__(*args, **kwargs)
@resource(
id="bioyond_sirna_reagent_trough",
category=["labware", "trough"],
description="试剂槽 for Sirna reagents",
)
class BioyondSirna_ReagentTrough(Container):
"""Reagent trough for Sirna station reagents (RiboGreen, etc.)."""
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 44.0)
kwargs.setdefault("max_volume", 300000.0)
kwargs.setdefault("model", "bioyond_sirna_reagent_trough")
super().__init__(*args, **kwargs)
# Material type code mapping for dynamic instantiation
MATERIAL_TYPE_CODE_TO_CLASS = {
"0016": BioyondSirna_G3_200ul_TipRack,
"0017": BioyondSirna_G3_50ul_TipRack,
"0015": BioyondSirna_384WellPlate,
"0001": BioyondSirna_CellCulturePlate,
"0006": BioyondSirna_ReagentTrough,
}
def get_material_class_by_type_code(type_code: str):
"""Get resource class by Bioyond material type code.
Args:
type_code: Bioyond materialTypeCode (e.g., "0016", "0017")
Returns:
Resource class or None if not found
"""
return MATERIAL_TYPE_CODE_TO_CLASS.get(type_code)

View File

@@ -25,6 +25,7 @@ def bioyond_warehouse_numeric_stack(
columns: int = 17,
bioyond_axis: str = "xy_row_col",
bioyond_key_axis: str = "row_col",
frontend_y_flip: bool = False,
) -> WareHouse:
"""创建 Bioyond 数字库位堆栈,库位名使用服务端返回的 行-列 格式。
@@ -46,17 +47,22 @@ def bioyond_warehouse_numeric_stack(
item_dx = 147.0
item_dy = 106.0
item_dz = 130.0
locations = [
Coordinate(dx + col * item_dx, dy + row * item_dy, dz)
for row in range(num_items_y)
for col in range(num_items_x)
]
resource_size_x = 127.0
resource_size_y = 86.0
resource_size_z = 25.0
size_y = dy + item_dy * num_items_y
locations = []
for row in range(num_items_y):
display_y = dy + row * item_dy
y = size_y - display_y - resource_size_y if frontend_y_flip else display_y
for col in range(num_items_x):
locations.append(Coordinate(dx + col * item_dx, y, dz))
holders = create_homogeneous_resources(
klass=ResourceHolder,
locations=locations,
resource_size_x=127.0,
resource_size_y=86.0,
resource_size_z=25.0,
resource_size_x=resource_size_x,
resource_size_y=resource_size_y,
resource_size_z=resource_size_z,
name_prefix=name,
)
if bioyond_key_axis == "row_col":
@@ -76,7 +82,7 @@ def bioyond_warehouse_numeric_stack(
warehouse = BioyondWareHouse(
name=name,
size_x=dx + item_dx * num_items_x,
size_y=dy + item_dy * num_items_y,
size_y=size_y,
size_z=dz + item_dz * num_items_z,
num_items_x=num_items_x,
num_items_y=num_items_y,
@@ -97,6 +103,7 @@ def bioyond_warehouse_live_grid(
slot_keys: list[str] | None = None,
bioyond_axis: str = "xy_col_row",
bioyond_key_axis: str = "row_col",
frontend_y_flip: bool = False,
) -> WareHouse:
"""创建 Bioyond 实测库位网格,按服务端 code 保存位点标签。
@@ -112,17 +119,22 @@ def bioyond_warehouse_live_grid(
item_dx = 147.0
item_dy = 106.0
item_dz = 130.0
locations = [
Coordinate(dx + col * item_dx, dy + row * item_dy, dz)
for row in range(num_items_y)
for col in range(num_items_x)
]
resource_size_x = 127.0
resource_size_y = 86.0
resource_size_z = 25.0
size_y = dy + item_dy * num_items_y
locations = []
for row in range(num_items_y):
display_y = dy + row * item_dy
y = size_y - display_y - resource_size_y if frontend_y_flip else display_y
for col in range(num_items_x):
locations.append(Coordinate(dx + col * item_dx, y, dz))
holders = create_homogeneous_resources(
klass=ResourceHolder,
locations=locations,
resource_size_x=127.0,
resource_size_y=86.0,
resource_size_z=25.0,
resource_size_x=resource_size_x,
resource_size_y=resource_size_y,
resource_size_z=resource_size_z,
name_prefix=name,
)
keys = slot_keys or [str(index + 1) for index in range(num_items_x * num_items_y)]
@@ -139,7 +151,7 @@ def bioyond_warehouse_live_grid(
return BioyondWareHouse(
name=name,
size_x=dx + item_dx * num_items_x,
size_y=dy + item_dy * num_items_y,
size_y=size_y,
size_z=dz + item_dz * num_items_z,
num_items_x=num_items_x,
num_items_y=num_items_y,

View File

@@ -18,3 +18,7 @@ def register():
from unilabos.devices.liquid_handling.rviz_backend import UniLiquidHandlerRvizBackend
from unilabos.devices.liquid_handling.laiyu.backend.laiyu_v_backend import UniLiquidHandlerLaiyuBackend
# noinspection PyUnresolvedReferences
from unilabos.resources.bioyond.decks import BIOYOND_SirnaStation_Deck
# noinspection PyUnresolvedReferences
from unilabos.resources.bioyond.decks import BIOYOND_PeptideStation_Deck