Core SDK Reference

The Core SDK provides the foundational type system, Pydantic models, and cross-plane utilities that power the Xovis Open SDK. It serves as the Base Layer ensuring strict validation and consistent behavior across the entire quadrifurcated architecture.

Architectural Pillars

Strict Type Enforcement: Leveraging Pydantic V2, the core models ensure that every piece of data entering the SDK—whether from an edge sensor or a Cloud HUB—is validated against rigorous schemas.
Universal Models: Unified representations for Device and HubDevice allow for seamless transition between single-sensor management and fleet-scale orchestration.
High-Performance Utilities: Specialized primitives for asynchronous loops, privacy hashing, and ISO-8601 time handling, optimized for the SDK's high-throughput requirements.

Models

`xovis.models.device`

Xovis SDK - Device Models

Operates within the Control Plane. Provides strict Pydantic V2 data validation and alias mapping for local edge sensor endpoints that fall outside the scope of the auto-generated OpenAPI schema.

Classes

`AgentConfig`

Bases: BaseModel

Composite configuration for a data push agent.

Source code in src/xovis/models/device.py

class AgentConfig(BaseModel):
    """Composite configuration for a data push agent."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    scheduler: Scheduler
    data: DataConfig
    filters: Optional[DataPushFilters] = None

`BlockedSpace`

Bases: BaseModel

Represents a Blocked Space zone where the sensor's view is obstructed.

Attributes:

Name	Type	Description
`id`	`int`	Unique identifier for the blocked space zone.
`name`	`str`	Human-readable name of the blocked space.
`coordinates`	`List[Tuple[float, float]]`	Bounding polygon coordinates.

Source code in src/xovis/models/device.py

class BlockedSpace(BaseModel):
    """
    Represents a Blocked Space zone where the sensor's view is obstructed.

    Attributes:
        id (int): Unique identifier for the blocked space zone.
        name (str): Human-readable name of the blocked space.
        coordinates (List[Tuple[float, float]]): Bounding polygon coordinates.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    id: int
    name: str = Field(default="Blocked Space")
    coordinates: list[tuple[float, float]] = Field(
        alias="polygon",
        default_factory=list,
        description="List of (x, y) coordinates forming the blocked space polygon.",
    )

`CountAction`

Bases: Enum

Types of count actions triggered by a modifier.

Source code in src/xovis/models/device.py

class CountAction(Enum):
    """Types of count actions triggered by a modifier."""

    INCREMENT = "INCREMENT"
    DECREMENT = "DECREMENT"
    WRONG_WAY = "WRONG_WAY"

`CountEvent`

Bases: BaseModel

Links a modifier trigger to a specific counter action.

Source code in src/xovis/models/device.py

class CountEvent(BaseModel):
    """Links a modifier trigger to a specific counter action."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    counter_id: int
    type: CountAction = Field(default=CountAction.INCREMENT)
    histogram: HistogramType | str | None = Field(None, description="Trigger for age histograms (e.g., PERSON_AGE).")
    geometry_id: int | None = Field(None, description="Optional geometry reference for dwell-time calculations.")

    def model_dump(self, **kwargs):
        kwargs.setdefault("mode", "json")
        return super().model_dump(**kwargs)

`Counter`

Bases: BaseModel

Represents a logic counter in a Custom Logic.

Bridges the simplified counter definition with the hardware's internal ID/Name/Type slots.

Age Histograms

Typically an ACCUMULATION counter with: - histogram: HistogramType.PERSON_AGE - bins: [20.0, 40.0, 60.0] (example boundaries)

Hardware Limit: Sensors typically handle ~60-80 Counters per layer. UI Tip: Use CounterName values to trigger specific dashboard icons (Male, Female, etc.).

Source code in src/xovis/models/device.py

class Counter(BaseModel):
    """
    Represents a logic counter in a Custom Logic.

    Bridges the simplified counter definition with the hardware's internal ID/Name/Type slots.

    Age Histograms:
        Typically an ACCUMULATION counter with:
        - histogram: HistogramType.PERSON_AGE
        - bins: [20.0, 40.0, 60.0] (example boundaries)

    Hardware Limit: Sensors typically handle ~60-80 Counters per layer.
    UI Tip: Use CounterName values to trigger specific dashboard icons (Male, Female, etc.).
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    id: int | None = Field(None, description="System-assigned or CLIENT-forced unique identifier.")
    name: CounterName | str = Field(..., description="Name of the counter. Use CounterName for UI icons.")
    type: CounterType = Field(default=CounterType.ACCUMULATION, description="Behavior: state (inc/dec) or accumulation.")
    quantity: CounterQuantity = Field(default=CounterQuantity.COUNT, description="Measurement unit (count or time).")
    logic_id: int | None = Field(None, description="The parent logic ID this counter belongs to.")
    histogram: HistogramType | str | None = Field(None, description="Optional histogram trigger (e.g., PERSON_AGE).")
    bins: list[float] | None = Field(None, description="Pre-defined histogram bins (e.g., [20, 40, 60]).")

    def model_dump(self, **kwargs):
        kwargs.setdefault("mode", "json")
        return super().model_dump(**kwargs)

`CounterName`

Bases: str, Enum

Standardized counter names that trigger specific UI rendering/icons. Relying on these strings enables the 'Naming Trick' for high-fidelity representation in the device UI.

Source code in src/xovis/models/device.py

class CounterName(str, Enum):
    """
    Standardized counter names that trigger specific UI rendering/icons.
    Relying on these strings enables the 'Naming Trick' for high-fidelity representation in the device UI.
    """

    # Directional
    FORWARD = "fw"
    BACKWARD = "bw"
    # Gender (Triggers CSS Icon Overrides)
    FORWARD_MALE = "fw-male"
    BACKWARD_MALE = "bw-male"
    FORWARD_FEMALE = "fw-female"
    BACKWARD_FEMALE = "bw-female"
    # Object Specific
    FORWARD_BICYCLE = "fw-bicycle"
    BACKWARD_BICYCLE = "bw-bicycle"
    FORWARD_WHEELCHAIR = "fw-wheelchair"
    BACKWARD_WHEELCHAIR = "bw-wheelchair"
    FORWARD_PRAM = "fw-pram"
    BACKWARD_PRAM = "bw-pram"
    # Mask Detection
    FORWARD_MASK = "fw-mask"
    FORWARD_NO_MASK = "fw-no_mask"
    # Occupancy / Balance
    BALANCE = "balance"
    VISITS = "visits"
    DWELL_TIME = "dwell_time"
    IN = "in"
    OUT = "out"
    # Queue Logic
    QUEUE_LENGTH = "queue-length"
    OUTFLOW = "outflow"
    QUEUEING_TIME = "queueing-time"

`CounterQuantity`

Bases: Enum

Measurement unit for the counter.

Source code in src/xovis/models/device.py

class CounterQuantity(Enum):
    """Measurement unit for the counter."""

    COUNT = "count"
    TIME = "time"

`CounterType`

Bases: Enum

Available types for logic counters.

Source code in src/xovis/models/device.py

class CounterType(Enum):
    """Available types for logic counters."""

    STATE = "state"
    ACCUMULATION = "accumulation"

`DataConfig`

Bases: BaseModel

Core configuration for data content and resolution.

Source code in src/xovis/models/device.py

class DataConfig(BaseModel):
    """Core configuration for data content and resolution."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    resolution: Optional[str] = None
    package_size: int = Field(default=1)
    include_empty: Optional[bool] = None
    empty_frames: Optional[str] = None
    meta_data_package_full: bool = Field(default=False)
    meta_data_sensor_full: bool = Field(default=False)
    meta_data_config_enable: bool = Field(default=False)
    format: DataFormat = Field(default_factory=DataFormat)
    normalization: Optional[Union[list[str], Literal["ALL", "NONE"]]] = Field(default=None)

`DataFormat`

Bases: BaseModel

Configuration for data serialization.

Source code in src/xovis/models/device.py

class DataFormat(BaseModel):
    """Configuration for data serialization."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    type: DataFormatType = Field(default=DataFormatType.JSON)
    version: str = Field(default="5.x")
    pretty: bool = Field(default=False)
    time: TimeFormat = Field(default=TimeFormat.UNIX_TIME_MS)

`DataFormatType`

Bases: str, Enum

Available serialization formats for pushed data.

Source code in src/xovis/models/device.py

class DataFormatType(str, Enum):
    """Available serialization formats for pushed data."""

    JSON = "JSON"
    PROTOBUF = "PROTOBUF"
    BINARY = "BINARY"
    LEGACY_XOVIS_XML = "LEGACY_XOVIS_XML"
    RECORDING = "RECORDING"

`DataPushAgent`

Bases: BaseModel

Bridge model for a DataPush Agent.

Source code in src/xovis/models/device.py

class DataPushAgent(BaseModel):
    """
    Bridge model for a DataPush Agent.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    id: Optional[int] = None
    name: str
    type: DataPushType
    enabled: bool = Field(default=True)
    connection: int = Field(..., validation_alias=AliasChoices("connection", "connectionId"))
    config: AgentConfig

`DataPushAgentCollection`

Bases: BaseModel

Collection of DataPush Agents.

Source code in src/xovis/models/device.py

class DataPushAgentCollection(BaseModel):
    """Collection of DataPush Agents."""

    agents: list[DataPushAgent] = Field(default_factory=list)

`DataPushConnection`

Bases: BaseModel

Bridge model for a DataPush Connection.

Source code in src/xovis/models/device.py

class DataPushConnection(BaseModel):
    """
    Bridge model for a DataPush Connection.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    id: Optional[int] = None
    name: str
    protocol: DataPushProtocol
    config: Union[HTTPConfig, FTPConfig, SFTPConfig, MQTTConfig, TCPConfig, UDPConfig]

    def model_dump(self, **kwargs) -> dict[str, Any]:
        data = super().model_dump(**kwargs)
        # Ensure config field is present and serialized correctly
        if "config" not in data:
            data["config"] = self.config.model_dump(**kwargs)
        return data

`DataPushConnectionCollection`

Bases: BaseModel

Collection of DataPush Connections.

Source code in src/xovis/models/device.py

class DataPushConnectionCollection(BaseModel):
    """Collection of DataPush Connections."""

    connections: list[DataPushConnection] = Field(default_factory=list)

`DataPushFilters`

Bases: BaseModel

Fine-grained filters for data push content.

Source code in src/xovis/models/device.py

class DataPushFilters(BaseModel):
    """Fine-grained filters for data push content."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    # Literal MUST come first to prevent Pydantic from coercing "ALL" into ["ALL"]
    included_objects: Optional[Union[Literal["ALL", "NONE"], list[str]]] = Field(
        default=None, validation_alias=AliasChoices("included_objects", "includedObjects")
    )
    included_scene_events: Optional[Union[Literal["ALL", "NONE"], list[str]]] = Field(
        default=None, validation_alias=AliasChoices("included_scene_events", "includedSceneEvents")
    )
    included_count_events: Optional[Union[Literal["ALL", "NONE"], list[str]]] = Field(
        default=None, validation_alias=AliasChoices("included_count_events", "includedCountEvents")
    )
    included_info_events: Optional[Union[Literal["ALL", "NONE"], list[str]]] = Field(
        default=None, validation_alias=AliasChoices("included_info_events", "includedInfoEvents")
    )
    filter_events_by_objects: Optional[bool] = Field(
        default=None,
        validation_alias=AliasChoices("filter_events_by_objects", "filterEventsByObjects"),
    )
    included_logics: Optional[Union[Literal["ALL", "NONE"], list[int]]] = Field(
        default=None, validation_alias=AliasChoices("included_logics", "includedLogics")
    )

`DataPushProtocol`

Bases: str, Enum

Supported data transfer protocols for connections.

Source code in src/xovis/models/device.py

class DataPushProtocol(str, Enum):
    """Supported data transfer protocols for connections."""

    HTTP = "HTTP"
    FTP = "FTP"
    SFTP = "SFTP"
    MQTT = "MQTT"
    TCP = "TCP"
    UDP = "UDP"

`DataPushStatus`

Bases: BaseModel

Bridge model for DataPush Agent status and diagnostics.

Source code in src/xovis/models/device.py

class DataPushStatus(BaseModel):
    """
    Bridge model for DataPush Agent status and diagnostics.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    id: int
    name: str
    type: DataPushType
    no_of_successful: int = Field(default=0, alias="transmit.no_of_successful")
    last_successful: Optional[TransmitStatus] = Field(None, alias="transmit.last_successful")
    no_of_failed: int = Field(default=0, alias="transmit.no_of_failed")
    last_failed: Optional[TransmitStatus] = Field(None, alias="transmit.last_failed")
    sent_total: str = Field(default="0B", alias="transmit.sent_total")
    sent_total_bytes: int = Field(default=0, alias="transmit.sent_total_bytes")

    @model_validator(mode="before")
    @classmethod
    def _flatten_transmit(cls, data: Any) -> Any:
        if isinstance(data, dict) and "transmit" in data:
            transmit = data.pop("transmit")
            if isinstance(transmit, dict):
                for k, v in transmit.items():
                    data[f"transmit.{k}"] = v
        return data

`DataPushStatusCollection`

Bases: BaseModel

Collection of DataPush Agent statuses.

Source code in src/xovis/models/device.py

class DataPushStatusCollection(BaseModel):
    """Collection of DataPush Agent statuses."""

    agent_states: list[DataPushStatus] = Field(default_factory=list, alias="status")
    last_stored: Optional[Union[dict[str, Any], str]] = None

    @model_validator(mode="before")
    @classmethod
    def _flatten_collection(cls, data: Any) -> Any:
        if isinstance(data, dict) and "status" in data and "agent_states" not in data:
            data["agent_states"] = data.pop("status")
        return data

`DataPushTestResponse`

Bases: BaseModel

Result of a DataPush connection test.

Source code in src/xovis/models/device.py

class DataPushTestResponse(BaseModel):
    """Result of a DataPush connection test."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    status: str = Field(..., alias="connection_test.status")
    code: Optional[int] = Field(None, alias="connection_test.server_response.code")
    info: Optional[str] = Field(None, alias="connection_test.server_response.info")

    @model_validator(mode="before")
    @classmethod
    def _flatten_response(cls, data: Any) -> Any:
        if isinstance(data, dict) and "connection_test" in data:
            test = data["connection_test"]
            if isinstance(test, dict):
                data["connection_test.status"] = test.get("status")
                resp = test.get("server_response")
                if isinstance(resp, dict):
                    data["connection_test.server_response.code"] = resp.get("code")
                    data["connection_test.server_response.info"] = resp.get("info")
        return data

`DataPushTriggerConfig`

Bases: BaseModel

Configuration for a manual data push trigger.

Source code in src/xovis/models/device.py

class DataPushTriggerConfig(BaseModel):
    """Configuration for a manual data push trigger."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    type: DataPushTriggerType
    time_from: Optional[XovisTime] = None
    time_to: Optional[XovisTime] = None
    package_id_start: Optional[int] = None
    file_name_prefix: Optional[str] = None

    @field_serializer("time_from", "time_to")
    def _serialize_as_iso8601(self, value: Optional[int]) -> Optional[str]:
        """Ensures timestamps are serialized as ISO-8601 UTC strings for the Trigger API."""
        if value is not None:
            dt = datetime.fromtimestamp(value / 1000.0, tz=timezone.utc)
            return dt.isoformat().replace("+00:00", "Z")
        return None

Methods:

`_serialize_as_iso8601(value)`

Ensures timestamps are serialized as ISO-8601 UTC strings for the Trigger API.

Source code in src/xovis/models/device.py

@field_serializer("time_from", "time_to")
def _serialize_as_iso8601(self, value: Optional[int]) -> Optional[str]:
    """Ensures timestamps are serialized as ISO-8601 UTC strings for the Trigger API."""
    if value is not None:
        dt = datetime.fromtimestamp(value / 1000.0, tz=timezone.utc)
        return dt.isoformat().replace("+00:00", "Z")
    return None

`DataPushTriggerInfo`

Bases: BaseModel

Status information for a running or finished trigger push.

Source code in src/xovis/models/device.py

class DataPushTriggerInfo(BaseModel):
    """Status information for a running or finished trigger push."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    status: DataPushTriggerStatus
    trigger_time: Optional[str] = None
    trigger_config: Optional[DataPushTriggerConfig] = None

`DataPushTriggerStatus`

Bases: str, Enum

Operational status of a trigger push.

Source code in src/xovis/models/device.py

class DataPushTriggerStatus(str, Enum):
    """Operational status of a trigger push."""

    IDLE = "IDLE"
    BUSY = "BUSY"

`DataPushTriggerType`

Bases: str, Enum

Available trigger modes for manual data recovery.

Source code in src/xovis/models/device.py

class DataPushTriggerType(str, Enum):
    """Available trigger modes for manual data recovery."""

    ALL = "ALL"
    TIME_RANGE = "TIME_RANGE"
    LAST_PACKAGE = "LAST_PACKAGE"
    DUMMY_DATA = "DUMMY_DATA"

`DataPushType`

Bases: str, Enum

Supported types of data push agents.

Source code in src/xovis/models/device.py

class DataPushType(str, Enum):
    """Supported types of data push agents."""

    LOGICS = "LOGICS"
    LIVE_DATA = "LIVE_DATA"
    STATUS = "STATUS"
    WIFI_BT = "WIFI_BT"
    RECORDING = "RECORDING"

`FTPConfig`

Bases: BaseModel

Configuration for FTP(S) data push connections.

Source code in src/xovis/models/device.py

class FTPConfig(BaseModel):
    """Configuration for FTP(S) data push connections."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    uri: str
    user: str
    password: str
    port: Optional[int] = None
    path: Optional[str] = None
    ssl_enable: bool = Field(default=False)
    account_info: Optional[str] = None
    alternative_to_user: Optional[str] = None
    connection_timeout_s: float = Field(default=2.0)
    response_timeout_s: float = Field(default=2.0)
    create_directories: bool = Field(default=True)
    directory_mode: FTPDirectoryMode = Field(default=FTPDirectoryMode.SINGLECWD)
    file_mode: FTPFileMode = Field(default=FTPFileMode.PACKAGE)
    max_file_size: int = Field(default=0)
    ignore_proxy: bool = Field(default=False)
    use_pret: bool = Field(default=False)

`FTPDirectoryMode`

Bases: str, Enum

FTP directory traversing method.

Source code in src/xovis/models/device.py

class FTPDirectoryMode(str, Enum):
    """FTP directory traversing method."""

    SINGLECWD = "SINGLECWD"
    MULTICWD = "MULTICWD"
    NOCWD = "NOCWD"

`FTPFileMode`

Bases: str, Enum

FTP file transmission mode.

Source code in src/xovis/models/device.py

class FTPFileMode(str, Enum):
    """FTP file transmission mode."""

    PACKAGE = "PACKAGE"
    APPEND_INTERVAL = "APPEND_INTERVAL"
    APPEND_MAX_SIZE = "APPEND_MAX_SIZE"

`Filter`

Bases: BaseModel

Represents a single filter operand or operator in a Reverse Polish Notation (RPN) stack.

Source code in src/xovis/models/device.py

class Filter(BaseModel):
    """
    Represents a single filter operand or operator in a Reverse Polish Notation (RPN) stack.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    type: FilterType | str = Field(..., description="Operand type (e.g., has_gender) or Operator (AND, OR).")
    payload: dict[str, Any] = Field(default_factory=dict, description="Additional parameters for the filter.")

    def model_dump(self, **kwargs):
        """Flat serialization for RPN compatibility."""
        # Use mode="json" by default for Filter to ensure nested types are converted
        kwargs.get("mode")
        kwargs.setdefault("mode", "json")

        # Get the standard dump
        data = super().model_dump(**kwargs)

        # Extract payload and flatten it
        payload = data.pop("payload", {})
        if isinstance(payload, dict):
            data.update(payload)

        return data

Methods:

`model_dump(**kwargs)`

Flat serialization for RPN compatibility.

Source code in src/xovis/models/device.py

def model_dump(self, **kwargs):
    """Flat serialization for RPN compatibility."""
    # Use mode="json" by default for Filter to ensure nested types are converted
    kwargs.get("mode")
    kwargs.setdefault("mode", "json")

    # Get the standard dump
    data = super().model_dump(**kwargs)

    # Extract payload and flatten it
    payload = data.pop("payload", {})
    if isinstance(payload, dict):
        data.update(payload)

    return data

`FilterType`

Bases: str, Enum

Comprehensive list of available filter types (Operands and Operators) for Custom Logic. Used in the RPN (Reverse Polish Notation) filter stack.

Source code in src/xovis/models/device.py

class FilterType(str, Enum):
    """
    Comprehensive list of available filter types (Operands and Operators) for Custom Logic.
    Used in the RPN (Reverse Polish Notation) filter stack.
    """

    # Logical Operators
    AND = "AND"
    OR = "OR"
    NOT = "NOT"
    # Basic
    TRUE = "true"
    FALSE = "false"
    # Attributes
    HAS_GENDER = "has_gender"
    HAS_TAG = "has_tag"
    HAS_FACE_MASK = "has_face_mask"
    # Geometry Interactions
    HAS_CROSSED_LINE = "has_crossed_line"
    HAS_VISITED_ZONE = "has_visited_zone"
    IS_IN_ZONE = "is_in_zone"
    IS_CREATED_IN_ZONE = "is_created_in_zone"
    # Height
    PERSON_HEIGHT_BIGGER_THAN = "person_height_bigger_than"
    PERSON_HEIGHT_SMALLER_THAN = "person_height_smaller_than"
    PERSON_HEIGHT_STRICTLY_BIGGER_THAN = "person_height_strictly_bigger_than"
    PERSON_HEIGHT_STRICTLY_SMALLER_THAN = "person_height_strictly_smaller_than"
    # Interaction Attributes (Evaluated at the moment of geometry interaction)
    HAS_FIRST_INTERACTION_GENDER = "has_first_interaction_gender"
    HAS_FIRST_INTERACTION_TAG = "has_first_interaction_tag"
    HAS_FIRST_INTERACTION_FACE_MASK = "has_first_interaction_face_mask"
    # Interaction Height
    FIRST_INTERACTION_HEIGHT_BIGGER_THAN = "first_interaction_person_height_bigger_than"
    FIRST_INTERACTION_HEIGHT_SMALLER_THAN = "first_interaction_person_height_smaller_than"
    FIRST_INTERACTION_HEIGHT_STRICTLY_BIGGER_THAN = "first_interaction_person_height_strictly_bigger_than"
    FIRST_INTERACTION_HEIGHT_STRICTLY_SMALLER_THAN = "first_interaction_person_height_strictly_smaller_than"
    # Advanced Geometry Counters
    NUMBER_OF_LINE_CROSSINGS = "number_of_line_crossings"
    NUMBER_OF_FORWARD_LINE_CROSSINGS = "number_of_forward_line_crossings"
    NUMBER_OF_BACKWARD_LINE_CROSSINGS = "number_of_backward_line_crossings"
    NUMBER_OF_ZONE_ENTRIES = "number_of_zone_entries"
    NUMBER_OF_ZONE_EXITS = "number_of_zone_exits"
    # Dwell Time
    ZONE_DWELL_TIME_BIGGER_THAN = "zone_dwell_time_bigger_than"
    ZONE_DWELL_TIME_SMALLER_THAN = "zone_dwell_time_smaller_than"
    ZONE_DWELL_TIME_STRICTLY_BIGGER_THAN = "zone_dwell_time_strictly_bigger_than"
    ZONE_DWELL_TIME_STRICTLY_SMALLER_THAN = "zone_dwell_time_strictly_smaller_than"
    ZONE_DWELL_TIME_CUMULATIVE_BIGGER_THAN = "zone_dwell_time_cumulative_bigger_than"
    ZONE_DWELL_TIME_CUMULATIVE_SMALLER_THAN = "zone_dwell_time_cumulative_smaller_than"
    ZONE_DWELL_TIME_CUMULATIVE_STRICTLY_BIGGER_THAN = "zone_dwell_time_cumulative_strictly_bigger_than"
    ZONE_DWELL_TIME_CUMULATIVE_STRICTLY_SMALLER_THAN = "zone_dwell_time_cumulative_strictly_smaller_than"
    # Directions
    FIRST_LINE_CROSS_DIRECTION = "first_line_cross_direction"
    LAST_LINE_CROSS_DIRECTION = "last_line_cross_direction"

`HTTPAuthMethod`

Bases: str, Enum

HTTP authentication methods.

Source code in src/xovis/models/device.py

class HTTPAuthMethod(str, Enum):
    """HTTP authentication methods."""

    NONE = "NONE"
    BASIC = "BASIC"
    DIGEST = "DIGEST"
    DIGEST_IE = "DIGEST_IE"
    BEARER_TOKEN = "BEARER_TOKEN"

`HTTPConfig`

Bases: BaseModel

Configuration for HTTP(S) data push connections.

Source code in src/xovis/models/device.py

class HTTPConfig(BaseModel):
    """Configuration for HTTP(S) data push connections."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    uri: str
    port: Optional[int] = None
    ssl_enable: bool = Field(default=False)
    auth_method: HTTPAuthMethod = Field(default=HTTPAuthMethod.NONE)
    auth_data: Optional[str] = None
    user: Optional[str] = None
    password: Optional[str] = None
    connection_timeout_s: float = Field(default=2.0)
    chunked_transfer_enabled: bool = Field(default=False)
    ignore_proxy: bool = Field(default=False)
    custom_header_fields: Optional[list[HTTPHeaderField]] = Field(default=None)

`HTTPHeaderField`

Bases: BaseModel

Custom HTTP header field.

Source code in src/xovis/models/device.py

class HTTPHeaderField(BaseModel):
    """Custom HTTP header field."""

    name: str
    value: str

`HeatHeightMap`

Bases: BaseModel

Bridge model for spatial heat and height map data.

Abstracts the 2D floating-point array and its mapping metadata. Note: The two-dimensional 'data' array must be scaled to the background image.

Source code in src/xovis/models/device.py

class HeatHeightMap(BaseModel):
    """
    Bridge model for spatial heat and height map data.

    Abstracts the 2D floating-point array and its mapping metadata.
    Note: The two-dimensional 'data' array must be scaled to the background image.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    width_px: int = Field(..., alias="width", validation_alias=AliasChoices("width", "width_px"))
    height_px: int = Field(..., alias="height", validation_alias=AliasChoices("height", "height_px"))
    data: list[list[float]] = Field(..., description="2D array of spatial metrics.")

`HistogramType`

Bases: str, Enum

Available histogram types for logic counters.

Source code in src/xovis/models/device.py

class HistogramType(str, Enum):
    """Available histogram types for logic counters."""

    PERSON_AGE = "PERSON_AGE"

`HistoryLogics`

Bases: BaseModel

Bridge model for historical logic data.

Provides a stable interface for time-series count data, abstracting away firmware-specific metadata structures.

Source code in src/xovis/models/device.py

class HistoryLogics(BaseModel):
    """
    Bridge model for historical logic data.

    Provides a stable interface for time-series count data, abstracting
    away firmware-specific metadata structures.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    begin: int | str
    end: int | str
    begin_data: int | str | None = None
    end_data: int | str | None = None
    resolution_ms: int | None = None
    number_of_bins: int | None = None
    measurements: list[HistoryMeasurement] = Field(default_factory=list)
    config: dict[str, Any] | None = None

`HistoryMeasurement`

Bases: BaseModel

Represents a single time-series bin in the historical data output.

Source code in src/xovis/models/device.py

class HistoryMeasurement(BaseModel):
    """
    Represents a single time-series bin in the historical data output.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    begin: int | str = Field(..., description="Start of the bin interval.")
    end: int | str = Field(..., description="End of the bin interval.")
    records: int = Field(..., description="Number of 1-minute records covered by this bin.")
    counts: list[dict[str, Any]] = Field(default_factory=list, description="List of counter values (id, value).")

`HistoryQuery`

Bases: BaseModel

Internal model used to validate and normalize historical data query parameters.

Source code in src/xovis/models/device.py

class HistoryQuery(BaseModel):
    """
    Internal model used to validate and normalize historical data query parameters.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    begin: XovisTime = Field(..., description="Start of the time range (Unix ms or relative).")
    end: XovisTime = Field(default="now", description="End of the time range (Unix ms or relative).")
    resolution_min: int = Field(default=0, description="Aggregation resolution in minutes.")
    time_format: TimeFormat = Field(default=TimeFormat.UNIX_TIME_MS)
    include_empty: bool = Field(default=False)

`HistoryStatus`

Bases: BaseModel

Bridge model for the historical data storage status.

Combines hardware capacity metrics and stored data metadata into a firmware-agnostic diagnostic object.

Source code in src/xovis/models/device.py

class HistoryStatus(BaseModel):
    """
    Bridge model for the historical data storage status.

    Combines hardware capacity metrics and stored data metadata into a
    firmware-agnostic diagnostic object.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    capacity: StorageCapacity = Field(..., alias="storage_capacity", validation_alias=AliasChoices("capacity", "storage"))
    stored_data: StoredData = Field(..., description="Details about current data on flash.")

    @model_validator(mode="before")
    @classmethod
    def _flatten_storage(cls, data: Any) -> Any:
        """Handles cases where capacity is nested under 'storage'."""
        if isinstance(data, dict) and "storage" in data and "capacity" in data["storage"]:
            # firmware v5 style
            storage = data["storage"]
            return {"capacity": storage["capacity"], "stored_data": storage.get("stored_data", {})}
        return data

Methods:

`_flatten_storage(data)` `classmethod`

Handles cases where capacity is nested under 'storage'.

Source code in src/xovis/models/device.py

@model_validator(mode="before")
@classmethod
def _flatten_storage(cls, data: Any) -> Any:
    """Handles cases where capacity is nested under 'storage'."""
    if isinstance(data, dict) and "storage" in data and "capacity" in data["storage"]:
        # firmware v5 style
        storage = data["storage"]
        return {"capacity": storage["capacity"], "stored_data": storage.get("stored_data", {})}
    return data

`IntervalType`

Bases: str, Enum

Discrete time intervals for data push.

Source code in src/xovis/models/device.py

class IntervalType(str, Enum):
    """Discrete time intervals for data push."""

    ONE_DAY = "ONE_DAY"
    ONE_HOUR = "ONE_HOUR"
    FIFTEEN_MINUTES = "FIFTEEN_MINUTES"
    FIVE_MINUTES = "FIVE_MINUTES"
    ONE_MINUTE = "ONE_MINUTE"
    THIRTY_SECONDS = "THIRTY_SECONDS"
    FIVE_SECONDS = "FIVE_SECONDS"

`Layer`

Bases: BaseModel

Represents a virtual counting layer in the Xovis scene.

Layers are used to group logics and define the spatial Zone of Interest (ZOI).

Source code in src/xovis/models/device.py

class Layer(BaseModel):
    """
    Represents a virtual counting layer in the Xovis scene.

    Layers are used to group logics and define the spatial Zone of Interest (ZOI).
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    id: int = Field(description="Unique identifier for the layer.")
    name: str = Field(description="Human-readable name of the layer.")
    zone_of_interest: list[tuple[float, float]] = Field(
        alias="zoi",
        default_factory=list,
        description="The spatial polygon defining the area of interest for this layer.",
    )

`Line`

Bases: BaseModel

Represents a spatial Line geometry used for crossing-based analytics.

Attributes:

Name	Type	Description
`id`	`int`	Unique identifier for the line.
`name`	`str`	Human-readable topological name of the line.
`coordinates`	`List[Tuple[float, float]]`	The line segments coordinates.

Source code in src/xovis/models/device.py

class Line(BaseModel):
    """
    Represents a spatial Line geometry used for crossing-based analytics.

    Attributes:
        id (int): Unique identifier for the line.
        name (str): Human-readable topological name of the line.
        coordinates (List[Tuple[float, float]]): The line segments coordinates.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    id: int
    name: str
    coordinates: list[tuple[float, float]] = Field(
        alias="geometry",
        default_factory=list,
        description="List of (x, y) coordinates forming the line segments.",
    )

`Logic`

Bases: BaseModel

Defines a counting or analytics logic applied to a geometry.

Bridges Logic1, LogicStatus, and LogicTemplate into a stable interface. Supports Custom Logic via associated Counters and Modifiers.

API Tip: When creating complex custom logics, use '?id_mode=CLIENT' on the Counter/Modifier endpoints to force the sensor to respect your provided IDs, ensuring the Web UI renders them correctly.

Source code in src/xovis/models/device.py

class Logic(BaseModel):
    """
    Defines a counting or analytics logic applied to a geometry.

    Bridges Logic1, LogicStatus, and LogicTemplate into a stable interface.
    Supports Custom Logic via associated Counters and Modifiers.

    API Tip: When creating complex custom logics, use '?id_mode=CLIENT' on the
    Counter/Modifier endpoints to force the sensor to respect your provided IDs,
    ensuring the Web UI renders them correctly.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    id: int = Field(description="Unique identifier for the logic.")
    name: str = Field(description="Human-readable name of the logic.")
    type: LogicType = Field(description="The template type of this logic.")
    layer_id: int | None = Field(default=None, description="The ID of the virtual counting layer.")
    optional_data: str | None = Field(default=None, description="Associated metadata or user-defined data.")

    def model_dump(self, **kwargs):
        """Ensures LogicType is serialized as a string value."""
        kwargs.setdefault("mode", "json")
        return super().model_dump(**kwargs)

Methods:

`model_dump(**kwargs)`

Ensures LogicType is serialized as a string value.

Source code in src/xovis/models/device.py

def model_dump(self, **kwargs):
    """Ensures LogicType is serialized as a string value."""
    kwargs.setdefault("mode", "json")
    return super().model_dump(**kwargs)

`LogicType`

Bases: Enum

Standardized logic template types for counting and analytics.

Normalizes the diverse set of template strings (e.g., XLT_LINE_IN_OUT_COUNT) into a stable enumeration.

Source code in src/xovis/models/device.py

class LogicType(Enum):
    """
    Standardized logic template types for counting and analytics.

    Normalizes the diverse set of template strings (e.g., XLT_LINE_IN_OUT_COUNT)
    into a stable enumeration.
    """

    CUSTOM = "XLT_CUSTOM"
    ZONE_OCCUPANCY = "XLT_ZONE_OCCUPANCY_COUNT"
    LINE_IN_OUT = "XLT_LINE_IN_OUT_COUNT"
    LINE_LATE = "XLT_LINE_LATE_COUNT"
    ZONE_IN_OUT = "XLT_ZONE_IN_OUT_COUNT"
    GROUP_LINE_IN_OUT = "XLT_GROUP_LINE_IN_OUT_COUNT"
    GROUP_LINE_LATE = "XLT_GROUP_LINE_LATE_COUNT"
    BICYCLE_LINE_IN_OUT = "XLT_BICYCLE_LINE_IN_OUT_COUNT"
    BICYCLE_LINE_LATE = "XLT_BICYCLE_LINE_LATE_COUNT"
    PRAM_LINE_IN_OUT = "XLT_PRAM_LINE_IN_OUT_COUNT"
    PRAM_LINE_LATE = "XLT_PRAM_LINE_LATE_COUNT"
    WHEELCHAIR_LINE_IN_OUT = "XLT_WHEELCHAIR_LINE_IN_OUT_COUNT"
    WHEELCHAIR_LINE_LATE = "XLT_WHEELCHAIR_LINE_LATE_COUNT"
    SHOPPING_CART_LINE_IN_OUT = "XLT_SHOPPING_CART_LINE_IN_OUT_COUNT"
    SHOPPING_CART_LINE_LATE = "XLT_SHOPPING_CART_LINE_LATE_COUNT"
    ZONE_DOOR = "XLT_ZONE_DOOR_COUNT"
    QUEUE_STATISTICS = "XLT_QUEUE_STATISTICS"
    WRONG_WAY_DETECTION = "XLT_WRONG_WAY_DETECTION"
    # Legacy/v4-compatible templates
    LEGACY_LINE_IN_OUT = "XLT_4X_LINE_IN_OUT_COUNT"
    LEGACY_LINE_LATE = "XLT_4X_LINE_LATE_COUNT"
    LEGACY_ZONE_COUNT = "XLT_4X_ZONE_COUNT"

`MQTTConfig`

Bases: BaseModel

Configuration for MQTT(S) data push connections.

Source code in src/xovis/models/device.py

class MQTTConfig(BaseModel):
    """Configuration for MQTT(S) data push connections."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    uri: str
    topic: str
    port: Optional[int] = None
    auth_enable: bool = Field(default=False)
    user: Optional[str] = None
    password: Optional[str] = None
    ssl_enable: bool = Field(default=False)
    qos_level: int = Field(default=0)
    websocket_enable: bool = Field(default=False)
    client_id: Optional[str] = None
    connection_timeout_s: float = Field(default=0.0)

`Modifier`

Bases: BaseModel

Defines the precise conditions (Modifiers) for triggering counts in a Custom Logic.

Hardware Limit: Sensors typically handle ~80 Modifiers per layer. Exceeding this will return a 'Max number of modifiers reached' error.

Source code in src/xovis/models/device.py

class Modifier(BaseModel):
    """
    Defines the precise conditions (Modifiers) for triggering counts in a Custom Logic.

    Hardware Limit: Sensors typically handle ~80 Modifiers per layer. Exceeding this
    will return a 'Max number of modifiers reached' error.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    id: int | None = Field(None, description="System-assigned or CLIENT-forced unique identifier.")
    name: str | None = Field(None, description="Optional description of the modifier.")
    logic_id: int | None = Field(None, description="Parent logic identifier.")
    object_type: ObjectType = Field(default=ObjectType.PERSON)
    trigger: TriggerType | dict[str, Any] = Field(..., description="The event that triggers evaluation.")
    count_events: list[CountEvent] = Field(default_factory=list)
    filter: list[Filter] = Field(default_factory=list, description="RPN filter stack.")
    zone_of_interest: int | None = Field(alias="zoi", default=None, description="Optional linked geometry ID.")

    def model_dump(self, **kwargs):
        kwargs.setdefault("mode", "json")
        data = super().model_dump(**kwargs)
        # Ensure filters are also flattened if they were dumped recursively
        if "filter" in data and isinstance(data["filter"], list):
            # Modifier contains a list of Filters. Each Filter's model_dump
            # flattens its payload. If Pydantic's recursive dump didn't use our override,
            # we check for 'payload' and flatten it here.
            flattened_filters = []
            for f in data["filter"]:
                if isinstance(f, dict) and "payload" in f:
                    payload = f.pop("payload", {})
                    if isinstance(payload, dict):
                        f.update(payload)
                flattened_filters.append(f)
            data["filter"] = flattened_filters
        return data

`ObjectType`

Bases: Enum

Standardized object classifications for Xovis sensors.

Bridges differences between firmware versions where ObjectType vs ObjectType1 definitions may vary.

Source code in src/xovis/models/device.py

class ObjectType(Enum):
    """
    Standardized object classifications for Xovis sensors.

    Bridges differences between firmware versions where ObjectType vs ObjectType1
    definitions may vary.
    """

    PERSON = "PERSON"
    GROUP = "GROUP"
    BICYCLE = "BICYCLE"
    PRAM = "PRAM"
    WHEELCHAIR = "WHEELCHAIR"
    SHOPPING_CART = "SHOPPING_CART"

`PathStitchingZone`

Bases: BaseModel

Represents a zone assigned to the Path Stitcher. Tracks lost in these zones are prolonged and potentially merged with new tracks.

Source code in src/xovis/models/device.py

class PathStitchingZone(BaseModel):
    """
    Represents a zone assigned to the Path Stitcher.
    Tracks lost in these zones are prolonged and potentially merged with new tracks.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    zone_id: int
    radius_mm: float = Field(alias="radius", default=1000.0, description="Max distance for merging tracks.")
    time_sec: float = Field(alias="time", default=2.0, description="Max duration to prolong lost tracks.")

`RetryConfig`

Bases: BaseModel

Configuration for data push retry logic.

Source code in src/xovis/models/device.py

class RetryConfig(BaseModel):
    """Configuration for data push retry logic."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    mode: RetryMode = Field(default=RetryMode.DROP)
    max_number: int = Field(default=0)
    reset_on_next_push_schedule: bool = Field(default=True)
    delay_start_min: float = Field(default=2.0)
    delay_start_max: float = Field(default=2.0)
    delay_interval_min: Optional[float] = None
    delay_interval_max: Optional[float] = None
    delay_increase_const: Optional[float] = None
    delay_increase_factor: Optional[float] = None

`RetryMode`

Bases: str, Enum

Strategies for handling transmission failures.

Source code in src/xovis/models/device.py

class RetryMode(str, Enum):
    """Strategies for handling transmission failures."""

    DROP = "DROP"
    INTERVAL = "INTERVAL"
    INCREASING_DELAY = "INCREASING_DELAY"
    INCREASING_DELAY_EXPONENTIAL = "INCREASING_DELAY_EXPONENTIAL"

`SFTPConfig`

Bases: BaseModel

Configuration for SFTP data push connections.

Source code in src/xovis/models/device.py

class SFTPConfig(BaseModel):
    """Configuration for SFTP data push connections."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    uri: str
    user: str
    password: str
    port: int = Field(default=22)
    path: Optional[str] = None
    host_key: Optional[str] = None
    create_directories: bool = Field(default=True)
    file_mode: FTPFileMode = Field(default=FTPFileMode.PACKAGE)
    max_file_size: int = Field(default=0)
    new_directory_permission: str = Field(default="rwxr-xr-x")
    new_file_permission: str = Field(default="rw-r--r--")
    ssh_compression_enable: bool = Field(default=True)
    connection_timeout_s: float = Field(default=2.0)
    ignore_proxy: bool = Field(default=False)

`SceneMask`

Bases: BaseModel

Represents a Scene Mask (e.g., Exclusion or Boarding) in the 3D environment.

Attributes:

Name	Type	Description
`id`	`int`	Unique identifier for the mask.
`type`	`SceneMaskType`	The functional type of the mask.
`coordinates`	`List[Tuple[float, float]]`	Bounding polygon coordinates.

Source code in src/xovis/models/device.py

class SceneMask(BaseModel):
    """
    Represents a Scene Mask (e.g., Exclusion or Boarding) in the 3D environment.

    Attributes:
        id (int): Unique identifier for the mask.
        type (SceneMaskType): The functional type of the mask.
        coordinates (List[Tuple[float, float]]): Bounding polygon coordinates.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    id: int
    type: SceneMaskType
    coordinates: list[tuple[float, float]] = Field(
        alias="polygon",
        default_factory=list,
        description="List of (x, y) coordinates forming the mask polygon. Max 15 scene masks per sensor context.",
    )

    def model_dump(self, **kwargs):
        """Ensures Enum values are serialized as strings even without by_alias=True."""
        kwargs.setdefault("mode", "json")
        return super().model_dump(**kwargs)

Methods:

`model_dump(**kwargs)`

Ensures Enum values are serialized as strings even without by_alias=True.

Source code in src/xovis/models/device.py

def model_dump(self, **kwargs):
    """Ensures Enum values are serialized as strings even without by_alias=True."""
    kwargs.setdefault("mode", "json")
    return super().model_dump(**kwargs)

`SceneMaskType`

Bases: Enum

Types of masks applied to the 3D scene.

Source code in src/xovis/models/device.py

class SceneMaskType(Enum):
    """Types of masks applied to the 3D scene."""

    BOARDING = "BOARDING"
    EXCLUSION = "EXCLUSION"
    LEGACY_EXCLUSION = "LEGACY_EXCLUSION"

`Scheduler`

Bases: BaseModel

Data push scheduling and retry policy.

Source code in src/xovis/models/device.py

class Scheduler(BaseModel):
    """Data push scheduling and retry policy."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    type: SchedulerType = Field(..., alias="type")
    interval: Optional[IntervalType] = None
    cron: Optional[str] = None
    retry: RetryConfig = Field(default_factory=RetryConfig)

    def model_dump(self, **kwargs) -> dict[str, Any]:
        # Always exclude none for scheduler to be safe with hardware schemas
        kwargs["exclude_none"] = True
        data = super().model_dump(**kwargs)
        if self.type != SchedulerType.INTERVAL:
            data.pop("interval", None)
        return data

`SchedulerType`

Bases: str, Enum

Scheduling strategies for data push.

Source code in src/xovis/models/device.py

class SchedulerType(str, Enum):
    """Scheduling strategies for data push."""

    INTERVAL = "INTERVAL"
    PERIODIC = "PERIODIC"
    IMMEDIATE = "IMMEDIATE"
    ADVANCED = "ADVANCED"

`StartStopPoints`

Bases: BaseModel

Bridge model for track start and stop coordinates.

Used by agents to optimize geometry placement by analyzing where tracks typically appear and vanish.

Source code in src/xovis/models/device.py

class StartStopPoints(BaseModel):
    """
    Bridge model for track start and stop coordinates.

    Used by agents to optimize geometry placement by analyzing
    where tracks typically appear and vanish.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    begin: int
    end: int
    start_points: list[list[float]] = Field(default_factory=list, description="List of [x, y, z] vectors.")
    stop_points: list[list[float]] = Field(default_factory=list, description="List of [x, y, z] vectors.")

`StartStopQuery`

Bases: BaseModel

Internal model used to validate and normalize start/stop points query parameters.

Source code in src/xovis/models/device.py

class StartStopQuery(BaseModel):
    """
    Internal model used to validate and normalize start/stop points query parameters.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    begin: XovisTime = Field(..., description="Start of the time range (Unix ms or relative).")
    end: XovisTime = Field(default="now", description="End of the time range (Unix ms or relative).")
    max: int = Field(default=1000, description="Maximum number of points.")
    points: bool = Field(default=True)

`StorageCapacity`

Bases: BaseModel

Hardware information related to data persistence capacity.

Source code in src/xovis/models/device.py

class StorageCapacity(BaseModel):
    """Hardware information related to data persistence capacity."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    memory: int = Field(..., description="Total memory in bytes.")
    count_records: int = Field(..., description="Maximum number of records.")
    time: str = Field(..., description="Remaining recording time as human readable string (e.g., 2y 126d).")
    time_s: int = Field(..., description="Remaining recording time in seconds.")
    fill_level_percent: float = Field(..., description="Percentage of used storage.")

`StoredData`

Bases: BaseModel

Status of the internally stored historical data.

Source code in src/xovis/models/device.py

class StoredData(BaseModel):
    """Status of the internally stored historical data."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    time_begin: str = Field(..., description="Timestamp of the oldest stored data (RFC3339).")
    time_end: str = Field(..., description="Timestamp of the latest stored data (RFC3339).")
    retention_time_s: int | None = None
    retention_time: str | None = None
    number_of_count_records: int
    oldest_count_record: StoredDataRecord | None = None
    newest_count_record: StoredDataRecord | None = None
    logics: list[dict[str, Any]] = Field(default_factory=list)
    counts: list[dict[str, Any]] = Field(default_factory=list)

`StoredDataRecord`

Bases: BaseModel

Metadata about a specific count record in the database.

Source code in src/xovis/models/device.py

class StoredDataRecord(BaseModel):
    """Metadata about a specific count record in the database."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    id: int
    time_begin: str
    duration_s: int
    data_size: int

`SystemInfo`

Bases: BaseModel

Core hardware and firmware details extracted during the Control Plane bootstrap phase.

Attributes:

Name	Type	Description
`mac_address`	`str`	The sensor's MAC address (mapped from network serial).
`sw_version`	`str`	The firmware version running on the edge device.
`serial_number`	`str`	The physical hardware identifier.

Source code in src/xovis/models/device.py

class SystemInfo(BaseModel):
    """
    Core hardware and firmware details extracted during the Control Plane bootstrap phase.

    Attributes:
        mac_address (str): The sensor's MAC address (mapped from network serial).
        sw_version (str): The firmware version running on the edge device.
        serial_number (str): The physical hardware identifier.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    mac_address: str = Field(alias="serial", default="", json_schema_extra={"ai_privacy": "HASH"})
    sw_version: str = Field(alias="fw_version", default="")
    serial_number: str = Field(alias="hw_id", default="")

`TCPConfig`

Bases: BaseModel

Configuration for TCP data push connections.

Source code in src/xovis/models/device.py

class TCPConfig(BaseModel):
    """Configuration for TCP data push connections."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    mode: TCPUDPMode = Field(default=TCPUDPMode.CLIENT)
    uri: Optional[str] = None
    port: Optional[int] = None
    connection_timeout_s: float = Field(default=2.0)

    def model_dump(self, **kwargs) -> dict[str, Any]:
        data = super().model_dump(**kwargs)
        # Force camelCase for hardware compatibility
        data["connectionTimeoutS"] = data.get("connection_timeout_s", 2.0)
        return data

`TCPUDPMode`

Bases: str, Enum

Operating modes for TCP and UDP connections.

Source code in src/xovis/models/device.py

class TCPUDPMode(str, Enum):
    """Operating modes for TCP and UDP connections."""

    CLIENT = "CLIENT"
    SERVER = "SERVER"
    LEGACY_EVENT_STREAM_SERVER = "LEGACY_EVENT_STREAM_SERVER"
    LEGACY_OBJECT_STREAM_SERVER = "LEGACY_OBJECT_STREAM_SERVER"

`TimeFormat`

Bases: str, Enum

Supported time formats for historical data serialization.

Source code in src/xovis/models/device.py

class TimeFormat(str, Enum):
    """Supported time formats for historical data serialization."""

    UNIX_TIME_MS = "UNIX_TIME_MS"
    UNIX_TIME_S = "UNIX_TIME_S"
    RFC3339 = "RFC3339"

`TransmitStatus`

Bases: BaseModel

Detailed statistics for data transmission.

Source code in src/xovis/models/device.py

class TransmitStatus(BaseModel):
    """Detailed statistics for data transmission."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    protocol: Optional[str] = None
    server: Optional[str] = None
    port: Optional[int] = None
    time: Optional[str] = None
    duration_ms: Optional[int] = None
    size_b: Optional[int] = None
    size: Optional[str] = None
    speed_bps: Optional[int] = None
    speed: Optional[str] = None
    code: Optional[int] = None
    info: Optional[str] = None

`TriggerType`

Bases: Enum

Timing and evaluation triggers for logic modifiers.

Source code in src/xovis/models/device.py

class TriggerType(Enum):
    """Timing and evaluation triggers for logic modifiers."""

    TRACK_DELETED = "track_deleted"
    LINE_CROSS_FORWARD = "line_cross_forward"
    LINE_CROSS_BACKWARD = "line_cross_backward"
    ZONE_ENTRY = "zone_entry"
    ZONE_EXIT = "zone_exit"
    DWELL_TIME_REACHED = "dwell_time_reached"

`UDPConfig`

Bases: BaseModel

Configuration for UDP data push connections.

Source code in src/xovis/models/device.py

class UDPConfig(BaseModel):
    """Configuration for UDP data push connections."""

    model_config = ConfigDict(populate_by_name=True, extra="ignore")
    mode: TCPUDPMode = Field(default=TCPUDPMode.CLIENT)
    uri: Optional[str] = None
    port: Optional[int] = None
    connection_timeout_s: float = Field(default=2.0)

    def model_dump(self, **kwargs) -> dict[str, Any]:
        data = super().model_dump(**kwargs)
        # Force camelCase for hardware compatibility
        data["connectionTimeoutS"] = data.get("connection_timeout_s", 2.0)
        return data

`ViewMask`

Bases: BaseModel

Represents a View Mask (e.g., Taboo or Illumination) on the sensor's image plane.

Attributes:

Name	Type	Description
`id`	`int`	Unique identifier for the mask.
`type`	`ViewMaskType`	The functional type of the mask.
`coordinates`	`List[Tuple[float, float]]`	Bounding polygon coordinates.

Source code in src/xovis/models/device.py

class ViewMask(BaseModel):
    """
    Represents a View Mask (e.g., Taboo or Illumination) on the sensor's image plane.

    Attributes:
        id (int): Unique identifier for the mask.
        type (ViewMaskType): The functional type of the mask.
        coordinates (List[Tuple[float, float]]): Bounding polygon coordinates.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    id: int
    type: ViewMaskType
    coordinates: list[tuple[float, float]] = Field(
        alias="polygon",
        default_factory=list,
        description="List of (x, y) coordinates forming the mask polygon. Max 15 view masks per sensor context.",
    )

    def model_dump(self, **kwargs):
        """Ensures Enum values are serialized as strings even without by_alias=True."""
        kwargs.setdefault("mode", "json")
        return super().model_dump(**kwargs)

Methods:

`model_dump(**kwargs)`

Ensures Enum values are serialized as strings even without by_alias=True.

Source code in src/xovis/models/device.py

def model_dump(self, **kwargs):
    """Ensures Enum values are serialized as strings even without by_alias=True."""
    kwargs.setdefault("mode", "json")
    return super().model_dump(**kwargs)

`ViewMaskType`

Bases: Enum

Types of masks applied to the sensor's 2D view projection.

Source code in src/xovis/models/device.py

class ViewMaskType(Enum):
    """Types of masks applied to the sensor's 2D view projection."""

    TABOO = "TABOO"
    VISIBLE_FLOOR = "VISIBLE_FLOOR"
    ILLUMINATION = "ILLUMINATION"

`Zone`

Bases: BaseModel

Represents a spatial Zone geometry mapped to a local Xovis sensor context.

Attributes:

Name	Type	Description
`id`	`int`	Unique identifier for the zone.
`name`	`str`	Human-readable topological name of the zone.
`coordinates`	`List[Tuple[float, float]]`	Bounding polygon coordinates.

Source code in src/xovis/models/device.py

class Zone(BaseModel):
    """
    Represents a spatial Zone geometry mapped to a local Xovis sensor context.

    Attributes:
        id (int): Unique identifier for the zone.
        name (str): Human-readable topological name of the zone.
        coordinates (List[Tuple[float, float]]): Bounding polygon coordinates.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    id: int
    name: str
    coordinates: list[tuple[float, float]] = Field(
        alias="polygon",
        default_factory=list,
        description="List of (x, y) coordinates forming the zone polygon.",
    )

Functions:

`xovis.models.hub_device`

Xovis SDK - Hub Device Models

Operates within the Control Plane. Provides strictly validated Pydantic V2 RootModels and enumerations directly derived from the Xovis HUB Cloud OpenAPI specification, incorporating strict AI privacy tags.

Classes

`Device`

Bases: BaseModel

Comprehensive hardware, telemetry, and network state of a managed sensor.

Source code in src/xovis/models/hub_device.py

class Device(BaseModel):
    """Comprehensive hardware, telemetry, and network state of a managed sensor."""

    device_name: str | None = Field(None, examples=["Xovis Kitchen"], json_schema_extra={"ai_privacy": "HASH"})
    device_group: str | None = Field(None, examples=["Office Xovis"], json_schema_extra={"ai_privacy": "HASH"})
    customer: str | None = Field(None, examples=["Xovis"], json_schema_extra={"ai_privacy": "HASH"})
    categories: list[str] | None = None
    type: str | None = Field(None, examples=["PC2S"])
    id: DeviceId | None = Field(None, json_schema_extra={"ai_privacy": "HASH"})
    ip: str | None = Field(None, examples=["10.10.10.2"], json_schema_extra={"ai_privacy": "BLOCK"})
    firmware_version: str | None = Field(None, examples=["5.0.3-9738700b2d"])
    device_status: DeviceStatus | None = None
    tilt_measured_alpha_deg: float | None = Field(None, examples=[-1.4])
    tilt_measured_beta_deg: float | None = Field(None, examples=[-2.4])
    tilt_active_alpha_deg: float | None = Field(None, examples=[-1.3])
    tilt_active_beta_deg: float | None = Field(None, examples=[-2.3])
    mounting_height_m: float | None = Field(None, examples=[2.42])
    privacy_mode: int | None = Field(None, examples=[1])
    last_config_refresh: AwareDatetime | None = Field(None, examples=["2023-02-08T18:04:28Z"])

`DeviceId`

Bases: RootModel[MACAddress]

Strictly validated MAC Address wrapper.

Source code in src/xovis/models/hub_device.py

class DeviceId(RootModel[MACAddress]):
    """Strictly validated MAC Address wrapper."""

    root: MACAddress = Field(..., examples=["12:34:56:78:9A:BC"])

`DeviceState`

Bases: Enum

Lifecycle state of a device within the HUB.

Source code in src/xovis/models/hub_device.py

class DeviceState(Enum):
    """Lifecycle state of a device within the HUB."""

    MANAGED = "MANAGED"
    UNMANAGED = "UNMANAGED"

`DeviceStatus`

Bases: Enum

Network connection status of the edge sensor.

Source code in src/xovis/models/hub_device.py

class DeviceStatus(Enum):
    """Network connection status of the edge sensor."""

    ONLINE = "ONLINE"
    OFFLINE = "OFFLINE"

`DeviceUiAccess`

Bases: BaseModel

Temporary authenticated access token for tunneling into a remote sensor.

Source code in src/xovis/models/hub_device.py

class DeviceUiAccess(BaseModel):
    """Temporary authenticated access token for tunneling into a remote sensor."""

    device_ui_link: str | None = Field(
        None,
        examples=["https://sensor-connect.cloudapp.azure.com/api/tunnel/AA:BB/fullui?otp=ompOlGw..."],
    )

`DevicesCategoriesAssignment`

Bases: BaseModel

Payload for modifying organizational categories across multiple devices.

Source code in src/xovis/models/hub_device.py

class DevicesCategoriesAssignment(BaseModel):
    """Payload for modifying organizational categories across multiple devices."""

    device_ids: list[DeviceId]
    categories_to_add: list[str] | None = None
    categories_to_remove: list[str] | None = None

`DevicesCustomerAssignment`

Bases: BaseModel

Payload for executing bulk customer assignments across a fleet.

Source code in src/xovis/models/hub_device.py

class DevicesCustomerAssignment(BaseModel):
    """Payload for executing bulk customer assignments across a fleet."""

    device_ids: list[DeviceId]
    customer_name: str = Field(..., examples=["customer name"])

`DevicesRequest`

Bases: BaseModel

Standardized fleet array payload for bulk requests.

Source code in src/xovis/models/hub_device.py

class DevicesRequest(BaseModel):
    """Standardized fleet array payload for bulk requests."""

    device_ids: list[DeviceId]

`DevicesResponse`

Bases: BaseModel

Paginated or bulk array response containing managed device states.

Source code in src/xovis/models/hub_device.py

class DevicesResponse(BaseModel):
    """Paginated or bulk array response containing managed device states."""

    items: list[Device] | None = None

`HubDevice`

Bases: BaseModel

Represents a managed edge sensor provisioned within a Xovis HUB Cloud tenant.

Source code in src/xovis/models/hub_device.py

class HubDevice(BaseModel):
    """
    Represents a managed edge sensor provisioned within a Xovis HUB Cloud tenant.
    """

    model_config = ConfigDict(populate_by_name=True, extra="ignore")

    device_id: str = Field(alias="deviceId", json_schema_extra={"ai_privacy": "BLOCK"})
    state: str = Field(description="Connection state, e.g., 'MANAGED' or 'OFFLINE'")
    sw_version: str = Field(alias="swVersion", default="Unknown")
    customer_name: str | None = Field(alias="customerName", default=None, json_schema_extra={"ai_privacy": "BLOCK"})

`Uuid`

Bases: RootModel[UUID]

Strictly validated UUID payload.

Source code in src/xovis/models/hub_device.py

class Uuid(RootModel[UUID]):
    """Strictly validated UUID payload."""

    root: UUID = Field(..., examples=["c7b27fe9-53d8-43f4-8654-c37effeb8908"])

Dynamic Type Safety (`xovis_types`)

While the core models provide structure, the SDK supports Dynamic Type Generation to provide literal-level safety for your specific environment.

Why use dynamic types?

Because every Xovis installation is unique (with different Agent names, Zone IDs, and Line configurations), static SDK code cannot know your specific topology. By generating a local xovis_types.py module, you gain:

IDE Autocomplete: See your actual sensor names in your editor.
Static Validation: Tools like mypy or pyright can catch invalid references before you run the code.
Strict Literals: Enforce that only existing zones or agents are used in your logic.

How to generate

Use the Xovis CLI to probe a live sensor and build your local type definitions:

xovis-cli generate-types --host <SENSOR_IP>

This will create src/xovis/models/xovis_types.py. Note that this file is environment-specific and is typically excluded from version control to prevent conflicts across different installations.

Utilities

`xovis.utils.time`

Xovis SDK - Time Utilities

Provides high-performance, zero-dependency time parsing and normalization for Xovis-specific time formats, including relative offsets and Unix milliseconds.

Attributes

`XovisTime = Annotated[Union[int, str, datetime], BeforeValidator(_parse_relative_time)]` `module-attribute`

Annotated type for Xovis-compliant time inputs.

Accepts Unix milliseconds (int), datetime objects, ISO 8601 strings, or relative time strings (e.g., 'now', '-1h', '-30d'). Normalizes all inputs to Unix milliseconds (int) during Pydantic validation.

Functions:

`_parse_relative_time(value, _now=None)`

Parses raw ms, datetime, or strings (ISO 8601, relative) to Unix ms in UTC.

Returns:

Name	Type	Description
`int`	`int`	Unix timestamp in milliseconds.

Source code in src/xovis/utils/time.py

def _parse_relative_time(value: Any, _now: Optional[float] = None) -> int:
    """Parses raw ms, datetime, or strings (ISO 8601, relative) to Unix ms in UTC.

    Returns:
        int: Unix timestamp in milliseconds.
    """
    if not value:
        return 0

    result_int = 0
    if isinstance(value, int):
        result_int = value
    elif isinstance(value, datetime):
        # We ensure it's UTC or convert to UTC if it has timezone info
        if value.tzinfo is None:
            # Naive datetimes are assumed to be in system local time by .timestamp()
            # but for SDK consistency with relative 'now', we keep it as is.
            result_int = int(value.timestamp() * 1000)
        else:
            result_int = int(value.timestamp() * 1000)
    elif isinstance(value, str):
        val_str = value.strip()
        if val_str.isdigit() or (val_str.startswith("-") and val_str[1:].isdigit()):
            result_int = int(val_str)
        else:
            match = _TIME_REGEX.match(val_str)
            if match:
                now_ts = _now if _now is not None else time.time()
                if match.group(1) == "now":
                    result_int = int(now_ts * 1000)
                else:
                    amount, unit = int(match.group(2)), match.group(3)
                    result_int = int(now_ts * 1000) - (amount * _MULTIPLIERS[unit])
            else:
                try:
                    dt = datetime.fromisoformat(val_str.replace("Z", "+00:00"))
                    result_int = int(dt.timestamp() * 1000)
                except ValueError:
                    raise ValueError(f"Invalid Xovis time format: {value}")
    else:
        raise ValueError(f"Expected int, datetime, or str, got {type(value)}")

    return result_int

`xovis.utils.privacy`

Xovis SDK - AI Privacy Engine

Operates at the DX (Developer Experience) boundary. Provides a high-performance, recursive sanitization utility for scrubbing sensitive fields from Pydantic models before they are exposed to LLMs.

Classes

`AIPrivacySession`

Session-bound Privacy Engine for two-way identifier mapping.

Maintains a cryptographic mapping between real sensitive values (e.g., MAC addresses) and short, stable hashes used by the LLM.

Source code in src/xovis/utils/privacy.py

class AIPrivacySession:
    """
    Session-bound Privacy Engine for two-way identifier mapping.

    Maintains a cryptographic mapping between real sensitive values (e.g., MAC addresses)
    and short, stable hashes used by the LLM.
    """

    def __init__(self):
        # Unique salt per session to prevent rainbow table attacks
        self._salt = uuid.uuid4().hex.encode()
        self._hash_to_real: dict[str, str] = {}
        self._real_to_hash: dict[str, str] = {}

    def _generate_hash(self, prefix: str, real_value: str) -> str:
        """Generates a stable, session-bound hash for a given value."""
        if real_value in self._real_to_hash:
            return self._real_to_hash[real_value]

        # Short 8-char hash for LLM context efficiency
        digest = hashlib.sha256(self._salt + real_value.encode()).hexdigest()[:8]
        safe_hash = f"{prefix}_{digest}"

        self._real_to_hash[real_value] = safe_hash
        self._hash_to_real[safe_hash] = real_value
        return safe_hash

    def restore(self, data: Any) -> Any:
        """
        The Reverse Pass: Restores real values from LLM-provided hashes.
        """
        if isinstance(data, list):
            return [self.restore(item) for item in data]
        if isinstance(data, dict):
            return {k: self.restore(v) for k, v in data.items()}
        if isinstance(data, str) and data in self._hash_to_real:
            return self._hash_to_real[data]

        return data

    def deanonymize_text(self, text: str) -> str:
        """
        Native Post-Processing: Replaces all LLM-facing hashes in a text
        block with their real plaintext values.
        """
        if not text:
            return text

        for hashed_val, real_val in self._hash_to_real.items():
            text = text.replace(hashed_val, str(real_val))

        return text

    def sanitize(self, data: Any) -> Any:
        """
        The Forward Pass: Scrubs or hashes fields based on Pydantic metadata.
        """
        if isinstance(data, (str, int, float, bool)) or data is None:
            return data

        if isinstance(data, list):
            return [self.sanitize(item) for item in data]

        if isinstance(data, BaseModel):
            return self._sanitize_model(data)

        if isinstance(data, dict):
            # Optimize: Only recurse if the dict contains potentially sensitive values
            return {k: self.sanitize(v) for k, v in data.items()}

        return data

    def _sanitize_model(self, model: BaseModel) -> dict[str, Any]:
        """
        Extracts and sanitizes a single Pydantic model based on metadata.
        """
        # Optimize: exclude_unset=True reduces the dictionary size significantly
        raw_dict = model.model_dump(mode="json", by_alias=True, exclude_unset=True)
        sanitized = {}

        for field_name, field_info in model.__class__.model_fields.items():
            alias = field_info.alias or field_name
            if alias not in raw_dict:
                continue

            value = raw_dict[alias]
            extra = field_info.json_schema_extra or {}
            privacy_rule = extra.get("ai_privacy") if isinstance(extra, dict) else None

            if privacy_rule == "BLOCK":
                continue

            if privacy_rule == "HASH" and (isinstance(value, str) or (isinstance(value, dict) and "root" in value)):
                # Handle RootModels or nested dicts that might be hashed
                str_val = str(value["root"]) if isinstance(value, dict) and "root" in value else str(value)
                prefix = alias.split("_")[0].capitalize()
                sanitized[alias] = self._generate_hash(prefix, str_val)
            else:
                sanitized[alias] = self.sanitize(value)

        return sanitized

Methods:

`_generate_hash(prefix, real_value)`

Generates a stable, session-bound hash for a given value.

Source code in src/xovis/utils/privacy.py

def _generate_hash(self, prefix: str, real_value: str) -> str:
    """Generates a stable, session-bound hash for a given value."""
    if real_value in self._real_to_hash:
        return self._real_to_hash[real_value]

    # Short 8-char hash for LLM context efficiency
    digest = hashlib.sha256(self._salt + real_value.encode()).hexdigest()[:8]
    safe_hash = f"{prefix}_{digest}"

    self._real_to_hash[real_value] = safe_hash
    self._hash_to_real[safe_hash] = real_value
    return safe_hash

`_sanitize_model(model)`

Extracts and sanitizes a single Pydantic model based on metadata.

Source code in src/xovis/utils/privacy.py

def _sanitize_model(self, model: BaseModel) -> dict[str, Any]:
    """
    Extracts and sanitizes a single Pydantic model based on metadata.
    """
    # Optimize: exclude_unset=True reduces the dictionary size significantly
    raw_dict = model.model_dump(mode="json", by_alias=True, exclude_unset=True)
    sanitized = {}

    for field_name, field_info in model.__class__.model_fields.items():
        alias = field_info.alias or field_name
        if alias not in raw_dict:
            continue

        value = raw_dict[alias]
        extra = field_info.json_schema_extra or {}
        privacy_rule = extra.get("ai_privacy") if isinstance(extra, dict) else None

        if privacy_rule == "BLOCK":
            continue

        if privacy_rule == "HASH" and (isinstance(value, str) or (isinstance(value, dict) and "root" in value)):
            # Handle RootModels or nested dicts that might be hashed
            str_val = str(value["root"]) if isinstance(value, dict) and "root" in value else str(value)
            prefix = alias.split("_")[0].capitalize()
            sanitized[alias] = self._generate_hash(prefix, str_val)
        else:
            sanitized[alias] = self.sanitize(value)

    return sanitized

`deanonymize_text(text)`

Native Post-Processing: Replaces all LLM-facing hashes in a text block with their real plaintext values.

Source code in src/xovis/utils/privacy.py

def deanonymize_text(self, text: str) -> str:
    """
    Native Post-Processing: Replaces all LLM-facing hashes in a text
    block with their real plaintext values.
    """
    if not text:
        return text

    for hashed_val, real_val in self._hash_to_real.items():
        text = text.replace(hashed_val, str(real_val))

    return text

`restore(data)`

The Reverse Pass: Restores real values from LLM-provided hashes.

Source code in src/xovis/utils/privacy.py

def restore(self, data: Any) -> Any:
    """
    The Reverse Pass: Restores real values from LLM-provided hashes.
    """
    if isinstance(data, list):
        return [self.restore(item) for item in data]
    if isinstance(data, dict):
        return {k: self.restore(v) for k, v in data.items()}
    if isinstance(data, str) and data in self._hash_to_real:
        return self._hash_to_real[data]

    return data

`sanitize(data)`

The Forward Pass: Scrubs or hashes fields based on Pydantic metadata.

Source code in src/xovis/utils/privacy.py

def sanitize(self, data: Any) -> Any:
    """
    The Forward Pass: Scrubs or hashes fields based on Pydantic metadata.
    """
    if isinstance(data, (str, int, float, bool)) or data is None:
        return data

    if isinstance(data, list):
        return [self.sanitize(item) for item in data]

    if isinstance(data, BaseModel):
        return self._sanitize_model(data)

    if isinstance(data, dict):
        # Optimize: Only recurse if the dict contains potentially sensitive values
        return {k: self.sanitize(v) for k, v in data.items()}

    return data

`xovis.utils.loop`

Xovis SDK - Event Loop Utilities

Provides utility functions for configuring the optimal asyncio event loop policy across different operating systems. This is critical for the Data Plane's high-throughput requirements, ensuring uvloop is utilized on Linux/macOS and ProactorEventLoop on Windows.

Functions:

`setup_optimal_loop()`

Configures the most performant asyncio event loop policy for the current platform.

On Windows, it ensures the use of ProactorEventLoop, which is required for high-performance TCP and subprocess operations. On Linux and macOS, it attempts to load and set uvloop as the global event loop policy.

Raises:

Type	Description
`ImportError`	Silently handled if `uvloop` or Windows-specific policies are unavailable.

Source code in src/xovis/utils/loop.py

def setup_optimal_loop():
    """
    Configures the most performant asyncio event loop policy for the current platform.

    On Windows, it ensures the use of `ProactorEventLoop`, which is required for
    high-performance TCP and subprocess operations. On Linux and macOS, it
    attempts to load and set `uvloop` as the global event loop policy.

    Raises:
        ImportError: Silently handled if `uvloop` or Windows-specific policies
            are unavailable.
    """
    if sys.platform == "win32":
        if sys.version_info < (3, 12):
            try:
                from asyncio import WindowsProactorEventLoopPolicy

                asyncio.set_event_loop_policy(WindowsProactorEventLoopPolicy())
                logger.debug("Configured WindowsProactorEventLoopPolicy")
            except ImportError:
                pass
        else:
            logger.debug("Python 3.12+ detected on Windows; using default ProactorEventLoop")
    else:
        try:
            import uvloop

            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
            logger.debug("Configured uvloop.EventLoopPolicy")
        except ImportError:
            logger.debug("uvloop not found, using default SelectorEventLoop")

Core SDK Reference

Architectural Pillars

Models

xovis.models.device

Classes

AgentConfig

BlockedSpace

CountAction

CountEvent

Counter

CounterName

CounterQuantity

CounterType

DataConfig

DataFormat

DataFormatType

DataPushAgent

DataPushAgentCollection

DataPushConnection

DataPushConnectionCollection

DataPushFilters

DataPushProtocol

DataPushStatus

DataPushStatusCollection

DataPushTestResponse

DataPushTriggerConfig

Methods:

_serialize_as_iso8601(value)

DataPushTriggerInfo

DataPushTriggerStatus

DataPushTriggerType

DataPushType

FTPConfig

FTPDirectoryMode

FTPFileMode

Filter

Methods:

model_dump(**kwargs)

FilterType

HTTPAuthMethod

HTTPConfig

HTTPHeaderField

HeatHeightMap

HistogramType

HistoryLogics

HistoryMeasurement

HistoryQuery

HistoryStatus

Methods:

_flatten_storage(data) classmethod

IntervalType

Layer

Line

Logic

Methods:

model_dump(**kwargs)

LogicType

MQTTConfig

Modifier

ObjectType

PathStitchingZone

RetryConfig

RetryMode

SFTPConfig

SceneMask

Methods:

model_dump(**kwargs)

SceneMaskType

Scheduler

SchedulerType

StartStopPoints

StartStopQuery

StorageCapacity

StoredData

StoredDataRecord

SystemInfo

TCPConfig

TCPUDPMode

TimeFormat

TransmitStatus

`xovis.models.device`

`AgentConfig`

`BlockedSpace`

`CountAction`

`CountEvent`

`Counter`

`CounterName`

`CounterQuantity`

`CounterType`

`DataConfig`

`DataFormat`

`DataFormatType`

`DataPushAgent`

`DataPushAgentCollection`

`DataPushConnection`

`DataPushConnectionCollection`

`DataPushFilters`

`DataPushProtocol`

`DataPushStatus`

`DataPushStatusCollection`

`DataPushTestResponse`

`DataPushTriggerConfig`

`_serialize_as_iso8601(value)`

`DataPushTriggerInfo`

`DataPushTriggerStatus`

`DataPushTriggerType`

`DataPushType`

`FTPConfig`

`FTPDirectoryMode`

`FTPFileMode`

`Filter`

`model_dump(**kwargs)`

`FilterType`

`HTTPAuthMethod`

`HTTPConfig`

`HTTPHeaderField`

`HeatHeightMap`

`HistogramType`

`HistoryLogics`

`HistoryMeasurement`

`HistoryQuery`

`HistoryStatus`

`_flatten_storage(data)` `classmethod`

`IntervalType`

`Layer`

`Line`

`Logic`

`model_dump(**kwargs)`

`LogicType`

`MQTTConfig`

`Modifier`

`ObjectType`

`PathStitchingZone`

`RetryConfig`

`RetryMode`

`SFTPConfig`

`SceneMask`

`model_dump(**kwargs)`

`SceneMaskType`

`Scheduler`

`SchedulerType`

`StartStopPoints`

`StartStopQuery`

`StorageCapacity`

`StoredData`

`StoredDataRecord`

`SystemInfo`

`TCPConfig`

`TCPUDPMode`

`TimeFormat`

`TransmitStatus`

`TriggerType`

`UDPConfig`

`ViewMask`

`model_dump(**kwargs)`

`ViewMaskType`

`Zone`

`xovis.models.hub_device`

`Device`

`DeviceId`