Schema

sim_panel.schema.get_schema(version)[source]
Return type:

SchemaSpec

Parameters:

version (str)

sim_panel.schema.validate_rows(rows, schema_version=None, *, max_errors=50)[source]

Validate rows against a specific schema version.

If schema_version is None, attempt to read it from each row’s “schema_version”. In that mode, rows with missing/unknown schema_version are marked invalid.

Return type:

ValidationReport

Parameters:

  • rows (Iterable[Dict[str, Any]])

  • schema_version (str | None)

  • max_errors (int)

sim_panel.schema.validate_unique_event_id(rows)[source]
Return type:

Tuple[bool, Optional[str]]

Parameters:

rows (Iterable[Dict[str, Any]])

Cross-row validation for v0.1.0 self-selection linkage rules:

  • Every selection row (event_type == “selection”) defines a (panelist_id, t) -> event_id mapping.

  • Every self_selection evaluation row must have selection_id that references an existing selection event_id.

  • Additionally, selection_id should match the same (panelist_id, t) as the evaluation row (sanity).

Returns (ok, problems).

Return type:

Tuple[bool, List[str]]

Parameters:

rows (Iterable[Dict[str, Any]])

class sim_panel.schema.registry.SchemaSpec(version, model)[source]

Bases: object

Parameters:

  • version (str)

  • model (Type[BaseModel])

version: str
model: Type[BaseModel]
sim_panel.schema.registry.get_schema(version)[source]
Return type:

SchemaSpec

Parameters:

version (str)

class sim_panel.schema.types.JSONValue(root=PydanticUndefined, **data)[source]

Bases: RootModel[Union[str, int, float, bool, NoneType, List[ForwardRef('JSONValue')], Dict[str, ForwardRef('JSONValue')]]]

Parameters:

root (RootModelRootType)

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sim_panel.schema.types.ColumnSpec[source]

Bases: TypedDict

name: str
dtype: str
required: bool
description: str
class sim_panel.schema.validate.RowError(index, message)[source]

Bases: object

Parameters:

  • index (int)

  • message (str)

index: int
message: str
class sim_panel.schema.validate.ValidationReport(schema_version, n_rows, n_valid, n_invalid, errors, warnings)[source]

Bases: object

Parameters:

  • schema_version (str)

  • n_rows (int)

  • n_valid (int)

  • n_invalid (int)

  • errors (List[RowError])

  • warnings (List[str])

schema_version: str
n_rows: int
n_valid: int
n_invalid: int
errors: List[RowError]
warnings: List[str]
property ok: bool
summary(*, max_errors=10, max_message_chars=300)[source]

Human-readable summary for logs/CLI/errors.

Return type:

str

Parameters:

  • max_errors (int)

  • max_message_chars (int)

sim_panel.schema.validate.validate_rows(rows, schema_version=None, *, max_errors=50)[source]

Validate rows against a specific schema version.

If schema_version is None, attempt to read it from each row’s “schema_version”. In that mode, rows with missing/unknown schema_version are marked invalid.

Return type:

ValidationReport

Parameters:

  • rows (Iterable[Dict[str, Any]])

  • schema_version (str | None)

  • max_errors (int)

sim_panel.schema.validate.validate_unique_event_id(rows)[source]
Return type:

Tuple[bool, Optional[str]]

Parameters:

rows (Iterable[Dict[str, Any]])

Cross-row validation for v0.1.0 self-selection linkage rules:

  • Every selection row (event_type == “selection”) defines a (panelist_id, t) -> event_id mapping.

  • Every self_selection evaluation row must have selection_id that references an existing selection event_id.

  • Additionally, selection_id should match the same (panelist_id, t) as the evaluation row (sanity).

Returns (ok, problems).

Return type:

Tuple[bool, List[str]]

Parameters:

rows (Iterable[Dict[str, Any]])

class sim_panel.schema.versions.v0_1_0.EventV0_1_0(**data)[source]

Bases: BaseModel

v0.1.0 supports two event types:

  1. selection (policy == “self_selection” only) - A choice set is presented at (panelist_id, t). - The panelist selects a subset (possibly empty). - This row records choice_set + selected_product_ids.

  2. evaluation (policy in {“random”,”manual”,”self_selection”}) - A single (panelist_id, product_id, t) evaluation. - Records product_display plus flexible outcomes/traces.

Note

  • selection rows are only allowed when policy==”self_selection”.

  • for self_selection evaluation rows, selection_id is required to link back.

Parameters:

  • schema_version (str)

  • event_id (str)

  • event_type (Literal['selection', 'evaluation'])

  • policy (Literal['random', 'manual', 'self_selection'])

  • panelist_id (str)

  • t (int)

  • selection_id (str | None)

  • choice_set (List[str] | None)

  • selected_product_ids (List[str] | None)

  • product_id (str | None)

  • product_display (str | None)

  • panelist_features (Dict[str, JSONValue])

  • product_features (Dict[str, JSONValue])

  • outcomes (Dict[str, JSONValue] | None)

  • traces (Dict[str, JSONValue] | None)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

schema_version: str
event_id: str
event_type: Literal['selection', 'evaluation']
policy: Literal['random', 'manual', 'self_selection']
panelist_id: str
t: int
selection_id: str | None
choice_set: List[str] | None
selected_product_ids: List[str] | None
product_id: str | None
product_display: str | None
panelist_features: Dict[str, JSONValue]
product_features: Dict[str, JSONValue]
outcomes: Dict[str, JSONValue] | None
traces: Dict[str, JSONValue] | None