Include optimization scalar

.pre-commit-config.yaml

@@ -6,10 +6,10 @@ repos:
     entry: bash -c "poetry run mypy ."
     language: system
 - repo: https://github.com/psf/black
-  rev: 23.10.0
+  rev: 23.12.1
   hooks:
   - id: black
 - repo: https://github.com/astral-sh/ruff-pre-commit
-  rev: v0.1.0
+  rev: v0.1.11
   hooks:
   - id: ruff

ixmp4/__init__.py

@@ -8,6 +8,8 @@
 from ixmp4.core import Scenario as Scenario
 from ixmp4.core import Unit as Unit
 from ixmp4.core import Variable as Variable
+from ixmp4.core import IndexSet as IndexSet
+from ixmp4.core import Scalar as Scalar
 from ixmp4.core.exceptions import InconsistentIamcType as InconsistentIamcType
 from ixmp4.core.exceptions import IxmpError as IxmpError
 from ixmp4.core.exceptions import NotFound as NotFound

ixmp4/cli/__init__.py

@@ -48,7 +48,10 @@ def login(
         user = auth.get_user()
         utils.good(f"Successfully authenticated as user '{user.username}'.")
         if typer.confirm(
-            "Are you sure you want to save your credentials in plain-text for future use?"
+            text=(
+                "Are you sure you want to save your credentials in plain-text for "
+                "future use?"
+            )
         ):
             settings.credentials.set("default", username, password)
 

ixmp4/cli/platforms.py

@@ -40,8 +40,8 @@ def prompt_sqlite_dsn(name: str):
     dsn = sqlite.get_dsn(path)
     if path.exists():
         if typer.confirm(
-            f"A file at the standard filesystem location for name '{name}' already exists. "
-            "Do you want to add the existing file to the platform registry?"
+            f"A file at the standard filesystem location for name '{name}' already "
+            "exists. Do you want to add the existing file to the platform registry?"
         ):
             return dsn
         else:
@@ -66,7 +66,7 @@ def add(
     ),
     dsn: Optional[str] = typer.Option(
         None,
-        help="A data source name. Can be a http(s) URl or a database connection string.",
+        help="Data source name. Can be a http(s) URL or a database connection string.",
         callback=validate_dsn,
     ),
 ):
@@ -114,7 +114,8 @@ def remove(
         raise typer.BadParameter(f"Platform '{name}' does not exist.")
 
     if typer.confirm(
-        f"Are you sure you want to remove the platform '{platform.name}' with dsn '{platform.dsn}'?"
+        f"Are you sure you want to remove the platform '{platform.name}' with dsn "
+        f"'{platform.dsn}'?"
     ):
         if platform.dsn.startswith("sqlite://"):
             prompt_sqlite_removal(platform.dsn)
@@ -155,7 +156,10 @@ def list_():
 
 
 @app.command(
-    help="Migrates all database platforms from your local toml file to the newest revision."
+    help=(
+        "Migrates all database platforms from your local toml file to the newest "
+        "revision."
+    )
 )
 def upgrade():
     for c in settings.toml.list_platforms():
@@ -167,7 +171,10 @@ def upgrade():
 
 
 @app.command(
-    help="Stamps all database platforms from your local toml file with the given revision."
+    help=(
+        "Stamps all database platforms from your local toml file with the given "
+        "revision."
+    )
 )
 def stamp(revision: str) -> None:
     for c in settings.toml.list_platforms():

ixmp4/core/__init__.py

@@ -2,6 +2,7 @@
 from .iamc.variable import Variable as Variable
 from .model import Model as Model
 from .optimization.indexset import IndexSet as IndexSet
+from .optimization.scalar import Scalar as Scalar
 from .platform import Platform as Platform
 from .region import Region as Region
 from .run import Run as Run

ixmp4/core/optimization/data.py

@@ -1,21 +1,18 @@
-from functools import partial
-
 from ixmp4.data.abstract import Run
 
 from ..base import BaseFacade
-from .indexset import IndexSet as IndexSetModel
 from .indexset import IndexSetRepository
+from .scalar import ScalarRepository
 
 
 class OptimizationData(BaseFacade):
     """An optimization data instance, which provides access to optimization data such as
     IndexSet, Table, Variable, etc."""
 
-    IndexSet: partial[IndexSetModel]
-
     indexsets: IndexSetRepository
+    scalars: ScalarRepository
 
     def __init__(self, *args, run: Run, **kwargs) -> None:
         super().__init__(*args, **kwargs)
-        self.IndexSet = partial(IndexSetModel, _backend=self.backend, _run=run)
         self.indexsets = IndexSetRepository(_backend=self.backend, _run=run)
+        self.scalars = ScalarRepository(_backend=self.backend, _run=run)

ixmp4/core/optimization/indexset.py

@@ -11,30 +11,9 @@
 
 class IndexSet(BaseModelFacade):
     _model: IndexSetModel
-    _run: Run
     NotFound: ClassVar = IndexSetModel.NotFound
     NotUnique: ClassVar = IndexSetModel.NotUnique
 
-    def __init__(
-        self,
-        name: str,
-        _run: Run,
-        **kwargs,
-    ) -> None:
-        super().__init__(**kwargs)
-        self._run = _run
-        if getattr(self, "_model", None) is None:
-            try:
-                self._model = self.backend.optimization.indexsets.get(
-                    run_id=self._run.id, name=name
-                )
-                # TODO: provide logging information if IndexSet already exists
-            except IndexSetModel.NotFound:
-                self._model = self.backend.optimization.indexsets.create(
-                    run_id=self._run.id,
-                    name=name,
-                )
-
     @property
     def id(self) -> int:
         return self._model.id
@@ -53,12 +32,12 @@ def add(self, elements: int | list[int | str] | str) -> None:
             indexset_id=self._model.id, elements=elements
         )
         self._model.elements = self.backend.optimization.indexsets.get(
-            run_id=self._run.id, name=self._model.name
+            run_id=self._model.run__id, name=self._model.name
         ).elements
 
     @property
     def run_id(self) -> int:
-        return self._run.id
+        return self._model.run__id
 
     @property
     def created_at(self) -> datetime | None:
@@ -101,14 +80,25 @@ def __init__(self, _run: Run, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
         self._run = _run
 
+    def create(self, name: str) -> IndexSet:
+        indexset = self.backend.optimization.indexsets.create(
+            run_id=self._run.id,
+            name=name,
+        )
+        return IndexSet(_backend=self.backend, _model=indexset)
+
+    def get(self, name: str) -> IndexSet:
+        indexset = self.backend.optimization.indexsets.get(
+            run_id=self._run.id, name=name
+        )
+        return IndexSet(_backend=self.backend, _model=indexset)
+
     def list(self, name: str | None = None) -> Iterable[IndexSet]:
         indexsets = self.backend.optimization.indexsets.list(name=name)
         return [
             IndexSet(
                 _backend=self.backend,
                 _model=i,
-                _run=self._run,
-                name=i.name,
             )
             for i in indexsets
         ]

ixmp4/core/optimization/scalar.py

@@ -0,0 +1,140 @@
+from datetime import datetime
+from typing import ClassVar, Iterable
+
+import pandas as pd
+
+from ixmp4.core.base import BaseFacade, BaseModelFacade
+from ixmp4.core.unit import Unit
+from ixmp4.data.abstract import Docs as DocsModel
+from ixmp4.data.abstract import Run
+from ixmp4.data.abstract import Scalar as ScalarModel
+
+
+class Scalar(BaseModelFacade):
+    _model: ScalarModel
+    NotFound: ClassVar = ScalarModel.NotFound
+    NotUnique: ClassVar = ScalarModel.NotUnique
+
+    @property
+    def id(self) -> int:
+        return self._model.id
+
+    @property
+    def name(self) -> str:
+        return self._model.name
+
+    @property
+    def value(self) -> float:
+        """Associated value."""
+        return self._model.value
+
+    @value.setter
+    def value(self, value: float):
+        self._model.value = value
+        self.backend.optimization.scalars.update(
+            id=self._model.id,
+            value=self._model.value,
+            unit_id=self._model.unit.id,
+        )
+
+    @property
+    def unit(self):
+        """Associated unit."""
+        return self._model.unit
+
+    @unit.setter
+    def unit(self, unit: str | Unit):
+        if isinstance(unit, Unit):
+            unit = unit
+        else:
+            unit_model = self.backend.units.get(unit)
+            unit = Unit(_backend=self.backend, _model=unit_model)
+        self._model = self.backend.optimization.scalars.update(
+            id=self._model.id,
+            value=self._model.value,
+            unit_id=unit.id,
+        )
+
+    @property
+    def run_id(self) -> int:
+        return self._model.run__id
+
+    @property
+    def created_at(self) -> datetime | None:
+        return self._model.created_at
+
+    @property
+    def created_by(self) -> str | None:
+        return self._model.created_by
+
+    @property
+    def docs(self):
+        try:
+            return self.backend.optimization.scalars.docs.get(self.id).description
+        except DocsModel.NotFound:
+            return None
+
+    @docs.setter
+    def docs(self, description):
+        if description is None:
+            self.backend.optimization.scalars.docs.delete(self.id)
+        else:
+            self.backend.optimization.scalars.docs.set(self.id, description)
+
+    @docs.deleter
+    def docs(self):
+        try:
+            self.backend.optimization.scalars.docs.delete(self.id)
+        # TODO: silently failing
+        except DocsModel.NotFound:
+            return None
+
+    def __str__(self) -> str:
+        return f"<Scalar {self.id} name={self.name}>"
+
+
+class ScalarRepository(BaseFacade):
+    _run: Run
+
+    def __init__(self, _run: Run, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self._run = _run
+
+    def create(self, name: str, value: float, unit: str | Unit | None = None) -> Scalar:
+        if isinstance(unit, Unit):
+            unit_name = unit.name
+        elif isinstance(unit, str):
+            unit_name = unit
+        else:
+            # TODO: provide logging information about None-units being converted
+            # to dimensionless
+            dimensionless_unit = self.backend.units.create(name="dimensionless")
+            unit_name = dimensionless_unit.name
+
+        try:
+            model = self.backend.optimization.scalars.create(
+                name=name, value=value, unit_name=unit_name, run_id=self._run.id
+            )
+        except Scalar.NotUnique as e:
+            raise Scalar.NotUnique(
+                message=f"Scalar '{name}' already exists! Did you mean to call "
+                "run.optimization.scalars.update()?"
+            ) from e
+        return Scalar(_backend=self.backend, _model=model)
+
+    def get(self, name: str) -> Scalar:
+        model = self.backend.optimization.scalars.get(run_id=self._run.id, name=name)
+        return Scalar(_backend=self.backend, _model=model)
+
+    def list(self, name: str | None = None) -> Iterable[Scalar]:
+        scalars = self.backend.optimization.scalars.list(run_id=self._run.id, name=name)
+        return [
+            Scalar(
+                _backend=self.backend,
+                _model=i,
+            )
+            for i in scalars
+        ]
+
+    def tabulate(self, name: str | None = None) -> pd.DataFrame:
+        return self.backend.optimization.scalars.tabulate(name=name)

ixmp4/core/run.py

@@ -52,7 +52,8 @@ def meta(self, meta):
         self._meta._set(meta)
 
     def set_as_default(self):
-        """Sets this run as default version for this `model - scenario` combination."""
+        """Sets this run as the default version for its `model` + `scenario`
+        combination."""
         self.backend.runs.set_as_default_version(self._model.id)
 
     def unset_as_default(self):

ixmp4/data/abstract/__init__.py

@@ -29,7 +29,7 @@
 )
 from .meta import MetaValue, RunMetaEntry, RunMetaEntryRepository, StrictMetaValue
 from .model import Model, ModelRepository
-from .optimization import IndexSet, IndexSetRepository
+from .optimization import IndexSet, IndexSetRepository, Scalar, ScalarRepository
 from .region import Region, RegionRepository
 from .run import Run, RunRepository
 from .scenario import Scenario, ScenarioRepository

ixmp4/data/abstract/docs.py

@@ -15,7 +15,7 @@ class Docs(base.BaseModel, Protocol):
     dimension: types.Mapped
     "The documented object."
 
-    # This doesn't work since each dimension has a different self.dimension object as of now
+    # This doesn't work since each dimension has a different self.dimension object atm
     # def __str__(self) -> str:
     #    return (
     #        f"<Documentation for {self.dimension} {self.dimension.name}>"
@@ -74,8 +74,8 @@ def delete(self, dimension_id: int) -> None:
         Parameters
         ----------
         dimension_id : int
-            The unique id of the object whose documentation should be deleted in its dimension's
-            table.
+            The unique id of the object whose documentation should be deleted in its
+            dimension's table.
 
         Raises
         ------
@@ -94,7 +94,8 @@ def list(
         Parameters
         ----------
         dimension_id : int
-            The id of an object in any dimension. If supplied only one result will be returned.
+            The id of an object in any dimension. If supplied only one result will be
+            returned.
 
         Returns
         -------