跳转至

数据结构

数据类型

BaseNumberType module-attribute

BaseNumberType: TypeAlias = Union[int, float]

ComponentId module-attribute

ComponentId: TypeAlias = Union[ComponentIntId, ComponentStrId]

ComponentIntId module-attribute

ComponentIntId: TypeAlias = int

ComponentStrId module-attribute

ComponentStrId: TypeAlias = str

DEFAULT_PYDANTIC_MODEL_CONFIG module-attribute

DEFAULT_PYDANTIC_MODEL_CONFIG = ConfigDict(strict=False, arbitrary_types_allowed=True, extra='ignore', validate_by_alias=True, validate_by_name=True, coerce_numbers_to_str=True, alias_generator=camelize, populate_by_name=True)

DEFAULT_UNIT_CONFIG module-attribute

DEFAULT_UNIT_CONFIG = TIME_UNIT_CONFIG | VOLUME_UNIT_CONFIG | LENGTH_UNIT_CONFIG

ErrorHandler module-attribute

ErrorHandler = Callable[[Exception], Any]

EventType module-attribute

EventType = NewType('EventType', str)

LENGTH_UNIT_CONFIG module-attribute

LENGTH_UNIT_CONFIG = UnitConfig(km=Fraction(1, 10 ** 3), m=1, dm=10, cm=100, mm=10 ** 3, nm=10 ** 9)

MODEL_CONFIG module-attribute

MODEL_CONFIG = ConfigDict(strict=False, arbitrary_types_allowed=True, extra='ignore', validate_by_alias=True, validate_by_name=True, coerce_numbers_to_str=True, alias_generator=camelize, populate_by_name=True)

Num module-attribute

Num = TypeVar('T', bound=Number)

NumberType module-attribute

NumberType: TypeAlias = Union[Decimal, int, float, Fraction]

Predicate module-attribute

Predicate: TypeAlias = Callable[[T], bool]

Quantity module-attribute

Quantity = Quantity

RatioComputer module-attribute

RatioComputer: TypeAlias = Callable[[Unit, Unit], Decimal]

SpeedVector module-attribute

SpeedVector = Tuple[Optional[Speed], Optional[Speed], Optional[Speed]]

StringNumber module-attribute

StringNumber: TypeAlias = str

T module-attribute

T = TypeVar('T')

TIME_UNIT_CONFIG module-attribute

TIME_UNIT_CONFIG = UnitConfig(=1, 小时=24, 分钟=24 * 60, =24 * 60 * 60, 毫秒=24 * 60 * 60 * 10 ** 3)

TaskType module-attribute

TaskType = NewType('TaskType', str)

TmpId module-attribute

TmpId: TypeAlias = Union[int, str]

Unit module-attribute

Unit = NewType('Unit', str)

V module-attribute

V = TypeVar('V')

VOLUME_UNIT_CONFIG module-attribute

VOLUME_UNIT_CONFIG = UnitConfig(kl=Fraction(1, 10 ** 3), l=1, dl=10, cl=100, ml=10 ** 3)

__doc__ module-attribute

__doc__ = '\n类型模块\n'

ureg module-attribute

ureg = UnitRegistry()

ABCEnumMeta

适用于enum类继承但是提示metaclass冲突的情况 1. 置顶metaclass为本类即可

源代码位于: logis/data_type/base.py
class ABCEnumMeta(EnumMeta, ABCMeta):
    """
    适用于enum类继承但是提示metaclass冲突的情况
    1. 置顶metaclass为本类即可
    """

    pass

ApiError

源代码位于: logis/data_type/net.py
class ApiError(BaseModel):
    code: int = -1
    message: str = "错误"

code class-attribute instance-attribute

code: int = -1

message class-attribute instance-attribute

message: str = '错误'

ApiResponse

源代码位于: logis/data_type/net.py
class ApiResponse(BaseModel, Generic[T]):

    success: bool
    data: Optional[T] = None
    error: Union[ApiError, Optional[str]] = None
    message: Optional[str] = None
    extra: Optional[dict] = None

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

    @staticmethod
    def positive(data: T, **kwargs):
        return ApiResponse[T](success=True, data=data, **kwargs)

    @staticmethod
    def negative(error: Union[ApiError, str], **kwargs):
        return ApiResponse[None](success=False, error=error, **kwargs)

    @staticmethod
    def from_http_response(
        r: requests.Response, content_type: Optional[Literal["json", "text"]] = None
    ):
        ct = content_type or r.headers.get("content-type") or ""
        try:
            data = r.json() if "json" in ct else r.text
        except Exception as e:
            logging.warning("解析响应体失败: %s", e)
            data = r.text
        return (
            ApiResponse.positive(data)
            if r.status_code == 200
            else ApiResponse.negative(
                ApiError(code=r.status_code, message=str(r.status_code))
            )
        )

data class-attribute instance-attribute

data: Optional[T] = None

error class-attribute instance-attribute

error: Union[ApiError, Optional[str]] = None

extra class-attribute instance-attribute

extra: Optional[dict] = None

message class-attribute instance-attribute

message: Optional[str] = None

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

success instance-attribute

success: bool

from_http_response staticmethod

from_http_response(r: Response, content_type: Optional[Literal['json', 'text']] = None)
源代码位于: logis/data_type/net.py
@staticmethod
def from_http_response(
    r: requests.Response, content_type: Optional[Literal["json", "text"]] = None
):
    ct = content_type or r.headers.get("content-type") or ""
    try:
        data = r.json() if "json" in ct else r.text
    except Exception as e:
        logging.warning("解析响应体失败: %s", e)
        data = r.text
    return (
        ApiResponse.positive(data)
        if r.status_code == 200
        else ApiResponse.negative(
            ApiError(code=r.status_code, message=str(r.status_code))
        )
    )

negative staticmethod

negative(error: Union[ApiError, str], **kwargs)
源代码位于: logis/data_type/net.py
@staticmethod
def negative(error: Union[ApiError, str], **kwargs):
    return ApiResponse[None](success=False, error=error, **kwargs)

positive staticmethod

positive(data: T, **kwargs)
源代码位于: logis/data_type/net.py
@staticmethod
def positive(data: T, **kwargs):
    return ApiResponse[T](success=True, data=data, **kwargs)

AppConfig

应用常用配置

源代码位于: logis/data_type/config.py
class AppConfig(BaseModel):
    """
    应用常用配置
    """

    app_name: str
    root_dir: Optional[Path] = None
    runtime_root_dir: Optional[Path] = None
    data_dir: Optional[Path] = None
    logs_dir: Optional[Path] = None
    sidecar_dir: Optional[Path] = None
    conf_dir: Optional[Path] = None

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

app_name instance-attribute

app_name: str

conf_dir class-attribute instance-attribute

conf_dir: Optional[Path] = None

data_dir class-attribute instance-attribute

data_dir: Optional[Path] = None

logs_dir class-attribute instance-attribute

logs_dir: Optional[Path] = None

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

root_dir class-attribute instance-attribute

root_dir: Optional[Path] = None

runtime_root_dir class-attribute instance-attribute

runtime_root_dir: Optional[Path] = None

sidecar_dir class-attribute instance-attribute

sidecar_dir: Optional[Path] = None

CallableInput

源代码位于: logis/data_type/base.py
class CallableInput(BaseModel):
    args: Tuple = Field(default_factory=tuple)
    kwargs: Dict[str, Any] = Field(default_factory=dict)

    @staticmethod
    def of(*args, **kwargs):
        return CallableInput(args=args, kwargs=kwargs)

args class-attribute instance-attribute

args: Tuple = Field(default_factory=tuple)

kwargs class-attribute instance-attribute

kwargs: Dict[str, Any] = Field(default_factory=dict)

of staticmethod

of(*args, **kwargs)
源代码位于: logis/data_type/base.py
@staticmethod
def of(*args, **kwargs):
    return CallableInput(args=args, kwargs=kwargs)

Capacity

源代码位于: logis/data_type/unitable.py
class Capacity(QuantifiedValue):
    pass

ConfigItem

源代码位于: logis/data_type/config.py
class ConfigItem(BaseModel, Generic[V]):
    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

    namespace: str = ""

    label: Union[str, None] = None
    key: str
    value: Optional[V] = None
    value_type: Optional[str] = None
    disabled: bool = False

disabled class-attribute instance-attribute

disabled: bool = False

key instance-attribute

key: str

label class-attribute instance-attribute

label: Union[str, None] = None

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

namespace class-attribute instance-attribute

namespace: str = ''

value class-attribute instance-attribute

value: Optional[V] = None

value_type class-attribute instance-attribute

value_type: Optional[str] = None

Data

源代码位于: logis/data_type/base.py
class Data(BaseModel, Generic[T]):
    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG
    type: str
    value: T
    tags: Optional[List[str]] = None
    description: Optional[str] = None

description class-attribute instance-attribute

description: Optional[str] = None

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

tags class-attribute instance-attribute

tags: Optional[List[str]] = None

type instance-attribute

type: str

value instance-attribute

value: T

EchartsConfig

源代码位于: logis/data_type/echarts.py
class EchartsConfig(BaseModel):
    option: dict = Field(default_factory=dict)
    attrs: dict = Field(default_factory=dict)

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

    def add_class(self, *class_list: str):
        self.attrs.setdefault("class", []).extend(class_list)
        return self

    def add_css_style(self, *pairs: Tuple[str, Any]):
        self.attrs.setdefault("style", {}).update({k: v for k, v in pairs})
        return self

attrs class-attribute instance-attribute

attrs: dict = Field(default_factory=dict)

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

option class-attribute instance-attribute

option: dict = Field(default_factory=dict)

add_class

add_class(*class_list: str)
源代码位于: logis/data_type/echarts.py
def add_class(self, *class_list: str):
    self.attrs.setdefault("class", []).extend(class_list)
    return self

add_css_style

add_css_style(*pairs: Tuple[str, Any])
源代码位于: logis/data_type/echarts.py
def add_css_style(self, *pairs: Tuple[str, Any]):
    self.attrs.setdefault("style", {}).update({k: v for k, v in pairs})
    return self

GenericPoint

源代码位于: logis/data_type/point.py
class GenericPoint(Generic[Num]):
    __global_precision__: Optional[int] = None

    @property
    def precision(self) -> Optional[int]:
        """
        优先读取__precision_,否则读取__global_precision__
        """
        return self.__precision__ or self.__class__.__global_precision__

    @precision.setter
    def precision(self, value: Optional[int]):
        self.__precision__ = value

    @classmethod
    def model_validate(cls, v: Dict[str, Num]):
        if v is None:
            return None
        if isinstance(v, dict):
            return cls(**v)
        raise ValueError("only dict type is supported")

    def __init__(
        self,
        x: Optional[Num] = None,
        y: Optional[Num] = None,
        z: Optional[Num] = None,
        *args,
        precision: Optional[int] = None,
        **kwargs,
    ):
        """
        Args:
            x (T | None, optional): X coordinate. Defaults to None.
            y (T | None, optional): Y coordinate. Defaults to None.
            z (T | None, optional): Z coordinate. Defaults to None.
            precision (int | None, optional): Precision. Defaults to None.
        """
        self.precision = precision
        if (p := self.precision) is not None:
            assert p > 0, "precision must be greater than 0."
            x = round_if_not_none(x, p)
            y = round_if_not_none(y, p)
            z = round_if_not_none(z, p)
        self.x, self.y, self.z = x, y, z

    @classmethod
    def from_tuple(
        cls, tp: Tuple[Num], precision: Optional[int] = None, **kwargs
    ) -> "GenericPoint[Num]":
        return cls(*tp, precision=precision, **kwargs)

    def __add__(self, other: "GenericPoint[Num]") -> "GenericPoint[Num]":
        assert self.precision == other.precision, "Precision must be the same."
        x = compute("add", self.x, other.x)
        y = compute("add", self.y, other.y)
        z = compute("add", self.z, other.z)
        return self.__class__(x, y, z, precision=self.precision)

    def __sub__(self, other: "GenericPoint[Num]") -> "GenericPoint[Num]":
        assert self.precision == other.precision, "Precision must be the same."
        x = compute("subtract", self.x, other.x)
        y = compute("subtract", self.y, other.y)
        z = compute("subtract", self.z, other.z)
        return self.__class__(x, y, z, precision=self.precision)

    def __mul__(self, other: Number) -> "GenericPoint[Num]":
        assert other is not None, "Multiplier must not be None."
        x = compute("multiply", self.x, other) if self.x is not None else None
        y = compute("multiply", self.y, other) if self.y is not None else None
        z = compute("multiply", self.z, other) if self.z is not None else None
        return self.__class__(x, y, z, precision=self.precision)

    def __truediv__(self, other: Number) -> "GenericPoint[Num]":
        assert other is not None, "Divisor must not be None."
        x = compute("divide", self.x, other) if self.x is not None else None
        y = compute("divide", self.y, other) if self.y is not None else None
        z = compute("divide", self.z, other) if self.z is not None else None
        return self.__class__(x, y, z, precision=self.precision)

    def __eq__(self, value):
        if not isinstance(value, self.__class__):
            return False
        dv = 1 / (10**self.precision) if self.precision else None

        def is_acceptable(a: Optional[Number], b: Optional[Number]) -> bool:
            if dv is None:
                return a == b
            if a is None and b is None:
                return True
            if a is None or b is None:
                return False
            return abs(a - b) < dv

        return (
            is_acceptable(self.x, value.x)
            and is_acceptable(self.y, value.y)
            and is_acceptable(self.z, value.z)
            and self.precision == value.precision
            and self.unit == value.unit
        )

__global_precision__ class-attribute instance-attribute

__global_precision__: Optional[int] = None

precision instance-attribute property writable

precision: Optional[int] = precision

优先读取__precision_,否则读取__global_precision__

__add__

__add__(other: GenericPoint[Num]) -> GenericPoint[Num]
源代码位于: logis/data_type/point.py
def __add__(self, other: "GenericPoint[Num]") -> "GenericPoint[Num]":
    assert self.precision == other.precision, "Precision must be the same."
    x = compute("add", self.x, other.x)
    y = compute("add", self.y, other.y)
    z = compute("add", self.z, other.z)
    return self.__class__(x, y, z, precision=self.precision)

__eq__

__eq__(value)
源代码位于: logis/data_type/point.py
def __eq__(self, value):
    if not isinstance(value, self.__class__):
        return False
    dv = 1 / (10**self.precision) if self.precision else None

    def is_acceptable(a: Optional[Number], b: Optional[Number]) -> bool:
        if dv is None:
            return a == b
        if a is None and b is None:
            return True
        if a is None or b is None:
            return False
        return abs(a - b) < dv

    return (
        is_acceptable(self.x, value.x)
        and is_acceptable(self.y, value.y)
        and is_acceptable(self.z, value.z)
        and self.precision == value.precision
        and self.unit == value.unit
    )

__init__

__init__(x: Optional[Num] = None, y: Optional[Num] = None, z: Optional[Num] = None, *args, precision: Optional[int] = None, **kwargs)

参数:

名称 类型 描述 默认
x T | None

X coordinate. Defaults to None.

None
y T | None

Y coordinate. Defaults to None.

None
z T | None

Z coordinate. Defaults to None.

None
precision int | None

Precision. Defaults to None.

None
源代码位于: logis/data_type/point.py
def __init__(
    self,
    x: Optional[Num] = None,
    y: Optional[Num] = None,
    z: Optional[Num] = None,
    *args,
    precision: Optional[int] = None,
    **kwargs,
):
    """
    Args:
        x (T | None, optional): X coordinate. Defaults to None.
        y (T | None, optional): Y coordinate. Defaults to None.
        z (T | None, optional): Z coordinate. Defaults to None.
        precision (int | None, optional): Precision. Defaults to None.
    """
    self.precision = precision
    if (p := self.precision) is not None:
        assert p > 0, "precision must be greater than 0."
        x = round_if_not_none(x, p)
        y = round_if_not_none(y, p)
        z = round_if_not_none(z, p)
    self.x, self.y, self.z = x, y, z

__mul__

__mul__(other: Number) -> GenericPoint[Num]
源代码位于: logis/data_type/point.py
def __mul__(self, other: Number) -> "GenericPoint[Num]":
    assert other is not None, "Multiplier must not be None."
    x = compute("multiply", self.x, other) if self.x is not None else None
    y = compute("multiply", self.y, other) if self.y is not None else None
    z = compute("multiply", self.z, other) if self.z is not None else None
    return self.__class__(x, y, z, precision=self.precision)

__sub__

__sub__(other: GenericPoint[Num]) -> GenericPoint[Num]
源代码位于: logis/data_type/point.py
def __sub__(self, other: "GenericPoint[Num]") -> "GenericPoint[Num]":
    assert self.precision == other.precision, "Precision must be the same."
    x = compute("subtract", self.x, other.x)
    y = compute("subtract", self.y, other.y)
    z = compute("subtract", self.z, other.z)
    return self.__class__(x, y, z, precision=self.precision)

__truediv__

__truediv__(other: Number) -> GenericPoint[Num]
源代码位于: logis/data_type/point.py
def __truediv__(self, other: Number) -> "GenericPoint[Num]":
    assert other is not None, "Divisor must not be None."
    x = compute("divide", self.x, other) if self.x is not None else None
    y = compute("divide", self.y, other) if self.y is not None else None
    z = compute("divide", self.z, other) if self.z is not None else None
    return self.__class__(x, y, z, precision=self.precision)

from_tuple classmethod

from_tuple(tp: Tuple[Num], precision: Optional[int] = None, **kwargs) -> GenericPoint[Num]
源代码位于: logis/data_type/point.py
@classmethod
def from_tuple(
    cls, tp: Tuple[Num], precision: Optional[int] = None, **kwargs
) -> "GenericPoint[Num]":
    return cls(*tp, precision=precision, **kwargs)

model_validate classmethod

model_validate(v: Dict[str, Num])
源代码位于: logis/data_type/point.py
@classmethod
def model_validate(cls, v: Dict[str, Num]):
    if v is None:
        return None
    if isinstance(v, dict):
        return cls(**v)
    raise ValueError("only dict type is supported")

InvokeResult

源代码位于: logis/data_type/base.py
class InvokeResult:
    def __init__(self):
        self.is_generator = False
        self.return_value: Any = None
        self.yield_values: Any = []

    @staticmethod
    def returns(value: Any):
        self = InvokeResult()
        self.return_value = value
        return self

is_generator instance-attribute

is_generator = False

return_value instance-attribute

return_value: Any = None

yield_values instance-attribute

yield_values: Any = []

__init__

__init__()
源代码位于: logis/data_type/base.py
def __init__(self):
    self.is_generator = False
    self.return_value: Any = None
    self.yield_values: Any = []

returns staticmethod

returns(value: Any)
源代码位于: logis/data_type/base.py
@staticmethod
def returns(value: Any):
    self = InvokeResult()
    self.return_value = value
    return self

Length

源代码位于: logis/data_type/unitable.py
class Length(QuantifiedValue):
    pass

MyBaseException

基础异常类,所有自定义异常都应该继承自这个类

源代码位于: logis/data_type/exception.py
class MyBaseException(Exception):
    """
    基础异常类,所有自定义异常都应该继承自这个类
    """

    def __init__(self, *args, **attrs):
        super().__init__(*args)
        for k, v in attrs.items():
            hasattr(self, k) or setattr(self, k, v)

__init__

__init__(*args, **attrs)
源代码位于: logis/data_type/exception.py
def __init__(self, *args, **attrs):
    super().__init__(*args)
    for k, v in attrs.items():
        hasattr(self, k) or setattr(self, k, v)

Notification

源代码位于: logis/data_type/notice.py
class Notification(BaseModel):
    title: Optional[str] = None
    duration: Union[float, int] = -1
    content: Optional[str] = None
    closable: bool = True
    type: Literal["info", "success", "warning", "error"] = "info"

closable class-attribute instance-attribute

closable: bool = True

content class-attribute instance-attribute

content: Optional[str] = None

duration class-attribute instance-attribute

duration: Union[float, int] = -1

title class-attribute instance-attribute

title: Optional[str] = None

type class-attribute instance-attribute

type: Literal['info', 'success', 'warning', 'error'] = 'info'

NumberUnit

具有数量和单位的数值类型

源代码位于: logis/data_type/unitable.py
class NumberUnit(metaclass=ABCMeta):
    """
    具有数量和单位的数值类型
    """

    quantity: NumberType = 0
    unit: Optional[Unit] = None

    # 自定义倍率转换器
    _unit_config_: Optional["UnitConfig"] = None

    def __init__(self, **kwargs):
        self.quantity = kwargs.get("quantity", 0)
        self.unit = kwargs.get("unit", None)

    @property
    def kind(self):
        """
        默认情况下返回类名,实际使用时可以重写
        """
        return self.__class__.__name__

    @abstractmethod
    def model_dump(self, *args, **kwargs) -> Dict[str, Any]:
        pass

    @classmethod
    @abstractmethod
    def model_validate(cls, data: Dict[str, Any], **kwargs) -> Self:
        pass

    def __auto_validate__(self, other: "NumberUnit"):
        if self._unit_config_ is None:
            assert (
                self.unit == other.unit
            ), f"unit must be the same, but {self.unit} != {other.unit}"
        else:
            other = unify_quantified_value(
                other, self.unit, unit_config=self._unit_config_
            )
        return other

    def __sub__(self, other: "NumberUnit") -> Self:
        other = self.__auto_validate__(other)
        dc = self.model_dump()
        dc["quantity"] = self.quantity - other.quantity
        return type(self).model_validate(dc)

    def __add__(self, other: "NumberUnit") -> Self:
        other = self.__auto_validate__(other)
        dc = self.model_dump()
        dc["quantity"] += other.quantity
        return type(self)(**dc)

    def __truediv__(self, other: "NumberUnit") -> Self:
        other = self.__auto_validate__(other)
        # TODO: 处理精度问题
        return self.quantity / other.quantity

    def __mul__(self, times: Union[int, float]) -> Self:
        data = self.model_dump()
        data["quantity"] *= times
        return type(self).model_validate(data)

    def __gt__(self, other: "NumberUnit"):
        other = self.__auto_validate__(other)
        return self.quantity > other.quantity

    def __lt__(self, other: "NumberUnit"):
        other = self.__auto_validate__(other)
        return self.quantity < other.quantity

    def __eq__(self, other: "NumberUnit"):
        if not other:
            return False
        other = self.__auto_validate__(other)
        return self.quantity == other.quantity

    def __ge__(self, other: "NumberUnit"):
        other = self.__auto_validate__(other)
        return self.quantity >= other.quantity

    def __le__(self, other: "NumberUnit"):
        other = self.__auto_validate__(other)
        return self.quantity <= other.quantity

    @property
    def value(self):
        return self.quantity

    @value.setter
    def value(self, new_value):
        self.quantity = new_value

    @classmethod
    def parse_tuple(cls, input: Tuple[NumberType, Unit]) -> Self:
        """
        解析形如[1,"个"]的数据
        """
        return cls(quantity=input[0], unit=input[1])

    @classmethod
    def parse_str(cls, input: str, delimiter="|", number_type=float) -> Optional[Self]:
        """
        解析形如1|m/s的数据
        """
        if not input:
            return None
        tmps = input.strip().split(delimiter)
        assert len(tmps) == 2, "unexpected format: " + input
        return cls(quantity=number_type(tmps[0]), unit=tmps[1])

    @classmethod
    def of(cls, num: NumberType, unit: Optional[str] = None) -> Self:
        return cls(quantity=num, unit=unit)

    def increase(self, num: NumberType):
        self.quantity += num

    def decrease(self, num: NumberType):
        self.quantity -= num

kind property

kind

默认情况下返回类名,实际使用时可以重写

quantity class-attribute instance-attribute

quantity: NumberType = get('quantity', 0)

unit class-attribute instance-attribute

unit: Optional[Unit] = get('unit', None)

value property writable

value

__add__

__add__(other: NumberUnit) -> Self
源代码位于: logis/data_type/unitable.py
def __add__(self, other: "NumberUnit") -> Self:
    other = self.__auto_validate__(other)
    dc = self.model_dump()
    dc["quantity"] += other.quantity
    return type(self)(**dc)

__auto_validate__

__auto_validate__(other: NumberUnit)
源代码位于: logis/data_type/unitable.py
def __auto_validate__(self, other: "NumberUnit"):
    if self._unit_config_ is None:
        assert (
            self.unit == other.unit
        ), f"unit must be the same, but {self.unit} != {other.unit}"
    else:
        other = unify_quantified_value(
            other, self.unit, unit_config=self._unit_config_
        )
    return other

__eq__

__eq__(other: NumberUnit)
源代码位于: logis/data_type/unitable.py
def __eq__(self, other: "NumberUnit"):
    if not other:
        return False
    other = self.__auto_validate__(other)
    return self.quantity == other.quantity

__ge__

__ge__(other: NumberUnit)
源代码位于: logis/data_type/unitable.py
def __ge__(self, other: "NumberUnit"):
    other = self.__auto_validate__(other)
    return self.quantity >= other.quantity

__gt__

__gt__(other: NumberUnit)
源代码位于: logis/data_type/unitable.py
def __gt__(self, other: "NumberUnit"):
    other = self.__auto_validate__(other)
    return self.quantity > other.quantity

__init__

__init__(**kwargs)
源代码位于: logis/data_type/unitable.py
def __init__(self, **kwargs):
    self.quantity = kwargs.get("quantity", 0)
    self.unit = kwargs.get("unit", None)

__le__

__le__(other: NumberUnit)
源代码位于: logis/data_type/unitable.py
def __le__(self, other: "NumberUnit"):
    other = self.__auto_validate__(other)
    return self.quantity <= other.quantity

__lt__

__lt__(other: NumberUnit)
源代码位于: logis/data_type/unitable.py
def __lt__(self, other: "NumberUnit"):
    other = self.__auto_validate__(other)
    return self.quantity < other.quantity

__mul__

__mul__(times: Union[int, float]) -> Self
源代码位于: logis/data_type/unitable.py
def __mul__(self, times: Union[int, float]) -> Self:
    data = self.model_dump()
    data["quantity"] *= times
    return type(self).model_validate(data)

__sub__

__sub__(other: NumberUnit) -> Self
源代码位于: logis/data_type/unitable.py
def __sub__(self, other: "NumberUnit") -> Self:
    other = self.__auto_validate__(other)
    dc = self.model_dump()
    dc["quantity"] = self.quantity - other.quantity
    return type(self).model_validate(dc)

__truediv__

__truediv__(other: NumberUnit) -> Self
源代码位于: logis/data_type/unitable.py
def __truediv__(self, other: "NumberUnit") -> Self:
    other = self.__auto_validate__(other)
    # TODO: 处理精度问题
    return self.quantity / other.quantity

decrease

decrease(num: NumberType)
源代码位于: logis/data_type/unitable.py
def decrease(self, num: NumberType):
    self.quantity -= num

increase

increase(num: NumberType)
源代码位于: logis/data_type/unitable.py
def increase(self, num: NumberType):
    self.quantity += num

model_dump abstractmethod

model_dump(*args, **kwargs) -> Dict[str, Any]
源代码位于: logis/data_type/unitable.py
@abstractmethod
def model_dump(self, *args, **kwargs) -> Dict[str, Any]:
    pass

model_validate abstractmethod classmethod

model_validate(data: Dict[str, Any], **kwargs) -> Self
源代码位于: logis/data_type/unitable.py
@classmethod
@abstractmethod
def model_validate(cls, data: Dict[str, Any], **kwargs) -> Self:
    pass

of classmethod

of(num: NumberType, unit: Optional[str] = None) -> Self
源代码位于: logis/data_type/unitable.py
@classmethod
def of(cls, num: NumberType, unit: Optional[str] = None) -> Self:
    return cls(quantity=num, unit=unit)

parse_str classmethod

parse_str(input: str, delimiter='|', number_type=float) -> Optional[Self]

解析形如1|m/s的数据

源代码位于: logis/data_type/unitable.py
@classmethod
def parse_str(cls, input: str, delimiter="|", number_type=float) -> Optional[Self]:
    """
    解析形如1|m/s的数据
    """
    if not input:
        return None
    tmps = input.strip().split(delimiter)
    assert len(tmps) == 2, "unexpected format: " + input
    return cls(quantity=number_type(tmps[0]), unit=tmps[1])

parse_tuple classmethod

parse_tuple(input: Tuple[NumberType, Unit]) -> Self

解析形如[1,"个"]的数据

源代码位于: logis/data_type/unitable.py
@classmethod
def parse_tuple(cls, input: Tuple[NumberType, Unit]) -> Self:
    """
    解析形如[1,"个"]的数据
    """
    return cls(quantity=input[0], unit=input[1])

PageResult

源代码位于: logis/data_type/page.py
class PageResult(Pager, Generic[T]):
    items: List[T]

    @staticmethod
    def of_page(page: Pager, items: List[T], total: int) -> "PageResult[T]":
        return PageResult[T](
            items=items,
            total=total,
            page_at=page.page_at,
            page_base=page.page_base,
            page_size=page.page_size,
        )

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

items instance-attribute

items: List[T]

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

of_page staticmethod

of_page(page: Pager, items: List[T], total: int) -> PageResult[T]
源代码位于: logis/data_type/page.py
@staticmethod
def of_page(page: Pager, items: List[T], total: int) -> "PageResult[T]":
    return PageResult[T](
        items=items,
        total=total,
        page_at=page.page_at,
        page_base=page.page_base,
        page_size=page.page_size,
    )

Pager

源代码位于: logis/data_type/page.py
class Pager(BaseModel):
    page_base: Optional[int] = Field(
        default=0, validation_alias=AliasChoices("page_base", "pageBase")
    )
    page_size: int = Field(
        default=100, validation_alias=AliasChoices("page_size", "pageSize")
    )
    page_at: int = Field(
        default=0, validation_alias=AliasChoices("page_at", "pageAt", "page")
    )
    total: Optional[int] = None

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

    @property
    def offset(self) -> int:
        return max(0, self.page_at - self.page_base) * self.page_size

    @property
    def limit(self) -> int:
        return self.page_size

limit property

limit: int

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

offset property

offset: int

page_at class-attribute instance-attribute

page_at: int = Field(default=0, validation_alias=AliasChoices('page_at', 'pageAt', 'page'))

page_base class-attribute instance-attribute

page_base: Optional[int] = Field(default=0, validation_alias=AliasChoices('page_base', 'pageBase'))

page_size class-attribute instance-attribute

page_size: int = Field(default=100, validation_alias=AliasChoices('page_size', 'pageSize'))

total class-attribute instance-attribute

total: Optional[int] = None

PlatformInfo

源代码位于: logis/data_type/updater.py
7
8
9
class PlatformInfo(BaseModel):
    signature: Optional[str] = None
    url: str

signature class-attribute instance-attribute

signature: Optional[str] = None

url instance-attribute

url: str

Point

此类不通用,历史遗留,不建议使用,如有需要建议使用GenericPoint 支持x,y,z三维坐标点,也可当作二维坐标使用 TODO: 处理单位

源代码位于: logis/data_type/point.py
class Point(GenericPoint[float]):
    """
    此类不通用,历史遗留,不建议使用,如有需要建议使用GenericPoint
    支持x,y,z三维坐标点,也可当作二维坐标使用
    TODO: 处理单位
    """

    __counter = count(0, step=1)
    __global_precision__ = 3

    @classmethod
    def try_parse(cls, dc: dict, **kwargs) -> "Point":
        """
        尝试读取X、Y、Z坐标,返回一个Point对象
        """
        if not isinstance(dc, dict):
            raise ValueError("Expected a dictionary for Point parsing.")
        xyz = [dc.get(k, None) for k in ("X", "Y", "Z")]
        return Point(xyz, **kwargs)

    @classmethod
    def from_tuple(cls, args: Tuple[Number], **kwargs):
        """
        从元组创建点
        """
        return cls(args, **kwargs)

    @classmethod
    def of(
        cls,
        x: Optional[Number] = None,
        y: Optional[Number] = None,
        z: Optional[Number] = None,
        **kwargs,
    ) -> "Point":
        p = Point(x, y, z, **kwargs)
        return p

    def __init__(self, *args, precision: Optional[int] = None, **kwargs):
        """
        支持[x,y]、x,y、[x,y,z]、x,y,z传参赋值
        默认x=y=z=0

        TODO: 不应该自动转float、不应该自动保留两位小数
        """
        self._id = next(self.__counter)
        self.unit: Optional[str] = kwargs.get("unit", None)

        x: Optional[Number] = kwargs.get("x", None)
        y: Optional[Number] = kwargs.get("y", None)
        z: Optional[Number] = kwargs.get("z", None)
        if len(args) == 1:
            assert isinstance(
                args[0], (list, tuple)
            ), "single argument must be a list or tuple."
            x = cast_if_not_none(args[0][0], float)
            y = cast_if_not_none(args[0][1], float)
            if len(args[0]) > 2 and (z := args[0][2]) is not None:
                z = cast_if_not_none(z, float)

        elif len(args) >= 2:
            x = cast_if_not_none(args[0], float)
            y = cast_if_not_none(args[1], float)
            if len(args) > 2 and (z := args[2]) is not None:
                z = cast_if_not_none(z, float)

        # 调试过程中发现z有时候是None有时候是0,因此这里统一给了默认值
        # 实际上应该外层规范
        x = x if x is not None else 0
        y = y if y is not None else 0
        z = z if z is not None else 0

        kwargs.update(x=x, y=y, z=z)

        super().__init__(precision=precision, **kwargs)

    def __repr__(self):
        return f"Point({self.x}, {self.y}, {self.z})"

    def __hash__(self):

        def s(v: Optional[Number]):
            if not self.precision:
                return v
            if v is None:
                return v
            return v
            # return str(Decimal(v).quantize(Decimal(f"0.{'0'*self.precision}")))

        # if self._id is not None:
        #     return hash((self.__class__.__name__, self._id))
        return hash((s(self.x), s(self.y), s(self.z), self.precision, self.unit))

    def __lt__(self, other):
        """
        FIXME: 考虑z
        """
        if isinstance(other, Point):
            # TODO: 只要任何一个大于就行?
            return (self.x, self.y) < (other.x, other.y)
        return NotImplemented

    def __sub__(self, other: "Point"):
        def sub(a: Number, b: Number):
            if a is None and b is None:
                return None
            a = a or 0
            b = b or 0
            return a - b

        assert self.unit == other.unit, "unit must be same"
        return Point.of(
            x=sub(self.x, other.x),
            y=sub(self.y, other.y),
            z=sub(self.z, other.z),
            unit=self.unit,
        )

    def to_tuple(self) -> TuplePoint:
        return (
            self.x,
            self.y,
            self.z,
        )

    def __str__(self):
        return f"Point(id={self._id},x={self.x},y={self.y},z={self.z})"

__counter class-attribute instance-attribute

__counter = count(0, step=1)

__global_precision__ class-attribute instance-attribute

__global_precision__ = 3

unit instance-attribute

unit: Optional[str] = get('unit', None)

__hash__

__hash__()
源代码位于: logis/data_type/point.py
def __hash__(self):

    def s(v: Optional[Number]):
        if not self.precision:
            return v
        if v is None:
            return v
        return v
        # return str(Decimal(v).quantize(Decimal(f"0.{'0'*self.precision}")))

    # if self._id is not None:
    #     return hash((self.__class__.__name__, self._id))
    return hash((s(self.x), s(self.y), s(self.z), self.precision, self.unit))

__init__

__init__(*args, precision: Optional[int] = None, **kwargs)

支持[x,y]、x,y、[x,y,z]、x,y,z传参赋值 默认x=y=z=0

TODO: 不应该自动转float、不应该自动保留两位小数

源代码位于: logis/data_type/point.py
def __init__(self, *args, precision: Optional[int] = None, **kwargs):
    """
    支持[x,y]、x,y、[x,y,z]、x,y,z传参赋值
    默认x=y=z=0

    TODO: 不应该自动转float、不应该自动保留两位小数
    """
    self._id = next(self.__counter)
    self.unit: Optional[str] = kwargs.get("unit", None)

    x: Optional[Number] = kwargs.get("x", None)
    y: Optional[Number] = kwargs.get("y", None)
    z: Optional[Number] = kwargs.get("z", None)
    if len(args) == 1:
        assert isinstance(
            args[0], (list, tuple)
        ), "single argument must be a list or tuple."
        x = cast_if_not_none(args[0][0], float)
        y = cast_if_not_none(args[0][1], float)
        if len(args[0]) > 2 and (z := args[0][2]) is not None:
            z = cast_if_not_none(z, float)

    elif len(args) >= 2:
        x = cast_if_not_none(args[0], float)
        y = cast_if_not_none(args[1], float)
        if len(args) > 2 and (z := args[2]) is not None:
            z = cast_if_not_none(z, float)

    # 调试过程中发现z有时候是None有时候是0,因此这里统一给了默认值
    # 实际上应该外层规范
    x = x if x is not None else 0
    y = y if y is not None else 0
    z = z if z is not None else 0

    kwargs.update(x=x, y=y, z=z)

    super().__init__(precision=precision, **kwargs)

__lt__

__lt__(other)

FIXME: 考虑z

源代码位于: logis/data_type/point.py
def __lt__(self, other):
    """
    FIXME: 考虑z
    """
    if isinstance(other, Point):
        # TODO: 只要任何一个大于就行?
        return (self.x, self.y) < (other.x, other.y)
    return NotImplemented

__repr__

__repr__()
源代码位于: logis/data_type/point.py
def __repr__(self):
    return f"Point({self.x}, {self.y}, {self.z})"

__str__

__str__()
源代码位于: logis/data_type/point.py
def __str__(self):
    return f"Point(id={self._id},x={self.x},y={self.y},z={self.z})"

__sub__

__sub__(other: Point)
源代码位于: logis/data_type/point.py
def __sub__(self, other: "Point"):
    def sub(a: Number, b: Number):
        if a is None and b is None:
            return None
        a = a or 0
        b = b or 0
        return a - b

    assert self.unit == other.unit, "unit must be same"
    return Point.of(
        x=sub(self.x, other.x),
        y=sub(self.y, other.y),
        z=sub(self.z, other.z),
        unit=self.unit,
    )

from_tuple classmethod

from_tuple(args: Tuple[Number], **kwargs)

从元组创建点

源代码位于: logis/data_type/point.py
@classmethod
def from_tuple(cls, args: Tuple[Number], **kwargs):
    """
    从元组创建点
    """
    return cls(args, **kwargs)

of classmethod

of(x: Optional[Number] = None, y: Optional[Number] = None, z: Optional[Number] = None, **kwargs) -> Point
源代码位于: logis/data_type/point.py
@classmethod
def of(
    cls,
    x: Optional[Number] = None,
    y: Optional[Number] = None,
    z: Optional[Number] = None,
    **kwargs,
) -> "Point":
    p = Point(x, y, z, **kwargs)
    return p

to_tuple

to_tuple() -> TuplePoint
源代码位于: logis/data_type/point.py
def to_tuple(self) -> TuplePoint:
    return (
        self.x,
        self.y,
        self.z,
    )

try_parse classmethod

try_parse(dc: dict, **kwargs) -> Point

尝试读取X、Y、Z坐标,返回一个Point对象

源代码位于: logis/data_type/point.py
@classmethod
def try_parse(cls, dc: dict, **kwargs) -> "Point":
    """
    尝试读取X、Y、Z坐标,返回一个Point对象
    """
    if not isinstance(dc, dict):
        raise ValueError("Expected a dictionary for Point parsing.")
    xyz = [dc.get(k, None) for k in ("X", "Y", "Z")]
    return Point(xyz, **kwargs)

QuantifiedValue

量化值,相比于接口NumberUnit,适合作为数据结构使用

源代码位于: logis/data_type/unitable.py
class QuantifiedValue(BaseModel, NumberUnit):
    """
    量化值,相比于接口NumberUnit,适合作为数据结构使用
    """
    model_config = MODEL_CONFIG

    name: Optional[str] = None

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

name class-attribute instance-attribute

name: Optional[str] = None

__init__

__init__(**kwargs)
源代码位于: logis/data_type/unitable.py
def __init__(self, **kwargs):
    super().__init__(**kwargs)

SpatialProps

空间属性

源代码位于: logis/data_type/spatial.py
class SpatialProps(BaseModel):
    """
    空间属性
    """

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

    width: Optional[Length] = Field(
        default=None,
        validation_alias=AliasChoices("单元格宽度", "width"),
    )
    height: Optional[Length] = Field(
        default=None, validation_alias=AliasChoices("层高", "height")
    )
    depth: Optional[Length] = Field(
        validation_alias=AliasChoices("货架深度", "托盘货架深度"), default=None
    )
    rotation: Optional[QuantifiedValue] = Field(
        validation_alias=AliasChoices("旋转", "rotate"), default=None
    )
    center_point: Optional[Point] = None

    @field_validator("center_point", mode="before")
    def center_point_validator_before(cls, v: Optional[Point]):
        if v is None:
            return None
        if isinstance(v, Point):
            return v
        if isinstance(v, dict):
            return Point.model_validate(v)
        elif isinstance(v, (tuple, list)):
            return Point.from_tuple(v)
        raise ValueError("only dict, tuple, list type is supported")

center_point class-attribute instance-attribute

center_point: Optional[Point] = None

depth class-attribute instance-attribute

depth: Optional[Length] = Field(validation_alias=AliasChoices('货架深度', '托盘货架深度'), default=None)

height class-attribute instance-attribute

height: Optional[Length] = Field(default=None, validation_alias=AliasChoices('层高', 'height'))

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

rotation class-attribute instance-attribute

rotation: Optional[QuantifiedValue] = Field(validation_alias=AliasChoices('旋转', 'rotate'), default=None)

width class-attribute instance-attribute

width: Optional[Length] = Field(default=None, validation_alias=AliasChoices('单元格宽度', 'width'))

center_point_validator_before

center_point_validator_before(v: Optional[Point])
源代码位于: logis/data_type/spatial.py
@field_validator("center_point", mode="before")
def center_point_validator_before(cls, v: Optional[Point]):
    if v is None:
        return None
    if isinstance(v, Point):
        return v
    if isinstance(v, dict):
        return Point.model_validate(v)
    elif isinstance(v, (tuple, list)):
        return Point.from_tuple(v)
    raise ValueError("only dict, tuple, list type is supported")

Speed

源代码位于: logis/data_type/unitable.py
class Speed(QuantifiedValue):
    pass

TableQuery

源代码位于: logis/data_type/page.py
class TableQuery(Pager):
    database: Optional[str] = None
    table: str

    model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

database class-attribute instance-attribute

database: Optional[str] = None

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

table instance-attribute

table: str

ThreeDimensionalVelocity

使用x,y,z表示的三维速度

源代码位于: logis/data_type/unitable.py
class ThreeDimensionalVelocity(BaseModel):
    """
    使用x,y,z表示的三维速度
    """

    model_config = MODEL_CONFIG

    x: Optional[Speed] = None
    y: Optional[Speed] = None
    z: Optional[Speed] = None

    @classmethod
    def of(
        cls,
        x: Optional[NumberType] = None,
        y: Optional[NumberType] = None,
        z: Optional[NumberType] = None,
        unit: Optional[str] = None,
    ):
        return cls(
            x=None if x is None else Speed.of(x, unit),
            y=None if y is None else Speed.of(y, unit),
            z=None if z is None else Speed.of(z, unit),
        )

model_config class-attribute instance-attribute

model_config = DEFAULT_PYDANTIC_MODEL_CONFIG

x class-attribute instance-attribute

x: Optional[Speed] = None

y class-attribute instance-attribute

y: Optional[Speed] = None

z class-attribute instance-attribute

z: Optional[Speed] = None

of classmethod

of(x: Optional[NumberType] = None, y: Optional[NumberType] = None, z: Optional[NumberType] = None, unit: Optional[str] = None)
源代码位于: logis/data_type/unitable.py
@classmethod
def of(
    cls,
    x: Optional[NumberType] = None,
    y: Optional[NumberType] = None,
    z: Optional[NumberType] = None,
    unit: Optional[str] = None,
):
    return cls(
        x=None if x is None else Speed.of(x, unit),
        y=None if y is None else Speed.of(y, unit),
        z=None if z is None else Speed.of(z, unit),
    )

Time

时间

源代码位于: logis/data_type/unitable.py
class Time(QuantifiedValue):
    """
    时间
    """

    pass

UnitConfig

有关单位的配置项

源代码位于: logis/data_type/unitable.py
class UnitConfig(dict):
    """
    有关单位的配置项
    """

    def get_ratio(self, src: Unit, dst: Unit, src_quantity: NumberType = 1) -> Fraction:
        """
        获取 src 到 dst 的倍率
        """
        if src == dst:
            return Fraction(src_quantity, 1)
        if src in self and dst in self:
            return Fraction(self[dst] * src_quantity, self[src])
        try:
            return Fraction(get_unit_ratio(src, dst, src_quantity=src_quantity))
        except:
            pass
        raise ValueError(f"unknown unit: {src} or {dst}")

    def get_float_ratio(self, src: Unit, dst: Unit, *args, **kwargs) -> float:
        """
        获取 src 到 dst 的倍率
        """
        return float(self.get_ratio(src, dst, *args, **kwargs))

    def get_int_ratio(self, src: Unit, dst: Unit, *args, **kwargs) -> int:
        """
        获取 src 到 dst 的倍率
        """
        return int(self.get_ratio(src, dst, *args, **kwargs))

    def alias(self, unit: Unit, *aliases: Unit):
        """
        为单位添加别名
        """
        for alias in aliases:
            self[alias] = self.get(unit)
        return self

    def __or__(self, value: Dict):

        obj = dict(value)
        obj.update(self)

        return UnitConfig(obj)

__or__

__or__(value: Dict)
源代码位于: logis/data_type/unitable.py
def __or__(self, value: Dict):

    obj = dict(value)
    obj.update(self)

    return UnitConfig(obj)

alias

alias(unit: Unit, *aliases: Unit)

为单位添加别名

源代码位于: logis/data_type/unitable.py
def alias(self, unit: Unit, *aliases: Unit):
    """
    为单位添加别名
    """
    for alias in aliases:
        self[alias] = self.get(unit)
    return self

get_float_ratio

get_float_ratio(src: Unit, dst: Unit, *args, **kwargs) -> float

获取 src 到 dst 的倍率

源代码位于: logis/data_type/unitable.py
def get_float_ratio(self, src: Unit, dst: Unit, *args, **kwargs) -> float:
    """
    获取 src 到 dst 的倍率
    """
    return float(self.get_ratio(src, dst, *args, **kwargs))

get_int_ratio

get_int_ratio(src: Unit, dst: Unit, *args, **kwargs) -> int

获取 src 到 dst 的倍率

源代码位于: logis/data_type/unitable.py
def get_int_ratio(self, src: Unit, dst: Unit, *args, **kwargs) -> int:
    """
    获取 src 到 dst 的倍率
    """
    return int(self.get_ratio(src, dst, *args, **kwargs))

get_ratio

get_ratio(src: Unit, dst: Unit, src_quantity: NumberType = 1) -> Fraction

获取 src 到 dst 的倍率

源代码位于: logis/data_type/unitable.py
def get_ratio(self, src: Unit, dst: Unit, src_quantity: NumberType = 1) -> Fraction:
    """
    获取 src 到 dst 的倍率
    """
    if src == dst:
        return Fraction(src_quantity, 1)
    if src in self and dst in self:
        return Fraction(self[dst] * src_quantity, self[src])
    try:
        return Fraction(get_unit_ratio(src, dst, src_quantity=src_quantity))
    except:
        pass
    raise ValueError(f"unknown unit: {src} or {dst}")

UnitConfigBuilder

单位配置构建器

源代码位于: logis/data_type/unitable.py
class UnitConfigBuilder:
    """
    单位配置构建器
    """

    def __init__(self, base_unit: Unit):
        self.__unit_config__ = UnitConfig([[base_unit, 1]])

    @staticmethod
    def with_base(unit: Unit):
        return UnitConfigBuilder(unit)

    def add(self, unit: Unit, multiplier_to_base: NumberType):
        """
        Args:
            unit: 单位
            multiplier_to_base: 转换到1base单位所需要的倍率,例如:base为1m时,cm到1m需要100倍
        """
        self.__unit_config__[unit] = multiplier_to_base
        return self

    def alias(self, unit: Unit, *aliases: Unit):
        """
        为单位添加别名
        """
        self.__unit_config__.alias(unit=unit, *aliases)
        return self

    def build(self) -> UnitConfig:
        return self.__unit_config__

__unit_config__ instance-attribute

__unit_config__ = UnitConfig([[base_unit, 1]])

__init__

__init__(base_unit: Unit)
源代码位于: logis/data_type/unitable.py
def __init__(self, base_unit: Unit):
    self.__unit_config__ = UnitConfig([[base_unit, 1]])

add

add(unit: Unit, multiplier_to_base: NumberType)

参数:

名称 类型 描述 默认
unit Unit

单位

必需
multiplier_to_base NumberType

转换到1base单位所需要的倍率,例如:base为1m时,cm到1m需要100倍

必需
源代码位于: logis/data_type/unitable.py
def add(self, unit: Unit, multiplier_to_base: NumberType):
    """
    Args:
        unit: 单位
        multiplier_to_base: 转换到1base单位所需要的倍率,例如:base为1m时,cm到1m需要100倍
    """
    self.__unit_config__[unit] = multiplier_to_base
    return self

alias

alias(unit: Unit, *aliases: Unit)

为单位添加别名

源代码位于: logis/data_type/unitable.py
def alias(self, unit: Unit, *aliases: Unit):
    """
    为单位添加别名
    """
    self.__unit_config__.alias(unit=unit, *aliases)
    return self

build

build() -> UnitConfig
源代码位于: logis/data_type/unitable.py
def build(self) -> UnitConfig:
    return self.__unit_config__

with_base staticmethod

with_base(unit: Unit)
源代码位于: logis/data_type/unitable.py
@staticmethod
def with_base(unit: Unit):
    return UnitConfigBuilder(unit)

UpdateResponse

源代码位于: logis/data_type/updater.py
class UpdateResponse(BaseModel):
    current_version: str
    version_info: VersionInfo
    need_update: bool

current_version instance-attribute

current_version: str

need_update instance-attribute

need_update: bool

version_info instance-attribute

version_info: VersionInfo

UserFacingException

用户面向的异常类,用于在前端展示异常信息

源代码位于: logis/data_type/exception.py
class UserFacingException(MyBaseException):
    """
    用户面向的异常类,用于在前端展示异常信息
    """

    def __init__(self, *args, **attrs):
        super().__init__(*args, **attrs)

__init__

__init__(*args, **attrs)
源代码位于: logis/data_type/exception.py
def __init__(self, *args, **attrs):
    super().__init__(*args, **attrs)

VersionInfo

源代码位于: logis/data_type/updater.py
class VersionInfo(BaseModel):
    version: str  # 版本号,必选字符串
    notes: str  # 更新说明,必选字符串
    pub_date: datetime  # 发布时间,自动解析ISO 8601格式(如2026-02-02T10:00:00Z)
    platforms: Dict[str, PlatformInfo]

notes instance-attribute

notes: str

platforms instance-attribute

platforms: Dict[str, PlatformInfo]

pub_date instance-attribute

pub_date: datetime

version instance-attribute

version: str

cast_if_not_none

cast_if_not_none(value: Optional[Num], type_: type[Num]) -> Optional[Num]

当value不为None时,转换为指定类型,否则返回None。

:param value: 要转换的值 :type value: Optional[T] :param type_: 目标类型 :type type_: type[T] :return: 转换后的值 :rtype: T | None

源代码位于: logis/data_type/point.py
def cast_if_not_none(value: Optional[Num], type_: type[Num]) -> Optional[Num]:
    """
    当value不为None时,转换为指定类型,否则返回None。

    :param value: 要转换的值
    :type value: Optional[T]
    :param type_: 目标类型
    :type type_: type[T]
    :return: 转换后的值
    :rtype: T | None
    """
    if value is not None:
        return type_(value)
    return None

compute

compute(operation: Literal['add', 'subtract', 'multiply', 'divide'], *nums: Number) -> Number
源代码位于: logis/data_type/_math.py
def compute(
    operation: Literal["add", "subtract", "multiply", "divide"], *nums: Number
) -> Number:
    if not nums:
        raise ValueError("At least one number is required.")

    if len(nums) == 1:
        return nums[0]

    all_none = all(num is None for num in nums)
    if all_none:
        return None

    all_numbers = all(isinstance(num, Number) for num in nums)
    if not all_numbers:
        raise ValueError("All arguments must be numbers.")

    if operation == "add":
        return sum(nums)
    elif operation == "subtract":
        return nums[0] - sum(nums[1:])
    elif operation == "multiply":
        return math.prod(nums)
    elif operation == "divide":
        return nums[0] / math.prod(nums[1:])
    else:
        raise ValueError(f"Unknown operation: {operation}")

get_time

get_time(l: Length, v: Speed)

距离/速度=时间 TODO:处理单位

源代码位于: logis/data_type/unitable.py
def get_time(l: Length, v: Speed):
    """
    距离/速度=时间
    TODO:处理单位
    """
    assert l.value is not None, "length is not specified"
    assert v.value is not None, "speed is not specified"
    assert v.value != 0, "speed cannot be zero"
    return abs(l.value / v.value)

get_time_3d

get_time_3d(delta_distance: Point, v: ThreeDimensionalVelocity)

分别计算各维度的时间,取最大值

源代码位于: logis/data_type/unitable.py
def get_time_3d(delta_distance: Point, v: ThreeDimensionalVelocity):
    """
    分别计算各维度的时间,取最大值
    """
    lz, ly, lx, lunit = (
        delta_distance.z,
        delta_distance.y,
        delta_distance.x,
        delta_distance.unit,
    )
    return max(
        0 if not lx else get_time(Length(quantity=lx, unit=lunit), v.x),
        0 if not ly else get_time(Length(quantity=ly, unit=lunit), v.y),
        0 if not lz else get_time(Length(quantity=lz, unit=lunit), v.z),
    )

get_unit_ratio

get_unit_ratio(src: Unit, dst: Unit, src_quantity: NumberType = 1) -> float

获取 src 到 dst 的倍率

源代码位于: logis/data_type/unitable.py
def get_unit_ratio(src: Unit, dst: Unit, src_quantity: NumberType = 1) -> float:
    """
    获取 src 到 dst 的倍率
    """
    src, dst = src.replace("每", "/"), dst.replace("每", "/")
    return ureg.Quantity(src_quantity, src).to(dst).magnitude

group_merge_quantified_value

group_merge_quantified_value(items: List[QuantifiedValue]) -> List[QuantifiedValue]

按照单位分组,并合并同类单位物理量

源代码位于: logis/data_type/unitable.py
def group_merge_quantified_value(items: List[QuantifiedValue]) -> List[QuantifiedValue]:
    """
    按照单位分组,并合并同类单位物理量
    """
    result = []
    unit_items_map = defaultdict(list)
    for item in items:
        unit_items_map[item.unit].append(item)
    for _, tmps in unit_items_map.items():
        result.append(merge_quantified_value(tmps))
    return result

merge_quantified_value

merge_quantified_value(items: List[QuantifiedValue], unit_config: Optional[UnitConfig] = DEFAULT_UNIT_CONFIG, target_unit: Optional[Unit] = None)

将所有的物理量合并,这里是假定所有的物理量都是同类,例如:全是长度,而不是既有长度又有速度

参数:

名称 类型 描述 默认
items List[QuantifiedValue]

待合并项

必需
ratio_computer

倍率计算器

必需
target_unit Optional[Unit]

目标单位,如果未指定,默认使用数组的第一个条目的单位

None
源代码位于: logis/data_type/unitable.py
def merge_quantified_value(
    items: List[QuantifiedValue],
    unit_config: Optional[UnitConfig] = DEFAULT_UNIT_CONFIG,
    target_unit: Optional[Unit] = None,
):
    """
    将所有的物理量合并,这里是假定所有的物理量都是同类,例如:全是长度,而不是既有长度又有速度

    Args:
        items: 待合并项
        ratio_computer: 倍率计算器
        target_unit: 目标单位,如果未指定,默认使用数组的第一个条目的单位
    """
    assert items, "item length need to be at least 1"
    target_unit = target_unit or items[0].unit
    if len(items) != 2:
        return reduce(
            lambda a, b: merge_quantified_value(
                [a, b], target_unit=target_unit, unit_config=unit_config
            ),
            items,
        )
    self, other = unify_quantified_value(
        items[0], target_unit=target_unit, unit_config=unit_config
    ), unify_quantified_value(
        items[1], target_unit=target_unit, unit_config=unit_config
    )
    dc = {
        **self.model_dump(),
        **dict(quantity=self.quantity + other.quantity, unit=target_unit),
    }
    return type(self)(**dc)

round_if_not_none

round_if_not_none(value: Optional[Num], ndigits: Optional[int] = None, rounding=ROUND_HALF_UP) -> Optional[Num]

当value和ndigits都不为None时,进行四舍五入,否则返回原数值。

:param value: 要四舍五入的值 :type value: Optional[T] :param ndigits: 四舍五入的小数位数,与原round的行为不同的是,这里为None时返回原数值 :type ndigits: Optional[int] :return: 四舍五入后的值 :rtype: T | None

源代码位于: logis/data_type/point.py
def round_if_not_none(
    value: Optional[Num], ndigits: Optional[int] = None, rounding=ROUND_HALF_UP
) -> Optional[Num]:
    """
    当value和ndigits都不为None时,进行四舍五入,否则返回原数值。

    :param value: 要四舍五入的值
    :type value: Optional[T]
    :param ndigits: 四舍五入的小数位数,**与原round的行为不同的是,这里为None时返回原数值**
    :type ndigits: Optional[int]
    :return: 四舍五入后的值
    :rtype: T | None
    """
    if value is not None and ndigits is not None:
        # 这里如果不用str转换,float的精度问题仍然会导致结果不正确,例如1216.745、1216.74512等
        v = Decimal(str(value)).quantize(
            Decimal("0." + "0" * ndigits), rounding=rounding
        )
        # if int(value) == 976:
        #     print(value, v, float(v))
        if type(value) is type(v):
            return v
        return type(value)(v)
    return value

unify_quantified_value

unify_quantified_value(self: QuantifiedValue, target_unit: Optional[Unit] = None, unit_config: UnitConfig = DEFAULT_UNIT_CONFIG, modify_ref: bool = False)

单位转换

源代码位于: logis/data_type/unitable.py
def unify_quantified_value(
    self: QuantifiedValue,
    target_unit: Optional[Unit] = None,
    unit_config: UnitConfig = DEFAULT_UNIT_CONFIG,
    modify_ref: bool = False,
):
    """
    单位转换
    """
    same_unit = self.unit == target_unit
    if unit_config is None:
        assert same_unit, "unit must be the same if no radio_computer given"
    else:
        ratio = 1 if same_unit else unit_config.get_ratio(self.unit, target_unit)
        if modify_ref:
            self.quantity = self.quantity * ratio
            self.unit = target_unit
            return self
        # 上面这种方式能尽可能避免属性丢失,但修改的是原始对象
        dc = {
            **self.model_dump(),
            **dict(quantity=self.quantity * ratio, unit=target_unit),
        }
        return type(self)(**dc)

接口类型

E module-attribute

E = TypeVar('E', bound=IEdge)

QueueType module-attribute

QueueType = TypeVar('QueueType', bound=IQueue[T])

T module-attribute

T = TypeVar('T')

V module-attribute

V = TypeVar('V')

Consumer

消费者

源代码位于: logis/iface/_event.py
@runtime_checkable
class Consumer(Protocol):
    """
    消费者
    """

    def consume(self, *args, **kwargs):
        pass

consume

consume(*args, **kwargs)
源代码位于: logis/iface/_event.py
def consume(self, *args, **kwargs):
    pass

IControl

控制器

源代码位于: logis/iface/manage.py
@runtime_checkable
class IControl(Protocol):
    """
    控制器
    """

    disabled: bool = False

disabled class-attribute instance-attribute

disabled: bool = False

IEdge

源代码位于: logis/iface/graph.py
class IEdge(ABC):

    @abstractmethod
    def get_weight(self, *args, **kwargs) -> float:
        """
        获取边的权重
        """
        pass

    @abstractmethod
    def is_directed(self) -> bool:
        """
        是否有向边
        """
        pass

get_weight abstractmethod

get_weight(*args, **kwargs) -> float

获取边的权重

源代码位于: logis/iface/graph.py
@abstractmethod
def get_weight(self, *args, **kwargs) -> float:
    """
    获取边的权重
    """
    pass

is_directed abstractmethod

is_directed() -> bool

是否有向边

源代码位于: logis/iface/graph.py
@abstractmethod
def is_directed(self) -> bool:
    """
    是否有向边
    """
    pass

INode

源代码位于: logis/iface/graph.py
class INode(ABC):
    pass

IPathGraph

路径关系图

源代码位于: logis/iface/graph.py
class IPathGraph(ABC, Generic[E, V]):
    """
    路径关系图
    """

    @abstractmethod
    def edges(self, *args, **kwargs) -> List[E]:
        """
        所有的边
        """
        pass

    @abstractmethod
    def nodes(self, *args, **kwargs) -> List[V]:
        """
        所有的节点
        """
        pass

edges abstractmethod

edges(*args, **kwargs) -> List[E]

所有的边

源代码位于: logis/iface/graph.py
@abstractmethod
def edges(self, *args, **kwargs) -> List[E]:
    """
    所有的边
    """
    pass

nodes abstractmethod

nodes(*args, **kwargs) -> List[V]

所有的节点

源代码位于: logis/iface/graph.py
@abstractmethod
def nodes(self, *args, **kwargs) -> List[V]:
    """
    所有的节点
    """
    pass

IQueue

队列接口

源代码位于: logis/iface/_queue.py
@runtime_checkable
class IQueue(Protocol[T]):
    """
    队列接口
    """

    def __init__(self, maxsize: int = 0):
        super().__init__(maxsize)

    def put(self, item: T, *args, **kwargs):
        pass

    def put_nowait(self, item: T):
        pass

    def get(self, *args, **kwargs) -> T:
        pass

    def get_nowait(self) -> T:
        pass

    def qsize(self) -> int:
        pass

    def empty(self) -> bool:
        pass

    def full(self) -> bool:
        pass

__init__

__init__(maxsize: int = 0)
源代码位于: logis/iface/_queue.py
def __init__(self, maxsize: int = 0):
    super().__init__(maxsize)

empty

empty() -> bool
源代码位于: logis/iface/_queue.py
def empty(self) -> bool:
    pass

full

full() -> bool
源代码位于: logis/iface/_queue.py
def full(self) -> bool:
    pass

get

get(*args, **kwargs) -> T
源代码位于: logis/iface/_queue.py
def get(self, *args, **kwargs) -> T:
    pass

get_nowait

get_nowait() -> T
源代码位于: logis/iface/_queue.py
def get_nowait(self) -> T:
    pass

put

put(item: T, *args, **kwargs)
源代码位于: logis/iface/_queue.py
def put(self, item: T, *args, **kwargs):
    pass

put_nowait

put_nowait(item: T)
源代码位于: logis/iface/_queue.py
def put_nowait(self, item: T):
    pass

qsize

qsize() -> int
源代码位于: logis/iface/_queue.py
def qsize(self) -> int:
    pass

Producer

生产者

源代码位于: logis/iface/_event.py
@runtime_checkable
class Producer(Protocol):
    """
    生产者
    """

    def produce(self, *args, **kwargs):
        pass

produce

produce(*args, **kwargs)
源代码位于: logis/iface/_event.py
def produce(self, *args, **kwargs):
    pass

QuantifiedValueContainer

源代码位于: logis/iface/container.py
class QuantifiedValueContainer(Storable):

    def __init__(self):
        self.__container__: Dict[str, Dict[str, NumberType]] = defaultdict(
            defaultdict(0)
        )

    def store(self, v: "NumberUnit"):
        self.__container__[v.kind][v.unit] += v.quantity

    def retrieve(self, v: "NumberUnit"):
        self.__container__[v.kind][v.unit] -= v.quantity

__container__ instance-attribute

__container__: Dict[str, Dict[str, NumberType]] = defaultdict(defaultdict(0))

__init__

__init__()
源代码位于: logis/iface/container.py
def __init__(self):
    self.__container__: Dict[str, Dict[str, NumberType]] = defaultdict(
        defaultdict(0)
    )

retrieve

retrieve(v: NumberUnit)
源代码位于: logis/iface/container.py
def retrieve(self, v: "NumberUnit"):
    self.__container__[v.kind][v.unit] -= v.quantity

store

store(v: NumberUnit)
源代码位于: logis/iface/container.py
def store(self, v: "NumberUnit"):
    self.__container__[v.kind][v.unit] += v.quantity

RetrieveResult

检索结果

源代码位于: logis/iface/container.py
class RetrieveResult(metaclass=ABCMeta):
    """
    检索结果
    """

    pass

Shape

形状,例如路径、节点等

源代码位于: logis/iface/__init__.py
class Shape(metaclass=ABCMeta):
    """
    形状,例如路径、节点等
    """

    @property
    def rotation_value(self):
        v = get_numeric_value(self.props.rotation) if self.props else None
        return v or 0

    def __init__(self, **kwargs):
        self.props: Union[SpatialProps, None] = None

props instance-attribute

props: Union[SpatialProps, None] = None

rotation_value property

rotation_value

__init__

__init__(**kwargs)
源代码位于: logis/iface/__init__.py
def __init__(self, **kwargs):
    self.props: Union[SpatialProps, None] = None

Storable

可存储的

使用鸭子类型实现,相比ABC更灵活

源代码位于: logis/iface/container.py
@runtime_checkable
class Storable(Protocol):
    # class Storable(metaclass=ABCMeta):
    """
    可存储的

    使用鸭子类型实现,相比ABC更灵活
    """

    def pre_store(self, *args, **kwargs):
        """
        预存储,一般用来实现资源检查、校验
        """
        pass

    def pre_retrieve(self, *args, **kwargs):
        """
        预取回,一般用来实现资源检查、校验
        """
        pass

    def store(self, *args, **kwargs) -> StoreResult:
        """
        真正的存储操作
        """
        pass

    def retrieve(self, *args, **kwargs) -> RetrieveResult:
        """
        真正的取回操作
        """
        pass

pre_retrieve

pre_retrieve(*args, **kwargs)

预取回,一般用来实现资源检查、校验

源代码位于: logis/iface/container.py
def pre_retrieve(self, *args, **kwargs):
    """
    预取回,一般用来实现资源检查、校验
    """
    pass

pre_store

pre_store(*args, **kwargs)

预存储,一般用来实现资源检查、校验

源代码位于: logis/iface/container.py
def pre_store(self, *args, **kwargs):
    """
    预存储,一般用来实现资源检查、校验
    """
    pass

retrieve

retrieve(*args, **kwargs) -> RetrieveResult

真正的取回操作

源代码位于: logis/iface/container.py
def retrieve(self, *args, **kwargs) -> RetrieveResult:
    """
    真正的取回操作
    """
    pass

store

store(*args, **kwargs) -> StoreResult

真正的存储操作

源代码位于: logis/iface/container.py
def store(self, *args, **kwargs) -> StoreResult:
    """
    真正的存储操作
    """
    pass

StoreResult

存储结果

源代码位于: logis/iface/container.py
class StoreResult(metaclass=ABCMeta):
    """
    存储结果
    """

    pass

StoreStrategy

存储策略

源代码位于: logis/iface/container.py
class StoreStrategy(metaclass=ABCMeta):
    """
    存储策略
    """