microspec.microspec

  1#!/usr/bin/env python3
  2"""microspec.py - Micro protocol parser and data validator from a simple JSON-based specification."""
  3from __future__ import annotations
  4import math
  5from dataclasses import dataclass
  6from enum import StrEnum
  7from typing import Any
  8import numpy as np
  9
 10MAX_SHAPE = 1_000_000_000
 11
 12@dataclass
 13class Endpoint:
 14    """An endpoint of the protocol. It has two sets of fields: input and output. Each field has a schema (dtype, etc)"""
 15    name: str
 16    input: dict[str, Field]
 17    output: dict[str, Field]
 18    description: str
 19
 20    def to_dict(self) -> dict:
 21        """dict representation of this endpoint"""
 22        return {"input": {k: v.to_dict() for k, v in self.input.items()},
 23                "output": {k: v.to_dict() for k, v in self.output.items()},
 24                "description": self.description}
 25
 26    @staticmethod
 27    def from_dict(name: str, data: dict, **kwargs) -> Endpoint:
 28        """create an endpoint from dict data"""
 29        return _parse_endpoint(name, endpoint_data=data, **kwargs)
 30
 31    def __hash__(self):
 32        return hash(self.name)
 33
 34    def __repr__(self):
 35        return f"{self.name}({','.join(self.input.keys())}) -> ({','.join(self.output.keys())})"
 36
 37@dataclass
 38class Field:
 39    """A field has a single required parameters (dtype) and a bunch of optional ones, some required based on dtype"""
 40    dtype: Dtype
 41    shape: tuple[int | None, *tuple[int, ...]] | None = None  # (1, 2, 3, ) or (None, 1, 2, 3., ..),
 42                                                              # valid for float32, int32, bool, str
 43    range: tuple[int, int] | None = None                      # valid for float32, int32
 44    enum: list[str] | list[int] | None = None                 # valid for int_enum or str_enum
 45    min_len: int | None = None                                # valid only if shape[0] is None (e.g. batched array)
 46    fields: dict[str, Field] | None = None                    # valid only for dict and is optional for opaque dicts
 47
 48    def to_dict(self) -> dict:
 49        """dict representation of this field"""
 50        res = {"dtype": self.dtype.value}
 51        if self.shape is not None:
 52            res["shape"] = list(self.shape)  # tuple -> list so from_dict(to_dict()) round-trips (parser wants a list)
 53        if self.range is not None:
 54            res["range"] = self.range
 55        if self.enum is not None:
 56            res["enum"] = self.enum
 57        if self.min_len is not None:
 58            res["min_len"] = self.min_len
 59        if self.fields is not None:
 60            res["fields"] = {k: v.to_dict() for k, v in self.fields.items()}
 61        return res
 62
 63class Dtype(StrEnum):
 64    """All allowed data types by this protocol"""
 65    STR      = "str"
 66    INT32    = "int32"
 67    FLOAT32  = "float32"
 68    BOOL     = "bool"
 69    DICT     = "dict"
 70    BYTES    = "bytes"
 71    STR_ENUM = "str_enum"
 72    INT_ENUM = "int_enum"
 73
 74@dataclass
 75class ValidationError:
 76    """Validation error is a dataclass for returning errors from validation (field + data + error msg)"""
 77    field: Field | None
 78    data: Any
 79    error: str
 80    endpoint: Endpoint | None = None
 81
 82class ParseError(ValueError):
 83    """custom exception for parsing errors"""
 84
 85# parsing
 86
 87def _interp(item: str | Any, **kwargs) -> Any:
 88    """value interpolation from ${variable_name} to an actual value provided via kwargs"""
 89    if isinstance(item, str) and item[0:2] == "${" and item[-1] == "}":
 90        if item[2:-1] not in kwargs:
 91            raise ParseError(f"{item} not in {kwargs=}")
 92        return kwargs[item[2:-1]]
 93    return item
 94
 95def _parse_field(name: str, field_data: dict[str, Any], **kwargs) -> Field:
 96    if "dtype" not in field_data:
 97        raise ParseError(f"{name=}: 'dtype' not in {list(field_data.keys())}")
 98    res = {}
 99    for k, v in field_data.items():
100        if k == "dtype":
101            res[k] = Dtype(_interp(v, **kwargs))
102
103        elif k == "shape":
104            # shape can be of the following structure: (1, 2, 3) or (None, 1, 2, 3)
105            if res["dtype"] not in (Dtype.INT32, Dtype.FLOAT32, Dtype.STR, Dtype.BOOL, Dtype.INT_ENUM, Dtype.STR_ENUM):
106                raise ParseError(f"{name=} {k=}: invalid dtype for shape: {res['dtype']}")
107            if not (isinstance(v, list) and len(v) > 0):
108                raise ParseError(f"{name=} {k=}: must be a non-empty list: {v}")
109            v_interp = [_interp(_v, **kwargs) for _v in v]
110            if not isinstance(v[0], (int, type(None))):
111                raise ParseError(f"{name=} {k=}: first dim must be int or None: {v_interp}")
112            v_checked = v_interp[1:] if v[0] is None else v_interp # check all if v[0] is not None otherwise >=1 dim
113            if any(not isinstance(_v, int) or isinstance(_v, bool) or _v <= 0 or _v >= MAX_SHAPE for _v in v_checked):
114                raise ParseError(f"{name=} {k=}: dims must be positive ints < {MAX_SHAPE}: {v_interp}")
115            res[k] = tuple(v_interp)
116
117        elif k == "enum":
118            if res["dtype"] not in (Dtype.INT_ENUM, Dtype.STR_ENUM):
119                raise ParseError(f"{name=} {k=}: invalid dtype for enum: {res['dtype']}")
120            if isinstance(v, str): # for the special case of enum: "${data}" -> enum: [data1, data2,...]
121                v = _interp(v, **kwargs)
122            if not isinstance(v, list):
123                raise ParseError(f"{name=} {k=}: must be a list: {v}")
124            v_interp = [_interp(_v, **kwargs) for _v in v]
125            if len(v_interp) == 0:
126                raise ParseError(f"{name=} {k=}: must be a non-empty list: {v_interp}")
127            if res["dtype"] == Dtype.INT_ENUM:
128                if not all(isinstance(_v, int) and not isinstance(_v, bool) for _v in v_interp):
129                    raise ParseError(f"{name=} {k=}: all values must be ints: {v_interp}")
130            else: # Dtype.STR_ENUM
131                if not all(isinstance(_v, str) for _v in v_interp):
132                    raise ParseError(f"{name=} {k=}: all values must be strs: {v_interp}")
133            res[k] = v_interp
134
135        elif k == "range":
136            if res["dtype"] not in (Dtype.INT32, Dtype.FLOAT32):
137                raise ParseError(f"{name=} {k=}: invalid dtype for range: {res['dtype']}")
138            if not isinstance(v, list):
139                raise ParseError(f"{name=} {k=}: must be a list: {v}")
140            if len(v) != 2: # TODO: or of array shape so we can do range [[-1000, 1000], [-500, 500]] for (2, ) array.
141                raise ParseError(f"{name=} {k=}: must be of length 2: {v}")
142            v_interp = [_interp(_v, **kwargs) for _v in v]
143            if not v_interp[0] <= v_interp[1]:
144                raise ParseError(f"{name=} {k=}: low must be <= high: {v_interp}")
145            if res["dtype"] == Dtype.INT32:
146                if not all(isinstance(_v, int) for _v in v_interp):
147                    raise ParseError(f"{name=} {k=}: all values must be ints: {v_interp}")
148            else: # float32
149                if not all(isinstance(_v, (int, float)) and not math.isnan(_v) for _v in v_interp):
150                    raise ParseError(f"{name=} {k=}: all values must be non-NaN numbers: {v_interp}")
151            res[k] = v_interp
152
153        elif k == "min_len":
154            if "shape" not in res:
155                raise ParseError(f"{name=} {k=}: 'shape' not in {res}. Maybe define shape before min_len?")
156            if res["shape"][0] is not None:
157                raise ParseError(f"{name=} {k=} First shape is not None ({res['shape']})")
158            v_interp = _interp(v, **kwargs)
159            if not isinstance(v_interp, int) or v_interp <= 0:
160                raise ParseError(f"{name=} {k=} Must be a positive integer: {v_interp}")
161            res[k] = v_interp
162
163        elif k == "fields":
164            if res["dtype"] != Dtype.DICT:
165                raise ParseError(f"{name=} {k=}: invalid dtype for fields: {res['dtype']}")
166            if not isinstance(v, dict):
167                raise TypeError(f"{name=} {k=}: Expected dict, got {type(v)}")
168            if len(v) == 0:
169                raise ParseError(f"{name=} {k=}: 0 fields provided. For opaque dicts, just don't provide 'fields' key")
170            res[k] = {k2: _parse_field(name=k2, field_data=v2, **kwargs) for k2, v2 in v.items()}
171
172        else:
173            raise NotImplementedError(k, v)
174
175    # checking dtype invariants
176    if res["dtype"] == Dtype.INT_ENUM and "enum" not in res:
177        raise ParseError(f"{name=}: int_enum requires 'enum': {res}")
178    if res["dtype"] == Dtype.STR_ENUM and "enum" not in res:
179        raise ParseError(f"{name=}: str_enum requires 'enum': {res}")
180
181    return Field(**res)
182
183def _parse_endpoint(name: str, endpoint_data: dict[str, Any], **kwargs) -> Endpoint:
184    if endpoint_data.keys() != (expected := {"input", "output", "description"}):
185        raise ParseError(f"{name=}: endpoint must have keys {expected}, got {list(endpoint_data.keys())}")
186    res = {"name": name, "input": {}, "output": {}, "description": endpoint_data["description"]}
187    for ioe in ["input", "output"]:
188        for k, v in endpoint_data[ioe].items():
189            res[ioe][k] = _parse_field(name=k, field_data=v, **kwargs)
190    return Endpoint(**res)
191
192# validation
193
194def _validate_array_and_convert_to_numpy_if_needed(field: Field, data: Any) \
195        -> tuple[Any | np.ndarray, ValidationError | None]:
196    """Converts a data item to numpy array if the field is arr. Returns tuple: converted (or not) data + error/None"""
197    if field.shape is None: # not an array
198        return data, None
199
200    if field.dtype in (Dtype.STR, Dtype.STR_ENUM):
201        dtype = "object"
202    elif field.dtype == Dtype.INT_ENUM:
203        dtype = "int32"
204    else: # int32, float32, bool, etc.
205        dtype = field.dtype
206
207    try:
208        arr_data = np.array(data, dtype=dtype)
209    except (ValueError, TypeError) as e:
210        return data, ValidationError(None, data, error=f"Exception at array conversion: {str(e)}")
211    return arr_data, None
212
213def _validate_array(field: Field, data: np.ndarray) -> ValidationError | None:
214    """validates a field that is known to be an numpy array"""
215    if not isinstance(data, np.ndarray) or len(data.shape) == 0:
216        _type = type(data.item() if isinstance(data, np.ndarray) and len(data.shape) == 0 else data)
217        return ValidationError(field, data, error=f"Expected np.ndarray, got {_type}")
218
219    if field.dtype in (Dtype.STR, Dtype.STR_ENUM):
220        if data.dtype != "object":
221            return ValidationError(field, data, error=f"Expected dtype=object, got {data.dtype}.")
222    elif field.dtype == Dtype.INT_ENUM:
223        if data.dtype != "int32":
224            return ValidationError(field, data, error=f"Expected dtype=int32, got {data.dtype}.")
225    elif data.dtype != field.dtype:
226        return ValidationError(field, data, error=f"Expected dtype={field.dtype}, got {data.dtype}")
227
228    expected_shape = field.shape if field.shape[0] is not None else (len(data), *field.shape[1:])
229    if data.shape != expected_shape:
230        return ValidationError(field, data, error=f"Expected shape={field.shape}, got {data.shape}")
231
232    if field.min_len is not None and len(data) < field.min_len:
233        return ValidationError(field, data, error=f"Expected min len={field.min_len}, got {len(data)}")
234
235    if field.dtype == Dtype.STR:
236        pass
237
238    elif field.dtype == Dtype.INT32:
239        if field.range is not None:
240            if (data < field.range[0]).any() or (data > field.range[1]).any():
241                return ValidationError(field, data, error=f"Out of range: {data} not in {field.range}")
242
243    elif field.dtype == Dtype.FLOAT32:
244        if np.isnan(data).any():
245            return ValidationError(field, data, error=f"NaN not allowed: {data}")
246        if field.range is not None:
247            if (data < field.range[0]).any() or (data > field.range[1]).any():
248                return ValidationError(field, data, error=f"Out of range: {data} not in {field.range}")
249
250    elif field.dtype == Dtype.BOOL:
251        if np.isin(data, field.enum, invert=True).any():
252            pass
253
254    elif field.dtype in (Dtype.INT_ENUM, Dtype.STR_ENUM):
255        if np.isin(data, field.enum, invert=True).any():
256            return ValidationError(field, data, error=f"One or more of {data} not in expected values: {field.enum}")
257
258    else:
259        raise NotImplementedError(field, data)
260
261    return None
262
263def _validate_field(field: Field, data: Any) -> ValidationError | None:
264    if field.shape is not None: # array types are handled by calling this function recursively again
265        return _validate_array(field, data)
266
267    # guaranteed to be scalars below
268
269    if field.dtype == Dtype.STR:
270        if not isinstance(data, str):
271            return ValidationError(field, data, error=f"Expected {field.dtype.value}, got {type(data)}")
272        return None
273
274    elif field.dtype == Dtype.INT32:
275        if not isinstance(data, int) or isinstance(data, bool): # isinstance(True, int) == True python -_-
276            return ValidationError(field, data, error=f"Expected {field.dtype.value}, got {type(data)}")
277        if field.range is not None:
278            if not field.range[0] <= data <= field.range[1]:
279                return ValidationError(field, data, error=f"Out of range: {data} not in {field.range}")
280        return None
281
282    elif field.dtype == Dtype.FLOAT32:
283        if not isinstance(data, float):
284            return ValidationError(field, data, error=f"Expected {field.dtype.value}, got {type(data)}")
285        if math.isnan(data):
286            return ValidationError(field, data, error=f"NaN not allowed: {data}")
287        if field.range is not None:
288            if not field.range[0] <= data <= field.range[1]:
289                return ValidationError(field, data, error=f"Out of range: {data} not in {field.range}")
290        return None
291
292    elif field.dtype == Dtype.BOOL:
293        if not isinstance(data, bool):
294            return ValidationError(field, data, error=f"Expected {field.dtype.value}, got {type(data)}")
295        return None
296
297    elif field.dtype == Dtype.DICT:
298        if not isinstance(data, dict):
299            return ValidationError(field, data, error=f"Expected {field.dtype.value}, got {type(data)}")
300
301        # recursively validate each field if the dict is not opaque.
302        if field.fields is not None:
303            if (field_keys := field.fields.keys()) != data.keys():
304                return ValidationError(field, data, error=f"Wrong data keys. Expected: {field_keys}, got: {list(data)}")
305
306            for k, nested_field in field.fields.items():
307                data[k], err = _validate_array_and_convert_to_numpy_if_needed(nested_field, data[k])
308                if err is not None:
309                    return err
310
311                if (err := _validate_field(field=nested_field, data=data[k])) is not None:
312                    return err
313        return None
314
315    elif field.dtype == Dtype.BYTES:
316        if not isinstance(data, bytes):
317            return ValidationError(field, data, error=f"Expected {field.dtype.value}, got {type(data)}")
318        return None
319
320    elif field.dtype == Dtype.STR_ENUM:
321        if not isinstance(data, str):
322            return ValidationError(field, data, error=f"Expected {field.dtype.value}, got {type(data)}")
323        if data not in field.enum:
324            return ValidationError(field, data, error=f"Data {data} not in expected values: {field.enum}")
325        return None
326
327    elif field.dtype == Dtype.INT_ENUM:
328        if not isinstance(data, int) or isinstance(data, bool): # isinstance(True, int) == True python -_-
329            return ValidationError(field, data, error=f"Expected {field.dtype.value}, got {type(data)}")
330        if data not in field.enum:
331            return ValidationError(field, data, error=f"Data {data} not in expected values: {field.enum}")
332        return None
333
334    else:
335        raise NotImplementedError(field, data)
336
337def _validate_endpoint(endpoint: Endpoint, data: dict[str, Any], mode: str) -> ValidationError | None:
338    """validates and endpoint call with some data"""
339    if mode not in ("input", "output"):
340        raise ValueError(f"Unknown mode: {mode}. Expected 'input' or 'output'")
341    if not isinstance(data, dict):
342        return ValidationError(field=None, data=data, error=f"Expected dict, got {type(data)}", endpoint=endpoint)
343    endpoint_fields = endpoint.input if mode == "input" else endpoint.output
344    # in robosim we get "cmd" as part of the message, but that's not part of the spec, so we need it to not exact match
345    if (diff := (names := set(endpoint_fields.keys())).difference(data.keys())) != set():
346        return ValidationError(field=None, data=data, endpoint=endpoint,
347                               error=f"Not all keys provided. Expected: {names}. Missing: {diff}")
348    for k, field in endpoint_fields.items():
349        data[k], err = _validate_array_and_convert_to_numpy_if_needed(field, data[k])
350        if err is not None:
351            err.endpoint = endpoint
352            return err
353
354        if (err := _validate_field(field, data[k])) is not None:
355            err.endpoint = endpoint
356            return err
357    return None
358
359class Protocol:
360    """The protocol class. A protocol is a dict of endpoints, each with some fields that follow some schema/rules"""
361    def __init__(self, endpoints: dict[str, Endpoint], error_field: Field | None = None):
362        self.endpoints = endpoints
363        self.error_field = error_field or Field(dtype=Dtype.STR)
364
365    def validate_endpoint(self, endpoint: str, data: dict[str, Any], mode: str = "input") -> ValidationError | None:
366        """validates and endpoint call with some data. Note: data is auto-converted to numpy arrays, if array fields!"""
367        check = _validate_endpoint(endpoint=self.endpoints[endpoint], data=data, mode=mode)
368        if check is not None and mode == "output": # maybe the data is not validated because it's an error
369            try:
370                if _validate_field(self.error_field, data.get("error")) is None:
371                    return None
372            except Exception:
373                pass
374        return check
375
376    @staticmethod
377    def from_dict(data: dict[str, Any], **kwargs) -> Protocol:
378        """creates a protocol from a dict"""
379        return Protocol(endpoints={k: Endpoint.from_dict(name=k, data=v, **kwargs) for k, v in data.items()})
380
381    def to_dict(self) -> dict[str, Any]:
382        """saves the protocol as a dict"""
383        return {k: v.to_dict() for k, v in self.endpoints.items()}
384
385    def __repr__(self):
386        return f"[Protocol] Endpoints ({len(self.endpoints)}):\n- {'\n- '.join(map(repr, self.endpoints.values()))}"