Skip to content

Instantly share code, notes, and snippets.

@danoneata
Created June 7, 2022 14:07
Show Gist options
  • Save danoneata/30a15130f92d95ec0c67bf5026add52a to your computer and use it in GitHub Desktop.
Save danoneata/30a15130f92d95ec0c67bf5026add52a to your computer and use it in GitHub Desktop.
JSON decoder in Python
from main import *
@dataclass
class Measurment:
temp: float
humidity: float
# date: datetime
@dataclass
class Sensor:
name: str
location: str
uptime: float
# charging: bool
readings: List[Measurment]
data = {
"name": "sigma sensor",
"location": "district 9",
"uptime": "39",
"charging": "yes",
"readings": [
{"temp": "20", "humidity": "0.4", "date": "2022/06/12"},
{"temp": "25", "humidity": "0.45", "date": "July 03 2021"},
{"temp": "30", "humidity": "0.39", "date": "2020-05-03"},
],
}
dec_m = decoders_to(Measurment, key("temp"), key("humidity"))
dec_s = decoders_to(
Sensor,
key("name"),
key("location"),
key("uptime") >> float_,
key("readings") >> dec_m.list(),
)
print(dec_s(data))
from dataclasses import dataclass [8/618]
from typing import Any, Callable, Dict, Generic, List, Iterable, Tuple, TypeVar
A = TypeVar("A")
B = TypeVar("B")
@dataclass
class Decoder(Generic[A]):
decode: Callable[[Dict], A]
def __call__(self, data: Dict) -> A:
return self.decode(data)
def fmap(self, f: Callable[[A], B]) -> "Decoder[B]":
return Decoder(lambda data: f(self(data)))
def __rshift__(self: "Decoder[Dict]", other: "Decoder[B]"):
"""Sequence two decoders. The first decoder should return a dictionary."""
return Decoder(lambda data: other(self(data)))
def __pow__(self, other: "Decoder[B]") -> "Decoder[Tuple[A, B]]":
"""Combine two decoders."""
return combine2(self, other)
def to(self, cls: object) -> "Decoder[Any]":
return self.fmap(lambda args: cls(*args))
def list(self):
return Decoder(lambda data: [self(datum) for datum in data])
def lift(a) -> Decoder:
return Decoder(lambda _: a)
identity: Decoder[Any] = Decoder(lambda data: data)
def combine2(d1: Decoder[A], d2: Decoder[B]) -> Decoder[Tuple[A, B]]:
return Decoder(lambda data: (d1(data), d2(data)))
def combinen(decoders: Iterable[Decoder]) -> Decoder[Tuple]:
return Decoder(lambda data: tuple(d(data) for d in decoders))
def decoders_to(cls, *decoders) -> Decoder:
return combinen(decoders).to(cls)
def key(k: str) -> Decoder[Dict]:
return Decoder(lambda data: data[k])
def index(i: int) -> Decoder[Dict]:
return Decoder(lambda data: data[i])
float_: Decoder[float] = Decoder(lambda data: float(data))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment