Add first visualisation revision

Caveats:
UI hangs while selecting recipes
No calculation is done based on input resource flow

Pipfile

@@ -7,6 +7,7 @@ name = "pypi"
 selenium = "*"
 click = "*"
 sqlalchemy = "*"
+nodegraphqt = "*"
 
 [dev-packages]
 
@@ -15,3 +16,4 @@ python_version = "3.11"
 
 [scripts]
 fetch = {call = "factorygame.data.fetch:main()"}
+vis = {call = "factorygame.data.vis:main()"}

Pipfile.lock

@@ -1,7 +1,7 @@
 {
     "_meta": {
         "hash": {
-            "sha256": "45cb685b915b5e5bab1ba5339a033373d210e09d6f0b4d9ac9acbe8fe15b5402"
+            "sha256": "0d398759a0afc9f55cbd2e892bf1f819b572167c2396e2b978515124ada547a9"
         },
         "pipfile-spec": 6,
         "requires": {
@@ -121,6 +121,15 @@
             "markers": "python_version >= '3.5'",
             "version": "==3.6"
         },
+        "nodegraphqt": {
+            "hashes": [
+                "sha256:2f0e4b0a2c1a3360deaa2fb6cc42cbf0dce25a7076d036473773d58b0f4fae31",
+                "sha256:97796bf36845c7e0413e72608507401b04e92106de620b804a234d0cad26e24f"
+            ],
+            "index": "pypi",
+            "markers": "python_version >= '3.6'",
+            "version": "==0.6.35"
+        },
         "outcome": {
             "hashes": [
                 "sha256:9dcf02e65f2971b80047b377468e72a268e15c0af3cf1238e6ff14f7f91143b8",
@@ -137,6 +146,12 @@
             ],
             "version": "==1.7.1"
         },
+        "qt.py": {
+            "hashes": [
+                "sha256:ab072ac955bdc9318966c078f1bb5941e77166f9e92df3db5726b1eb5afea83b"
+            ],
+            "version": "==1.3.9"
+        },
         "selenium": {
             "hashes": [
                 "sha256:5aee79026c07985dc1b0c909f34084aa996dfe5b307602de9016d7a621a473f2",
@@ -233,6 +248,13 @@
             "markers": "python_version >= '3.7'",
             "version": "==0.11.1"
         },
+        "types-pyside2": {
+            "hashes": [
+                "sha256:5bc2763bc6b595b2c5fc1191ce5147dcc6f44d56efdc2b264b6c082282cc1c57",
+                "sha256:de4b575e57fdb9e5fd8507537cdd770866fdbe1972eadc3793b2026eaec92875"
+            ],
+            "version": "==5.15.2.1.6"
+        },
         "typing-extensions": {
             "hashes": [
                 "sha256:23478f88c37f27d76ac8aee6c905017a143b0b1b886c3c9f66bc2fd94f9f5783",

README.md

@@ -5,3 +5,16 @@ Using pipenv, run e.g.
 ```sh
 pipenv run fetch --result 'Molten Iron'
 ```
+
+## Visualisation
+
+Thanks to [NodeGraphQt](https://github.com/jchanvfx/NodeGraphQt) a graph base visualisation is available, which looks like the following:
+
+[!Graph Visualisation Example](vis.png)
+
+The visualisation window can be opened with e.g.
+
+```sh
+pipenv run fetch --result 'Plastic'
+pipenv run vis 'Plastic'
+```

factorygame/data/fetch.py

@@ -1,3 +1,5 @@
+#!/usr/bin/env python3
+
 import datetime
 from typing import Optional
 
@@ -7,6 +9,7 @@ from sqlalchemy.orm import Session
 
 from .models import Base, Resource, ResourceFlow, Recipe
 from .sfp import SatisfactoryPlus
+from ..helper import prompt
 
 __recipe_info_timeout = datetime.timedelta(days=30)
 
@@ -27,14 +30,17 @@ def main(result: bool, debug: bool, refetch: bool, search: str):
             if len(matching_resources) == 0:
                 print("Could not find existing resources matching the search string.. starting wiki search")
             else:
-                for idx in range(1, len(matching_resources) + 1):
+                options = {(idx + 1): str(matching_resources[idx].label) for idx in range(len(matching_resources))}
-                    print(f"{idx}: {matching_resources[idx - 1].label}")
+                options[0] = ""
-                user_choice = click.prompt(
+                selected_res = prompt(
-                    "Chose a resource to continue or 0 to continue with a wiki search", default=1
+                    options=options, text="Chose a resource to continue or 0 to continue with a wiki search", default=1
                 )
-                if user_choice != 0:
+                if selected_res is None:
-                    resource = matching_resources[user_choice - 1]
+                    return
+                if selected_res != 0:
+                    resource = matching_resources[selected_res - 1]
                     do_provider_search = False
+                    exists_in_db = True
 
             with SatisfactoryPlus(debug=debug) as data_provider:
                 if do_provider_search:
@@ -76,8 +82,8 @@ def main(result: bool, debug: bool, refetch: bool, search: str):
 
                 stmt = select(Recipe).join(Recipe.results).filter(ResourceFlow.resource_id == resource.id)
                 for recipe in session.scalars(stmt):
-                    print(recipe)
+                    print("Recipe:", recipe.describe())
-                    for flow in recipe.ingredients:
+
-                        print("ingredient:", flow.resource, flow)
+
-                    for flow in recipe.results:
+if __name__ == "__main__":
-                        print("result:    ", flow.resource, flow)
+    main()

factorygame/data/models.py

@@ -1,4 +1,5 @@
 import datetime
+import re
 from typing import Optional
 
 from sqlalchemy import String, Select, select, Table, Column, ForeignKey
@@ -59,6 +60,9 @@ results_table = Table(
     Column("resource_flow_id", ForeignKey("resource_flows.id"), primary_key=True),
 )
 
+amount_number = re.compile(r"^\d*(\.\d+)?")
+time_number = re.compile(r"^(\d+) seconds?")
+
 
 class ResourceFlow(Base):
     __tablename__ = "resource_flows"
@@ -71,6 +75,16 @@ class ResourceFlow(Base):
     amount: Mapped[str]
     time: Mapped[str]
 
+    def amount_per_minute(self) -> float:
+        per_crafting_step_str = amount_number.match(self.amount).group(0)
+        per_crafting_step = float(per_crafting_step_str)
+        crafting_time_seconds_str = time_number.match(self.time).group(1)
+        crafting_time_seconds = float(crafting_time_seconds_str)
+        return per_crafting_step * (60.0 / crafting_time_seconds)
+
+    def describe(self) -> str:
+        return f"{repr(str(self.resource.label))}x{self.amount}"
+
     def __repr__(self):
         return f"ResourceFlow(id={self.id}, resource_id={self.resource_id}, amount={self.amount}, time={self.time})"
 
@@ -86,5 +100,15 @@ class Recipe(Base):
     )
     results: Mapped[list["ResourceFlow"]] = relationship(secondary=results_table, back_populates="result_of")
 
+    def describe(self) -> str:
+        def list_flows(flows: list["ResourceFlow"]) -> str:
+            return ", ".join(map(ResourceFlow.describe, flows))
+
+        return (
+            f"in machine: {self.factory.label}, "
+            f"ingredients: {list_flows(self.ingredients)}, "
+            f"results: {list_flows(self.results)}"
+        )
+
     def __repr__(self):
         return f"Recipe(id={self.id}, factory={self.factory}, ingredients={self.ingredients}, results={self.results})"

factorygame/data/sfp.py

@@ -3,7 +3,6 @@ from datetime import datetime
 from typing import Optional
 from urllib.parse import urljoin
 
-import click
 from selenium.webdriver import Firefox
 from selenium.webdriver.common.by import By
 from selenium.webdriver.firefox.options import Options
@@ -11,6 +10,7 @@ from sqlalchemy.orm import Session
 
 from .models import Resource, ResourceFlow, Factory, Recipe
 from .provider import RecipeProvider
+from ..helper import prompt
 
 
 class SatisfactoryPlus(RecipeProvider, AbstractContextManager):
@@ -59,20 +59,16 @@ class SatisfactoryPlus(RecipeProvider, AbstractContextManager):
 
         elif len(choices) > 1:
             default_choice = 1
-            choice_names: list[str] = []
+            options: dict[int, str] = {}
-            for choice_idx in range(1, len(choices) + 1):
+            for idx in range(len(choices)):
-                recipe_choice = choices[choice_idx - 1]
+                recipe_choice = choices[idx]
                 name = recipe_choice.find_element(By.TAG_NAME, "img").get_attribute("alt")
-                choice_names.append(name)
+                options[idx + 1] = name
                 if name.casefold() == search.casefold():
-                    default_choice = choice_idx
+                    default_choice = idx + 1
-                print(f"{choice_idx}: {name}")
+            user_choice = prompt(options=options, text="Chose a recipe to continue…", default=default_choice)
-            user_choice = click.prompt("Chose a recipe to continue…", default=default_choice)
+            if user_choice is None:
-            if not user_choice:
+                return None
-                user_choice = default_choice
-            else:
-                user_choice = int(user_choice)
-
             link_html_elem = choices[user_choice - 1]
         else:
             link_html_elem = choices[0]

factorygame/data/vis.py

@@ -0,0 +1,254 @@
+#!/usr/bin/env python3
+
+import click
+from NodeGraphQt import BaseNode, NodeBaseWidget
+from NodeGraphQt import NodeGraph, Port
+from NodeGraphQt.constants import PortTypeEnum
+from PySide2.QtCore import Qt
+from PySide2.QtWidgets import QSlider
+from Qt import QtWidgets
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session
+
+from .models import Recipe, ResourceFlow, Resource
+from .sfp import SatisfactoryPlus
+from ..helper import prompt
+
+
+class NodeSlider(NodeBaseWidget):
+    def __init__(self, name, label="", parent=None):
+        super(NodeSlider, self).__init__(parent)
+        self.set_name(name)
+        self.pure_label = label or name
+        slider = QSlider(Qt.Horizontal)
+        slider.setMinimum(1)
+        slider.setMaximum(250)
+        slider.setValue(100)
+        slider.setSingleStep(1)
+        slider.setTickInterval(10)
+        slider.setTickPosition(QSlider.TicksBelow)
+        slider.valueChanged.connect(self.on_value_changed)
+        slider.valueChanged.connect(self._update_label)
+        self.set_custom_widget(slider)
+        self._update_label()
+
+    def get_value(self):
+        widget: QSlider = self.get_custom_widget()
+        return widget.value()
+
+    def set_value(self, value):
+        widget: QSlider = self.get_custom_widget()
+        widget.setValue(value)
+
+    def _update_label(self):
+        self.set_label(f"{self.pure_label} ({int(self.get_value())}%)")
+
+
+class GlobalInput(BaseNode):
+    __identifier__ = "factorygame"
+    NODE_NAME = "Global Input"
+
+    def __init__(self):
+        super().__init__()
+        self.add_output("Create Machine")
+
+
+class Machine(BaseNode):
+    __identifier__ = "factorygame"
+    NODE_NAME = "FactoryGame Machine"
+
+    def assign_recipe(self, recipe: Recipe):
+        for port_idx in range(len(self._inputs)):
+            self.delete_input(port_idx)
+
+        for port_idx in range(len(self._outputs)):
+            self.delete_output(port_idx)
+
+        self.set_property("name", recipe.factory.label, push_undo=False)
+
+        def in_amount_name(resource_label: str):
+            return f"In {resource_label} amount"
+
+        def out_amount_name(resource_label: str):
+            return f"Out {resource_label} amount"
+
+        input_resources: dict[str, float] = {}
+        for ingredient in recipe.ingredients:
+            resource_label = ingredient.resource.label
+            if resource_label in input_resources:
+                input_resources[resource_label] += ingredient.amount_per_minute()
+                self.get_widget(in_amount_name(resource_label)).set_value(
+                    f"{input_resources[resource_label]:.2f} / min"
+                )
+            else:
+                port = self.add_input(resource_label, color=(180, 80, 0))
+                port.add_accept_port_type(resource_label, PortTypeEnum.OUT.value, "factorygame.Machine")
+                port.add_accept_port_type("Create Machine", PortTypeEnum.OUT.value, "factorygame.GlobalInput")
+                self._add_text_label(in_amount_name(resource_label), ingredient, input_resources, resource_label)
+
+        output_resources: dict[str, float] = {}
+        for result in recipe.results:
+            resource_label = result.resource.label
+            if resource_label in output_resources:
+                output_resources[resource_label] += result.amount_per_minute()
+                self.get_widget(out_amount_name(resource_label)).set_value(
+                    f"{output_resources[resource_label]:.2f} / min"
+                )
+            else:
+                port = self.add_output(resource_label, color=(200, 20, 0))
+                port.add_accept_port_type(resource_label, PortTypeEnum.IN.value, "factorygame.Machine")
+                self._add_text_label(out_amount_name(resource_label), result, output_resources, resource_label)
+
+        performance_slider_name = "machine performance"
+
+        def set_performance(percentage):
+            factor = percentage / 100.0
+            for ingredient_label, amount in input_resources.items():
+                self.get_widget(in_amount_name(ingredient_label)).set_value(f"{amount * factor:.2f} / min")
+            for result_label, amount in output_resources.items():
+                self.get_widget(out_amount_name(result_label)).set_value(f"{amount * factor:.2f} / min")
+            self.get_widget(performance_slider_name).get_value()
+
+        slider = NodeSlider(name=performance_slider_name, label="Overclocking Performance", parent=self.view)
+        self.add_custom_widget(slider)
+        slider.value_changed.connect(lambda name, value: set_performance(max(1, min(250, value))))
+
+    def _add_text_label(self, name, flow, resource_amounts, resource_label):
+        resource_amounts[resource_label] = flow.amount_per_minute()
+        self.add_text_input(name=name, label=name, text=f"{resource_amounts[resource_label]} / min")
+        widget = self.get_widget(name)
+        widget.get_custom_widget().setReadOnly(True)
+        widget.widget().setMaximumWidth(220)
+
+
+def on_port_connected(input_port: Port, output_port: Port):
+    global debug
+    if debug:
+        print(f"Port {output_port} connected to {input_port}")
+    if isinstance(output_port.node(), GlobalInput) and output_port.name() == "Create Machine":
+        output_port.clear_connections(push_undo=False, emit_signal=False)
+        with Session(engine) as session, SatisfactoryPlus() as data_provider:
+            do_provider_search = False
+            resource_label = input_port.name()
+            matching_resources = session.scalars(Resource.by_label(resource_label)).all()
+            if len(matching_resources) == 0:
+                print("Could not find existing resources matching the search string.. starting wiki search")
+            else:
+                options = {(idx + 1): str(matching_resources[idx].label) for idx in range(len(matching_resources))}
+                options[0] = ""
+                selected_res = prompt(
+                    options=options,
+                    text="Chose a resource to continue or 0 to continue with a wiki search",
+                    default=1,
+                )
+                if selected_res is None:
+                    return
+                if selected_res != 0:
+                    resource = matching_resources[selected_res - 1]
+                    do_provider_search = False
+
+            if do_provider_search:
+                ret = data_provider.search_for_resource(session=session, search=resource_label)
+                if ret is None:
+                    print("Resource not found")
+                    return
+
+                resource, exists_in_db = ret
+                if not exists_in_db:
+                    print("Resource not yet fetched, run fetch first")
+                    return
+
+            stmt = select(Recipe).join(Recipe.results).filter(ResourceFlow.resource_id == resource.id)
+            recipes = session.scalars(stmt).all()
+            if len(recipes) == 0:
+                print("No recipes found for resource")
+                return
+            elif len(recipes) > 1:
+                options = {(idx + 1): recipes[idx].describe() for idx in range(len(recipes))}
+                user_choice = prompt(options=options, text="Select recipe", default=1)
+                if user_choice is None:
+                    return
+                recipe = recipes[user_choice - 1]
+            else:
+                recipe = recipes[0]
+
+        recipe_machine = graph.create_node("factorygame.Machine", push_undo=True)
+        recipe_machine.assign_recipe(recipe)
+        recipe_machine.update()
+        recipe_machine.set_x_pos(input_port.node().x_pos() - recipe_machine.view.width - 200)
+        recipe_machine.get_output(input_port.name()).connect_to(input_port)
+        if recipe_machine.x_pos() - (global_input.x_pos() - global_input.view.width) < 200:
+            global_input.set_x_pos(recipe_machine.x_pos() - global_input.view.width - 200)
+
+
+@click.command
+@click.option("--debug", is_flag=True)
+@click.argument("search")
+def main(debug: bool, search: str):
+    global engine
+    globals()["debug"] = debug
+    engine = create_engine("sqlite:///file.db", echo=debug)
+    with Session(engine) as session, SatisfactoryPlus(debug=debug) as data_provider:
+        do_provider_search = False
+        matching_resources = session.scalars(Resource.by_label(search)).all()
+        if len(matching_resources) == 0:
+            print("Could not find existing resources matching the search string.. starting wiki search")
+        else:
+            options = {(idx + 1): str(matching_resources[idx].label) for idx in range(len(matching_resources))}
+            options[0] = ""
+            selected_res = prompt(
+                options=options, text="Chose a resource to continue or 0 to continue with a wiki search", default=1
+            )
+            if selected_res is None:
+                return
+            if selected_res != 0:
+                resource = matching_resources[selected_res - 1]
+                do_provider_search = False
+
+        if do_provider_search:
+            ret = data_provider.search_for_resource(session=session, search=search)
+            if ret is None:
+                print("Resource not found")
+                return
+
+            resource, exists_in_db = ret
+            if not exists_in_db:
+                print("Resource not yet fetched, run fetch first")
+                return
+
+        stmt = select(Recipe).join(Recipe.results).filter(ResourceFlow.resource_id == resource.id)
+        recipes = session.scalars(stmt).all()
+        if len(recipes) == 0:
+            print("No recipes found for resource")
+            return
+        elif len(recipes) > 1:
+            options = {(idx + 1): recipes[idx].describe() for idx in range(len(recipes))}
+            user_choice = prompt(options=options, text="Select recipe", default=1)
+            if user_choice is None:
+                return
+            recipe = recipes[user_choice - 1]
+        else:
+            recipe = recipes[0]
+
+    app = QtWidgets.QApplication([])
+    global graph
+    graph = NodeGraph()
+    graph.widget.resize(1280, 720)
+    graph.register_node(Machine)
+    graph.register_node(GlobalInput)
+    graph_widget = graph.widget
+    graph_widget.show()
+    global global_input
+    global_input = graph.create_node("factorygame.GlobalInput", push_undo=False)
+    recipe_machine = graph.create_node("factorygame.Machine", push_undo=False)
+    recipe_machine.assign_recipe(recipe)
+    graph.auto_layout_nodes([global_input, recipe_machine])
+    global_input.set_y_pos(recipe_machine.y_pos())
+    global_input.set_x_pos(global_input.x_pos() - global_input.model.width - 200)
+    graph.center_on([global_input, recipe_machine])
+    graph.port_connected.connect(on_port_connected)
+    app.exec_()
+
+
+if __name__ == "__main__":
+    main()

factorygame/helper.py

@@ -0,0 +1,13 @@
+import click
+
+
+def prompt(options: dict[int, str], **kwargs) -> int | None:
+    for idx, label in options.items():
+        if label:
+            click.echo(f"{idx}: {label}")
+    ret = click.prompt(**kwargs)
+    ret = int(ret)
+    if ret not in options:
+        click.echo("Invalid choice.")
+        return None
+    return ret