Add flux/flows schema separation support (v2.0.0)

.gitignore

@@ -121,6 +121,8 @@ celerybeat.pid
 
 # Environments
 .env
+local.env
+local.*.env.json
 .venv
 env/
 venv/
@@ -170,3 +172,9 @@ simulation_output.csv
 simulation_output_summary.csv
 reporting_output_summary.csv
 docs/sim_examples/data/outputs
+
+# Database integration tests (contains env-specific outputs)
+dbtests/
+
+# Claude Code
+.claude/

CHANGELOG.md

@@ -0,0 +1,60 @@
+# Changelog
+
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [2.0.0] - Unreleased
+
+### Breaking Changes
+
+- **Environment configuration**: Added required `flux` section to env.json for database connections
+- The `skypro report` command now requires a `flux` database URL in the environment configuration
+
+### Added
+
+- Support for separate `flows` and `flux` database schemas
+- Configurable schema names via env.json (`schema` field in `flows` and `flux` sections)
+- Schema defaults: `flows.schema = "flows"`, `flux.schema = "flux"`
+
+### Changed
+
+- Meter readings function (`get_meter_readings_5m`) now queries from `flux` schema
+- BESS readings now query from `flux.mg_bess_readings_30m`
+- Market data (imbalance prices) now query from `flux.market_data`
+- Plot meter readings continue to query from `flows` schema tables
+
+### Migration Guide
+
+Update your env.json to include the `flux` section:
+
+```json
+{
+  "flows": { "dbUrl": "postgres://..." },
+  "flux": { "dbUrl": "postgres://...", "schema": "flux" },
+  "rates": { "dbUrl": "postgres://..." }
+}
+```
+
+For databases where all data is in the `flows` schema (legacy setup), set `flux.schema = "flows"`:
+
+```json
+{
+  "flux": { "dbUrl": "postgres://...", "schema": "flows" }
+}
+```
+
+## [1.2.0] - 2024
+
+### Added
+
+- Example configurations for simulations (self-contained examples with data)
+- LP optimiser with constraint management example
+- Multiple tagged load profiles example
+- File-based rates example
+- Peak configuration can now be disabled in price curve algorithm
+
+### Fixed
+
+- Database-based simulations now work correctly

CLAUDE.md

@@ -0,0 +1,115 @@
+# Skypro Codebase
+
+Python CLI for microgrid simulation and reporting. Published to test.pypi.org.
+
+## Project Structure
+
+```
+src/skypro/
+├── main.py              # CLI entry point
+├── commands/
+│   ├── simulator/       # `skypro simulate` command
+│   └── report/          # `skypro report` command
+├── common/
+│   └── rates/           # Rate modelling (volumetric, fixed, market, internal)
+└── reporting_webapp/    # DEPRECATED Streamlit app
+```
+
+## CLI Commands
+
+### `skypro simulate`
+```bash
+skypro simulate -c <config.yaml> -o ./output.csv --plot
+skypro simulate --help
+```
+
+Projects microgrid behaviour over historical data with configurable control strategies:
+- `perfectHindsightOptimiser` - LP-based optimal hindsight
+- `priceCurveAlgo` - Real-time price curve following
+
+### `skypro report`
+```bash
+skypro report -c <config.yaml> -m 2025-04 --plot
+skypro report --help
+```
+
+Analyzes real metering data, generates supplier invoice estimates, reports data inconsistencies as Notices.
+
+## Seven Energy Flows
+
+```
+solar_to_batt    solar_to_load    solar_to_grid
+grid_to_batt     grid_to_load
+batt_to_load     batt_to_grid
+```
+
+Each flow has associated market rates (actual £) and internal rates (notional value for optimization).
+
+## Configuration
+
+### Environment (`~/.simt/env.json`)
+
+Since v2.0.0, the environment file requires a `flux` section for the flux database:
+
+```json
+{
+  "vars": { "SKYPRO_DIR": "/path/to/skyprospector.com" },
+  "flows": { "dbUrl": "postgres://..." },
+  "flux": { "dbUrl": "postgres://...", "schema": "flux" },
+  "rates": { "dbUrl": "postgres://..." }
+}
+```
+
+**Schema configuration:**
+- `flows.schema` - defaults to "flows" (plot meter tables)
+- `flux.schema` - defaults to "flux" (meter readings function, BESS readings, market data)
+
+For legacy databases where everything is in `flows` schema, set `flux.schema = "flows"`.
+
+### Simulation Config (YAML)
+See `src/tests/integration/fixtures/simulation/config.yaml` for annotated example.
+
+Key sections:
+- `timeFrame` - start/end dates
+- `site.gridConnection` - import/export limits (kW)
+- `site.bess` - energyCapacity (kWh), nameplatePower (kW), chargeEfficiency
+- `site.solar/load` - profile sources (CSV dirs, constants, scaling)
+- `rates` - JSON/YAML files for each of the 7 flows
+- `strategy` - control algorithm config
+- `output` - CSV paths, aggregation, detail level
+
+## Development
+
+### Install
+```bash
+pip install --upgrade --extra-index-url https://test.pypi.org/simple/ skypro
+```
+
+### Run Tests
+```bash
+PYTHONPATH=src python -m unittest discover --start-directory src
+```
+
+### Publish
+1. Update version in `pyproject.toml`
+2. `poetry build`
+3. `poetry publish -r test-pypi`
+
+## Key Concepts
+
+### Rates
+- **Volumetric** (p/kWh) - DUoS, supplier fees, levies
+- **Fixed** (p/day, p/kVA/day) - standing charges
+- **Market** - actual cashflows with suppliers
+- **Internal** - opportunity cost for optimization
+
+### OSAM (P395)
+On-site Allocation Methodology for calculating final demand levies. Runs in parallel with Skypro's own methodology; discrepancies reported as Notices.
+
+### Control Strategies
+- **Perfect Hindsight LP** - Optimal solution knowing future prices
+- **Price Curve Algorithm** - Real-time with NIV chase, peak shaving, load following
+
+## Dependencies
+
+pandas, plotly, pulp, pendulum, sqlalchemy, psycopg2-binary, marshmallow, pyyaml

README.md

@@ -35,16 +35,22 @@ The structure of this file should be as follows:
     "flows": {
       "dbUrl": "url-with-credentials-for-the-flows-database"
     },
+    "flux": {
+      "dbUrl": "url-with-credentials-for-the-flux-database",
+      "schema": "flux"
+    },
     "rates": {
       "dbUrl": "url-with-credentials-for-the-rates-database"
     }
 }
 ```
 The `vars` section allows you to define arbitrary variables that are resolved in configuration file paths.
-For example, if you're configuring a simulation run and all the load profiles are in a certain directory, then you could configure a variable like `"PROFILE_DIR": "~/myprofiles"`, and then 
+For example, if you're configuring a simulation run and all the load profiles are in a certain directory, then you could configure a variable like `"PROFILE_DIR": "~/myprofiles"`, and then
 anywhere you use `$PROFILE_DIR` in the configuration it will be resolved appropriately.
 
-The `flows` section configures how to access the Flows database - this is only used if Skypro is configured to pull data from the Flows database.
+The `flows` section configures how to access the Flows database - this is only used if Skypro is configured to pull data from the Flows database (e.g., plot meter readings).
+
+The `flux` section (added in v2.0.0) configures how to access the Flux database - this is used for meter readings, BESS readings, and market data. The optional `schema` field defaults to "flux". For legacy databases where all data is in the `flows` schema, set `"schema": "flows"`.
 
 The `rates` section configures how to access the Rates database - this is only used if Skypro is configured to pull data from a Rates database.
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "skypro"
-version = "1.2.0"
+version = "2.0.0"
 description = "Skyprospector by Cepro"
 authors = ["damonrand <damon@cepro.energy>"]
 license = "AGPL-3.0"

src/skypro/commands/report/main.py

@@ -33,6 +33,20 @@
 TIMEZONE_STR = "Europe/London"
 
 
+def get_db_url(env_var_name: str, env_config: dict, config_key: str) -> str:
+    """Get DB URL from environment variable, falling back to env.json config."""
+    url = os.environ.get(env_var_name)
+    if url:
+        return url
+    return env_config[config_key]["dbUrl"]
+
+
+def get_schema_name(env_config: dict, config_key: str, default: str) -> str:
+    """Get schema name from env config, with a default fallback."""
+    config_section = env_config.get(config_key, {})
+    return config_section.get("schema", default)
+
+
 @dataclass
 class Report:
     """
@@ -104,8 +118,11 @@ def file_path_resolver_func(file: str) -> str:
     # Run the actual reporting logic
     result = report(
         config=config,
-        flows_db_url=env_config["flows"]["dbUrl"],
-        rates_db_url=env_config["rates"]["dbUrl"],
+        flows_db_url=get_db_url("FLOWS_DB_URL", env_config, "flows"),
+        rates_db_url=get_db_url("RATES_DB_URL", env_config, "rates"),
+        flux_db_url=get_db_url("FLUX_DB_URL", env_config, "flux"),
+        flows_schema=get_schema_name(env_config, "flows", default="flows"),
+        flux_schema=get_schema_name(env_config, "flux", default="flux"),
         start=start,
         end=end,
         step_size=step_size,
@@ -179,6 +196,9 @@ def report(
     config: Config,
     flows_db_url: str,
     rates_db_url: str,
+    flux_db_url: str,
+    flows_schema: str,
+    flux_schema: str,
     start: datetime,
     end: datetime,
     step_size: timedelta,
@@ -201,12 +221,14 @@ def report(
         file_path_resolver_func=file_path_resolver_func,
         flows_db_engine=flows_db_url,
         rates_db_engine=rates_db_url,
+        flux_db_engine=flux_db_url,
+        flux_schema=flux_schema,
     )
     notices.extend(new_notices)
 
     _log_rates_to_screen(rates, time_index)
 
-    readings, new_notices = get_readings(config, time_index, flows_db_url, file_path_resolver_func)
+    readings, new_notices = get_readings(config, time_index, flows_db_url, flux_db_url, flows_schema, flux_schema, file_path_resolver_func)
     notices.extend(new_notices)
 
     df, new_notices = calc_flows(

src/skypro/commands/report/rates.py

@@ -32,6 +32,8 @@ def get_rates_from_config(
         file_path_resolver_func: Callable,
         flows_db_engine,
         rates_db_engine,
+        flux_db_engine,
+        flux_schema: str = "flux",
 ) -> Tuple[ParsedRates, List[Notice]]:
     """
     This reads the rates configuration block and returns the ParsedRates, and a list of Notices if there are issues with data quality.
@@ -46,7 +48,8 @@ def get_rates_from_config(
         start=time_index[0],
         end=time_index[-1],
         file_path_resolver_func=file_path_resolver_func,
-        db_engine=flows_db_engine,
+        db_engine=flux_db_engine,  # Market data is in flux schema
+        schema=flux_schema,
         context="elexon imbalance data"
     )
     notices.extend(new_notices)

src/skypro/commands/report/readings.py

@@ -25,7 +25,7 @@ class AllReadings:
     plot_meter: pd.DataFrame
 
 
-def get_readings(config, time_index, flows_db_url, file_path_resolver_func) -> Tuple[AllReadings, List[Notice]]:
+def get_readings(config, time_index, flows_db_url, flux_db_url, flows_schema, flux_schema, file_path_resolver_func) -> Tuple[AllReadings, List[Notice]]:
     """
     Pulls in the various meter and BESS readings (these will be resolved either from DB or local CSV files)
     """
@@ -39,7 +39,8 @@ def _get_meter_readings_wrapped(source: MeterReadingDataSource, context: str) ->
             start=time_index[0],
             end=time_index[-1],
             file_path_resolver_func=file_path_resolver_func,
-            db_engine=flows_db_url,
+            db_engine=flux_db_url,  # Meter readings function is in flux schema
+            schema=flux_schema,
             context=context
         )
         notices.extend(new_notices_2)
@@ -61,6 +62,7 @@ def _get_meter_readings_wrapped(source: MeterReadingDataSource, context: str) ->
         end=time_index[-1],
         file_path_resolver_func=file_path_resolver_func,
         db_engine=flows_db_url,
+        schema=flows_schema,
         context="plot meter readings"
     )
     notices.extend(new_notices)
@@ -71,7 +73,8 @@ def _get_meter_readings_wrapped(source: MeterReadingDataSource, context: str) ->
         start=time_index[0],
         end=time_index[-1],
         file_path_resolver_func=file_path_resolver_func,
-        db_engine=flows_db_url,
+        db_engine=flux_db_url,  # BESS readings are in flux schema
+        schema=flux_schema,
         context="bess readings"
     )
     notices.extend(new_notices)

src/skypro/common/data/get_bess_readings.py

@@ -15,7 +15,8 @@ def get_bess_readings(
         end: Optional[datetime],
         file_path_resolver_func: Optional[Callable],
         db_engine: Optional,
-        context: Optional[str],
+        schema: str = "flux",
+        context: Optional[str] = None,
 ) -> Tuple[pd.DataFrame, List[Notice]]:
     """
     This reads a data source - either CSVs from disk or directly from a database - and returns BESS readings in a dataframe alongside a list of warnings.
@@ -33,7 +34,8 @@ def get_bess_readings(
             source=source.flows_bess_readings_data_source,
             start=start,
             end=end,
-            db_engine=db_engine
+            db_engine=db_engine,
+            schema=schema
         )
     elif source.csv_bess_readings_data_source:
         df = _get_csv_bess_readings(
@@ -52,15 +54,16 @@ def _get_flows_bess_readings(
         source: FlowsBessReadingsDataSource,
         start: datetime,
         end: datetime,
-        db_engine
+        db_engine,
+        schema: str = "flux"
 ) -> pd.DataFrame:
     """
     Pulls readings about the identified BESS from the mg_bess_readings table.
     """
 
     query = (
         f"SELECT time_b, device_id, soe_avg, target_power_avg "
-        "FROM flows.mg_bess_readings_30m WHERE "
+        f"FROM {schema}.mg_bess_readings_30m WHERE "
         f"time_b >= '{start.isoformat()}' AND "
         f"time_b < '{end.isoformat()}' AND "
         f"device_id = '{source.bess_id}' "