chariot.drift package
Submodules
chariot.drift.monitor module
- class chariot.drift.monitor.Check(*, time_create: ~datetime.datetime = <factory>, time_modify: ~datetime.datetime = <factory>, time_delete: ~datetime.datetime | None = None, check_id: ~chariot.drift._types.CheckID, model_id: ~chariot.drift._types.ModelID, check_status: ~chariot.drift._types.CheckStatus = CheckStatus.DRIFT_NONE, window_begin: ~datetime.datetime, window_end: ~datetime.datetime, metric: ~chariot.drift._types.DriftMetric, attributes: dict[str, ~typing.Any] = <factory>)[source]
Bases:
Record- attributes: dict[str, Any]
- check_id: CheckID
- check_status: CheckStatus
- metric: DriftMetric
- model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_id: ModelID
- window_begin: datetime
- window_end: datetime
- class chariot.drift.monitor.CheckJob(*, job_id: JobID)[source]
Bases:
BaseModel- job_id: JobID
- model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class chariot.drift.monitor.CheckPoint(*, time: datetime, drift: float)[source]
Bases:
BaseModel- drift: float
- model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- time: datetime
- class chariot.drift.monitor.DriftAlert(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
str,Enum- NOTIFY: str = 'notify'
- SILENT: str = 'silent'
- class chariot.drift.monitor.DriftMetric(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
str,Enum- CRAMER_VON_MISES: str = 'cramer-von-mises'
- KOLMOGOROV_SMIRNOV: str = 'kolmogorov-smirnov'
- SEMANTIC: str = 'semantic'
- class chariot.drift.monitor.Monitor(*, time_create: ~datetime.datetime = <factory>, time_modify: ~datetime.datetime = <factory>, time_delete: ~datetime.datetime | None = None, monitor_status: ~chariot.drift._types.MonitorStatus = MonitorStatus.DISABLED, model_id: ~chariot.drift._types.ModelID, drift_alert: ~chariot.drift._types.DriftAlert = DriftAlert.SILENT, drift_metric: ~chariot.drift._types.DriftMetric, window_seconds: int, current_workflow: str | None = None, last_run: ~datetime.datetime | None = None, minimum_cardinality: int | None = None, maximum_cardinality: int | None = None)[source]
Bases:
Record- current_workflow: str | None
- drift_alert: DriftAlert
- drift_metric: DriftMetric
- last_run: datetime | None
- maximum_cardinality: int | None
- minimum_cardinality: int | None
- model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_id: ModelID
- monitor_status: MonitorStatus
- window_seconds: int
- class chariot.drift.monitor.MonitorAddSpec(*, model_id: ModelID, monitor_status: MonitorStatus, drift_alert: DriftAlert, drift_metric: DriftMetric, window_seconds: int, minimum_cardinality: int | None = None, maximum_cardinality: int | None = None, data_source: str = '*')[source]
Bases:
BaseModel- data_source: str
- drift_alert: DriftAlert
- drift_metric: DriftMetric
- maximum_cardinality: int | None
- minimum_cardinality: int | None
- model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_id: ModelID
- monitor_status: MonitorStatus
- window_seconds: int
- class chariot.drift.monitor.MonitorStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
str,Enum- DISABLED: str = 'disabled'
- ENABLED: str = 'enabled'
- class chariot.drift.monitor.MonitorUpdateSpec(*, monitor_status: MonitorStatus | None = None, drift_alert: DriftAlert | None = None, drift_metric: DriftMetric | None = None, window_seconds: int | None = None, current_workflow: str | Literal['NONE'] | None = None, last_run: datetime | Literal['NONE'] | None = None, minimum_cardinality: int | None = None, maximum_cardinality: int | None = None)[source]
Bases:
BaseModel- current_workflow: str | Literal['NONE'] | None
- drift_alert: DriftAlert | None
- drift_metric: DriftMetric | None
- last_run: datetime | Literal['NONE'] | None
- maximum_cardinality: int | None
- minimum_cardinality: int | None
- model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- monitor_status: MonitorStatus | None
- window_seconds: int | None
- chariot.drift.monitor.check_now(model_id: ModelID, metric: DriftMetric) CheckJob[source]
- chariot.drift.monitor.check_summary(model_id: ModelID, metric: DriftMetric, time_begin: datetime, time_end: datetime) list[CheckPoint][source]
- chariot.drift.monitor.check_window(model_id: ModelID, metric: DriftMetric, time_begin: datetime, time_end: datetime) list[Check][source]
- chariot.drift.monitor.monitor_add(spec: MonitorAddSpec) None[source]
- chariot.drift.monitor.monitor_remove(model_id: ModelID, metric: DriftMetric) None[source]
- chariot.drift.monitor.monitor_update(model_id: ModelID, metric: DriftMetric, spec: MonitorUpdateSpec) None[source]
chariot.drift.representer module
- class chariot.drift.representer.DetectorJob(*, time_create: ~datetime.datetime = <factory>, time_modify: ~datetime.datetime = <factory>, time_delete: ~datetime.datetime | None = None, generated_id: str, project_id: str, model_id: str, metric: ~chariot.drift._types.DriftMetric, workflow_id: str | None = None, job_state: ~chariot.drift._types.DetectorJobState, storage_url: str | None = None, attributes: dict)[source]
Bases:
Record- attributes: dict
- generated_id: str
- job_state: DetectorJobState
- metric: DriftMetric
- model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_id: str
- project_id: str
- storage_url: str | None
- workflow_id: str | None
- class chariot.drift.representer.DetectorJobSpec(*, project_id: str, model_id: str, metric: DriftMetric, embedding_count_per_dataset_version: int | None = None)[source]
Bases:
BaseModel- embedding_count_per_dataset_version: int | None
- metric: DriftMetric
- model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_id: str
- project_id: str
- class chariot.drift.representer.DetectorJobState(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
str,Enum- ABORTED = 'aborted'
- COMPLETED = 'completed'
- FAILED = 'failed'
- PENDING = 'pending'
- RUNNING = 'running'
- class chariot.drift.representer.EmbeddingJob(*, generated_id: str, project_id: str | None = None, dataset_id: str | None = None, dataset_version_id: str | None = None, snapshot_id: str | None = None, split: str | None = None, embedding_model_id: str, workflow_id: str | None = None, job_state: EmbeddingJobState, attributes: dict)[source]
Bases:
BaseModel- attributes: dict
- dataset_id: str | None
- dataset_version_id: str | None
- embedding_model_id: str
- generated_id: str
- job_state: EmbeddingJobState
- model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- project_id: str | None
- snapshot_id: str | None
- split: str | None
- workflow_id: str | None
- class chariot.drift.representer.EmbeddingJobSpec(*, snapshot_id: str, split: str | None = None, embedding_model_id: str | None = None, embedding_count: int | None = None)[source]
Bases:
BaseModel- embedding_count: int | None
- embedding_model_id: str | None
- model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- snapshot_id: str
- split: str | None
- class chariot.drift.representer.EmbeddingJobState(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]
Bases:
str,Enum- ABORTED = 'aborted'
- COMPLETED = 'completed'
- FAILED = 'failed'
- PENDING = 'pending'
- RUNNING = 'running'
- chariot.drift.representer.detector_job_model(model_id: ModelID, metric: DriftMetric) DetectorJob | None[source]
- chariot.drift.representer.detector_job_remove(model_id: ModelID, metric: DriftMetric) None[source]
- chariot.drift.representer.detector_job_run(spec: DetectorJobSpec) DetectorJob[source]
- chariot.drift.representer.detector_job_status(job_id: str) DetectorJob | None[source]
- chariot.drift.representer.embedding_job_dataset(snapshot_id: str, split: str | None = None) EmbeddingJob | None[source]
- chariot.drift.representer.embedding_job_remove(snapshot_id: str = None, split: str | None = None) None[source]
- chariot.drift.representer.embedding_job_run(spec: EmbeddingJobSpec) EmbeddingJob[source]
- chariot.drift.representer.embedding_job_status(job_id: str) EmbeddingJob | None[source]