store objects to postgresql

This commit is contained in:
Hunter 2024-07-30 17:50:08 -04:00
parent 8d9b371af4
commit 79c170b73e
3 changed files with 81 additions and 110 deletions

2
.gitignore vendored
View File

@ -1,2 +1,2 @@
__pycache__/ __pycache__/
.env

View File

@ -1,85 +1,62 @@
from dataclasses import dataclass, field from sqlalchemy import Column, String, Float, Integer, DateTime, ForeignKey
from typing import List, Any from sqlalchemy.orm import relationship, declarative_base
from dataclasses import dataclass
Interval = tuple[float, float] | None Base = declarative_base()
@dataclass @dataclass
class Measurement: class Measurement(Base):
"""These are the fields that represent the outputs of the experiment.""" __tablename__ = "measurements"
name = Column(String, nullable=False)
unit = Column(String, nullable=False)
experiment = relationship("Experiment", back_populates="measurements")
experiment_id = Column(Integer, ForeignKey("experiments.id"))
values = relationship("MeasurementValue", back_populates="measurement")
id = Column(Integer, primary_key=True, autoincrement=True)
name: str def take_measurement(self, value: float):
unit: str self.values.append(MeasurementValue(value=value))
interval: Interval = None
values: List[float] = field(default_factory=list)
@dataclass @dataclass
class Variable: class MeasurementValue(Base):
"""These are the fields that represent the inputs of the experiment.""" __tablename__ = "measurement_values"
value = Column(Float, nullable=False)
name: str measurement = relationship("Measurement", back_populates="values")
unit: str measurement_id = Column(Integer, ForeignKey("measurements.id"))
interval: Interval = None id = Column(Integer, primary_key=True, autoincrement=True)
value: Any = None
@dataclass @dataclass
class Experiment: class Variable(Base):
"""This is the data class that represents the experiment.""" __tablename__ = "variables"
name = Column(String, nullable=False)
unit = Column(String, nullable=False)
value = Column(Integer, nullable=False) # Storing Any type as string
experiment = relationship("Experiment", back_populates="variables")
experiment_id = Column(Integer, ForeignKey("experiments.id"))
id = Column(Integer, primary_key=True, autoincrement=True)
experiment: str
date: str
measurements: dict[str, Measurement] = field(default_factory=dict)
variables: dict[str, Variable] = field(default_factory=dict)
def to_json(self): @dataclass
# Convert the data class to a dictionary, which can be easily converted to JSON. class Experiment(Base):
import json __tablename__ = "experiments"
experiment = Column(String, nullable=False)
date = Column(DateTime, nullable=False)
measurements = relationship("Measurement", back_populates="experiment")
variables = relationship("Variable", back_populates="experiment")
id = Column(Integer, primary_key=True, autoincrement=True)
return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4) def get_measurement(self, source: str):
# Get a measurement from the experiment.
for m in self.measurements:
if m.name == source:
return m
raise ValueError(f"Measurement {source} not found in the experiment.")
def from_json(self, json_str): def take_measurement(self, source: str | Measurement, value: float):
# Convert a JSON string to a data class. measurement: Measurement = (
import json source if isinstance(source, Measurement) else self.get_measurement(source)
)
data = json.loads(json_str) measurement.take_measurement(value)
self.__dict__.update(data)
def add_measurement(self, measurement: Measurement):
# Add a measurement to the experiment.
if measurement.name in self.measurements:
raise ValueError(
f"Measurement {measurement.name} already exists in the experiment."
)
self.measurements[measurement.name] = measurement
def add_measurements(self, measurements: List[Measurement]):
# Add a list of measurements to the experiment.
for measurement in measurements:
self.add_measurement(measurement)
def take_measurement(self, source: str, value: float):
# Add a measurement to the experiment.
if source not in self.measurements:
raise ValueError(f"Measurement {source} not found in the experiment.")
self.measurements[source].values.append(value)
def add_variable(self, variable: Variable):
# Add a variable to the experiment.
if variable.name in self.variables:
raise ValueError(
f"Variable {variable.name} already exists in the experiment."
)
self.variables[variable.name] = variable
def add_variables(self, variables: List[Variable]):
# Add a list of variables to the experiment.
for variable in variables:
self.add_variable(variable)
def set_variable(self, source: str, value: Any):
# Set a variable in the experiment.
if source not in self.variables:
raise ValueError(f"Variable {source} not found in the experiment.")
self.variables[source].value = value

View File

@ -1,43 +1,37 @@
from dataclasses import dataclass
from experiment import Experiment, Measurement, Variable from experiment import Experiment, Measurement, Variable
@dataclass def make_shelling_experiment(
class ShellingExperiment(Experiment): date: str,
def __init__( drum_rpm: int,
self, paddle_shaft_rpm: int,
date: str, ring_gap: float,
drum_rpm: int, tilt_angle: float,
paddle_shaft_rpm: int, moisture_content: float,
ring_gap: float, feed_rate: int = 500,
tilt_angle: float, pecan_variety_index: int = 0,
moisture_content: float, ) -> Experiment:
feed_rate: int = 500,
pecan_variety: str = "desirable",
):
super().__init__(experiment="Shelling Analysis", date=date) return Experiment(
date=date,
variables = [ experiment="shelling",
Variable("drum-rpm", "rpm", (30, 40), drum_rpm), variables=[
Variable("paddle-shaft-rpm", "rpm", (400, 800), paddle_shaft_rpm), Variable(name="drum-rpm", unit="rpm", value=drum_rpm),
Variable("ring-gap", "in", value=ring_gap), Variable(name="paddle-shaft-rpm", unit="rpm", value=paddle_shaft_rpm),
Variable("tilt-angle", "deg", (2, 5), tilt_angle), Variable(name="ring-gap", unit="in", value=ring_gap),
Variable("feed-rate", "lb/hr", (300, 500), feed_rate), Variable(name="tilt-angle", unit="deg", value=tilt_angle),
Variable("moisture-content", "%", (5, 9), moisture_content), Variable(name="feed-rate", unit="lb/hr", value=feed_rate),
Variable("pecan-variety", "", value=pecan_variety), Variable(name="moisture-content", unit="%", value=moisture_content),
] Variable(name="pecan-variety", unit="", value=pecan_variety_index),
self.add_variables(variables) ],
measurements=[
measurements = [ Measurement(name="bin1-weight", unit="lb"),
Measurement("bin1-weight", "lb"), Measurement(name="bin2-weight", unit="lb"),
Measurement("bin2-weight", "lb"), Measurement(name="bin3-weight", unit="lb"),
Measurement("bin3-weight", "lb"), Measurement(name="recirculated-weight", unit="lb"),
Measurement("recirculated-weight", "lb"), Measurement(name="final-discharge-weight", unit="lb"),
Measurement("final-discharge-weight", "lb"), Measurement(name="bin1-half-yield", unit="%"),
Measurement("bin1-half-yield", "%"), Measurement(name="bin2-half-yield", unit="%"),
Measurement("bin2-half-yield", "%"), Measurement(name="bin3-half-yield", unit="%"),
Measurement("bin3-half-yield", "%"), ],
] )
self.add_measurements(measurements)