JSON serialization

Lossless round-trip of TSeries / MVTSeries / Workspace / MIT / MITRange / Frequency to and from JSON. Each composite type carries a _type discriminator so the decoder can rebuild the right class.

tsecon.io.json

JSON round-trip for MIT, Duration, MITRange, TSeries, and Workspace.

The Julia serialize.jl is a binary protocol coupled to Serialization.jl; it has no JSON. We invent the schema here: every tsecon object is encoded as a JSON object with a _type discriminator, so a decoder can reconstruct the right Python type without ambiguity.

Schema (stable starting from this version — a versioned _schema field will be added if the schema ever needs to change):

• MIT → {"_type": "MIT", "freq": <Freq>, "value": <int>}
• Duration → {"_type": "Duration", "freq": <Freq>, "value": <int>}
• MITRange → {"_type": "MITRange", "start": <MIT>, "stop": <MIT>, "step": <int>}
• TSeries → {"_type": "TSeries", "firstdate": <MIT>, "dtype": <str>, "values": [<scalar>...]}
• MVTSeries → {"_type": "MVTSeries", "firstdate": <MIT>, "dtype": <str>, "names": [<str>, ...], "values": [[<scalar>, ...], ...]} (row-major)
• Workspace → {"_type": "Workspace", "items": [[<key>, <value>], ...]}

Frequencies are encoded as {"name": <class>, ...}, with the extra parameter where applicable (end_month for Yearly/HalfYearly/Quarterly, end_day for Weekly).

NaN / +Inf / -Inf are emitted as JavaScript-style literals (NaN / Infinity / -Infinity), which is the Python json module's default when allow_nan=True. Strictly-valid-JSON consumers can pass allow_nan=False and supply their own sentinel handling.

to_jsonable

to_jsonable(obj: Any) -> Any

Convert a tsecon object (or nested structure) to a plain JSON-friendly tree.

Pure Python int, float, bool, str, None, list, and dict values pass through unchanged. NumPy scalars are unwrapped via .item(). NumPy arrays become Python lists.

Source code in src/tsecon/io/json.py

def to_jsonable(obj: Any) -> Any:
    """Convert a tsecon object (or nested structure) to a plain JSON-friendly tree.

    Pure Python ``int``, ``float``, ``bool``, ``str``, ``None``, ``list``, and
    ``dict`` values pass through unchanged. NumPy scalars are unwrapped via
    ``.item()``. NumPy arrays become Python lists.
    """
    if isinstance(obj, MIT):
        return {
            "_type": "MIT",
            "freq": _freq_to_dict(obj.frequency),
            "value": int(obj.value),
        }
    if isinstance(obj, Duration):
        return {
            "_type": "Duration",
            "freq": _freq_to_dict(obj.frequency),
            "value": int(obj.value),
        }
    if isinstance(obj, MITRange):
        return {
            "_type": "MITRange",
            "start": to_jsonable(obj.start),
            "stop": to_jsonable(obj.stop),
            "step": int(obj.step),
        }
    if isinstance(obj, TSeries):
        return {
            "_type": "TSeries",
            "firstdate": to_jsonable(obj.firstdate),
            "dtype": str(obj.values.dtype),
            "values": obj.values.tolist(),
        }
    if isinstance(obj, MVTSeries):
        return {
            "_type": "MVTSeries",
            "firstdate": to_jsonable(obj.firstdate),
            "dtype": str(obj.values.dtype),
            "names": list(obj.column_names),
            "values": obj.values.tolist(),
        }
    if isinstance(obj, Workspace):
        return {
            "_type": "Workspace",
            "items": [[k, to_jsonable(v)] for k, v in obj.items()],
        }
    if isinstance(obj, Frequency):
        return {"_type": "Frequency", "freq": _freq_to_dict(obj)}
    if isinstance(obj, np.ndarray):
        return obj.tolist()
    if isinstance(obj, np.generic):
        return obj.item()
    if isinstance(obj, dict):
        return {str(k): to_jsonable(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [to_jsonable(x) for x in obj]
    if obj is None or isinstance(obj, (bool, int, float, str)):
        return obj
    msg = f"Cannot JSON-encode value of type {type(obj).__name__}."
    raise TypeError(msg)

from_jsonable

from_jsonable(obj: Any) -> Any

Inverse of :func:to_jsonable: rebuild tsecon objects from a JSON tree.

Source code in src/tsecon/io/json.py

def from_jsonable(obj: Any) -> Any:
    """Inverse of :func:`to_jsonable`: rebuild tsecon objects from a JSON tree."""
    if isinstance(obj, dict):
        tag = obj.get("_type")
        if tag is not None:
            return _decode_tagged(tag, obj)
        return {k: from_jsonable(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [from_jsonable(x) for x in obj]
    return obj

dumps

dumps(
    obj: Any,
    *,
    indent: int | None = None,
    allow_nan: bool = True,
) -> str

Serialize a tsecon (or plain) value to a JSON string.

Source code in src/tsecon/io/json.py

def dumps(obj: Any, *, indent: int | None = None, allow_nan: bool = True) -> str:
    """Serialize a tsecon (or plain) value to a JSON string."""
    return json.dumps(to_jsonable(obj), indent=indent, allow_nan=allow_nan)

loads

loads(s: str | bytes | bytearray) -> Any

Parse a JSON string and rebuild tsecon objects from any _type-tagged nodes.

Source code in src/tsecon/io/json.py

def loads(s: str | bytes | bytearray) -> Any:
    """Parse a JSON string and rebuild tsecon objects from any ``_type``-tagged nodes."""
    return from_jsonable(json.loads(s))

dump

dump(
    obj: Any,
    fp: IO[str],
    *,
    indent: int | None = None,
    allow_nan: bool = True,
) -> None

Write a tsecon (or plain) value as JSON to a writable text stream.

Source code in src/tsecon/io/json.py

def dump(obj: Any, fp: IO[str], *, indent: int | None = None, allow_nan: bool = True) -> None:
    """Write a tsecon (or plain) value as JSON to a writable text stream."""
    json.dump(to_jsonable(obj), fp, indent=indent, allow_nan=allow_nan)

load

load(fp: IO[str]) -> Any

Read a JSON document from a text stream and rebuild tsecon objects.

Source code in src/tsecon/io/json.py

def load(fp: IO[str]) -> Any:
    """Read a JSON document from a text stream and rebuild tsecon objects."""
    return from_jsonable(json.load(fp))