refactored model into future streams of streams idk izzy told me to do it

Cargo.lock

@@ -66,6 +66,22 @@ dependencies = [
  "futures-lite",
 ]
 
+[[package]]
+name = "async-global-executor"
+version = "2.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9586ec52317f36de58453159d48351bc244bc24ced3effc1fce22f3d48664af6"
+dependencies = [
+ "async-channel",
+ "async-executor",
+ "async-io",
+ "async-mutex",
+ "blocking",
+ "futures-lite",
+ "num_cpus",
+ "once_cell",
+]
+
 [[package]]
 name = "async-io"
 version = "1.6.0"
@@ -94,6 +110,15 @@ dependencies = [
  "event-listener",
 ]
 
+[[package]]
+name = "async-mutex"
+version = "1.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "479db852db25d9dbf6204e6cb6253698f175c15726470f78af0d918e99d6156e"
+dependencies = [
+ "event-listener",
+]
+
 [[package]]
 name = "async-net"
 version = "1.6.1"
@@ -122,6 +147,34 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "async-std"
+version = "1.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d9f06685bad74e0570f5213741bea82158279a4103d988e57bfada11ad230341"
+dependencies = [
+ "async-channel",
+ "async-global-executor",
+ "async-io",
+ "async-lock",
+ "async-process",
+ "crossbeam-utils 0.8.5",
+ "futures-channel",
+ "futures-core",
+ "futures-io",
+ "futures-lite",
+ "gloo-timers",
+ "kv-log-macro",
+ "log 0.4.14",
+ "memchr",
+ "num_cpus",
+ "once_cell",
+ "pin-project-lite",
+ "pin-utils",
+ "slab",
+ "wasm-bindgen-futures",
+]
+
 [[package]]
 name = "async-task"
 version = "4.0.3"
@@ -440,7 +493,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9f02af974daeee82218205558e51ec8768b48cf524bd01d550abe5573a608285"
 dependencies = [
  "crossbeam-epoch",
- "crossbeam-utils",
+ "crossbeam-utils 0.7.2",
  "maybe-uninit",
 ]
 
@@ -452,7 +505,7 @@ checksum = "058ed274caafc1f60c4997b5fc07bf7dc7cca454af7c6e81edffe5f33f70dace"
 dependencies = [
  "autocfg 1.0.1",
  "cfg-if 0.1.10",
- "crossbeam-utils",
+ "crossbeam-utils 0.7.2",
  "lazy_static",
  "maybe-uninit",
  "memoffset",
@@ -466,7 +519,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "774ba60a54c213d409d5353bda12d49cd68d14e45036a285234c8d6f91f92570"
 dependencies = [
  "cfg-if 0.1.10",
- "crossbeam-utils",
+ "crossbeam-utils 0.7.2",
  "maybe-uninit",
 ]
 
@@ -481,6 +534,26 @@ dependencies = [
  "lazy_static",
 ]
 
+[[package]]
+name = "crossbeam-utils"
+version = "0.8.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d82cfc11ce7f2c3faef78d8a684447b40d503d9681acebed6cb728d45940c4db"
+dependencies = [
+ "cfg-if 1.0.0",
+ "lazy_static",
+]
+
+[[package]]
+name = "ctor"
+version = "0.1.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5e98e2ad1a782e33928b96fc3948e7c355e5af34ba4de7670fe8bac2a3b2006d"
+dependencies = [
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "data-encoding"
 version = "2.3.2"
@@ -853,6 +926,19 @@ version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9b919933a397b79c37e33b77bb2aa3dc8eb6e165ad809e58ff75bc7db2e34574"
 
+[[package]]
+name = "gloo-timers"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47204a46aaff920a1ea58b11d03dec6f704287d27561724a4631e450654a891f"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "js-sys",
+ "wasm-bindgen",
+ "web-sys",
+]
+
 [[package]]
 name = "groupable"
 version = "0.2.0"
@@ -1107,6 +1193,7 @@ checksum = "dd25036021b0de88a0aff6b850051563c6516d0bf53f8638938edbb9de732736"
 name = "izzilis"
 version = "0.1.0"
 dependencies = [
+ "async-std",
  "chrono",
  "elefren",
  "frankenstein",
@@ -1137,6 +1224,15 @@ dependencies = [
  "winapi-build",
 ]
 
+[[package]]
+name = "kv-log-macro"
+version = "1.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0de8b303297635ad57c9f5059fd9cee7a47f8e8daa09df0fcd07dd39fb22977f"
+dependencies = [
+ "log 0.4.14",
+]
+
 [[package]]
 name = "language-tags"
 version = "0.2.2"
@@ -1180,6 +1276,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "51b9bbe6c47d51fc3e1a9b945965946b4c44142ab8792c50835a980d362c2710"
 dependencies = [
  "cfg-if 1.0.0",
+ "value-bag",
 ]
 
 [[package]]
@@ -2375,7 +2472,7 @@ version = "0.1.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fb2d1b8f4548dbf5e1f7818512e9c406860678f29c300cdf0ebac72d1a3a1671"
 dependencies = [
- "crossbeam-utils",
+ "crossbeam-utils 0.7.2",
  "futures 0.1.31",
 ]
 
@@ -2396,7 +2493,7 @@ version = "0.1.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "09bc590ec4ba8ba87652da2068d150dcada2cfa2e07faae270a5e0409aa51351"
 dependencies = [
- "crossbeam-utils",
+ "crossbeam-utils 0.7.2",
  "futures 0.1.31",
  "lazy_static",
  "log 0.4.14",
@@ -2441,7 +2538,7 @@ checksum = "df720b6581784c118f0eb4310796b12b1d242a7eb95f716a8367855325c25f89"
 dependencies = [
  "crossbeam-deque",
  "crossbeam-queue",
- "crossbeam-utils",
+ "crossbeam-utils 0.7.2",
  "futures 0.1.31",
  "lazy_static",
  "log 0.4.14",
@@ -2456,7 +2553,7 @@ version = "0.2.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "93044f2d313c95ff1cb7809ce9a7a05735b012288a888b62d4434fd58c94f296"
 dependencies = [
- "crossbeam-utils",
+ "crossbeam-utils 0.7.2",
  "futures 0.1.31",
  "slab",
  "tokio-executor",
@@ -2659,6 +2756,16 @@ dependencies = [
  "rand 0.6.5",
 ]
 
+[[package]]
+name = "value-bag"
+version = "1.0.0-alpha.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dd320e1520f94261153e96f7534476ad869c14022aee1e59af7c778075d840ae"
+dependencies = [
+ "ctor",
+ "version_check 0.9.3",
+]
+
 [[package]]
 name = "vcpkg"
 version = "0.2.15"
@@ -2742,6 +2849,18 @@ dependencies = [
  "wasm-bindgen-shared",
 ]
 
+[[package]]
+name = "wasm-bindgen-futures"
+version = "0.4.24"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5fba7978c679d53ce2d0ac80c8c175840feb849a161664365d1287b41f2e67f1"
+dependencies = [
+ "cfg-if 1.0.0",
+ "js-sys",
+ "wasm-bindgen",
+ "web-sys",
+]
+
 [[package]]
 name = "wasm-bindgen-macro"
 version = "0.2.74"

Cargo.toml

@@ -7,6 +7,7 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+async-std = { version = "1.9.0", features = ["unstable"] }
 chrono = "0.4.19"
 elefren = { version = "0.22.0", features = ["toml"] }
 frankenstein = "0.4.0"

src/bot.rs

@@ -1,11 +1,14 @@
 use rand::Rng;
 use std::error::Error;
 
-use crate::{generator, model, publish, selection};
+use crate::{
+    model::{self, SampleModel},
+    publish, selection,
+};
 
-pub struct IzzilisBot<T: model::SampleModel, U: publish::Publisher, V: selection::Selector> {
+pub struct IzzilisBot<T: SampleModel, U: publish::Publisher, V: selection::Selector> {
-    generator: generator::Generator<T>,
+    model: T,
-    publisher: U, // One day I'll figure out how to make this a vector with differing Publisher types
+    publisher: U,
     selector: V,
     loaded_samples: Vec<String>,
 }
@@ -16,13 +19,9 @@ where
     U: publish::Publisher,
     V: selection::Selector,
 {
-    pub fn new(
+    pub fn new(model: T, publisher: U, selector: V) -> IzzilisBot<T, U, V> {
-        generator: generator::Generator<T>,
-        publisher: U,
-        selector: V,
-    ) -> IzzilisBot<T, U, V> {
         Self {
-            generator,
+            model,
             publisher,
             loaded_samples: Vec::new(),
             selector: selector,
@@ -30,18 +29,19 @@ where
     }
 
     pub fn generate_samples(&mut self) -> Result<(), Box<dyn Error>> {
-        let lines = self.generator.generate_sample_lines()?;
+        // let lines = self.model.generate_sample_lines()?;
-        for line in lines {
+        // for line in lines {
-            self.selector.send_for_review(line)?;
+        //     self.selector.send_for_review(line)?;
-        }
+        // }
 
-        self.loaded_samples = self.selector.collect_selected_samples(); // wtf happens to the original self.loaded_samples???????
+        // self.loaded_samples = self.selector.collect_selected_samples(); // wtf happens to the original self.loaded_samples???????
 
-        Ok(())
+        // Ok(())
+        todo!()
     }
 
     pub fn publish(&mut self) -> Result<(), U::Error> {
-        if self.loaded_samples.len() == 0 {
+        if self.loaded_samples.len() < 5 {
             // Refresh samples. Either none have been generated so far,
             // or generated ones are stale.
             //

src/generator.rs

@@ -1,30 +0,0 @@
-use std::io;
-
-use crate::model;
-const SAMPLE_SPLIT_WORD: &str = "<|endoftext|>";
-const SAMPLE_SAMPLE_LINE: &str =
-    "======================================== SAMPLE 1 ========================================";
-pub struct Generator<T: model::SampleModel> {
-    model: T,
-}
-
-// Why did this fucking shit take so long to sort out??
-impl<T> Generator<T>
-where
-    T: model::SampleModel,
-{
-    pub fn new(model: T) -> Generator<T> {
-        Self { model }
-    }
-
-    pub fn generate_sample_lines(&self) -> Result<Vec<String>, io::Error> {
-        Ok(self
-            .model
-            .get_sample()?
-            .replace(SAMPLE_SAMPLE_LINE, "")
-            .split(SAMPLE_SPLIT_WORD)
-            .into_iter()
-            .map(|elem| elem.trim().to_owned())
-            .collect::<Vec<_>>())
-    }
-}

src/main.rs

@@ -3,15 +3,14 @@ use std::{error::Error, process, time::Duration};
 use chrono::Local;
 use rand::Rng;
 
-use crate::{
+use crate::{bot::IzzilisBot, publish::FediversePublisher, selection::ConsoleSelector};
-    bot::IzzilisBot, generator::Generator, publish::FediversePublisher, selection::ConsoleSelector,
-};
 
+use futures::StreamExt;
 use futures_timer::Delay;
+use model::SampleModelExt;
 
 mod bot;
 mod config;
-mod generator;
 mod model;
 mod publish;
 mod selection;
@@ -35,7 +34,7 @@ fn main() -> Result<(), Box<dyn Error>> {
             }
         };
 
-        let gpt_model = model::GPTSampleModel::new(
+        let mut gpt_model = model::GPTSampleModel::new(
             cfg.python_path(),
             cfg.gpt_code_path(),
             vec![
@@ -49,25 +48,32 @@ fn main() -> Result<(), Box<dyn Error>> {
                 "--nsamples".to_string(),
                 "1".to_string(),
             ],
-        );
+        )
-        let publisher = FediversePublisher::new(cfg.fediverse_base_url())?;
+        .into_stream()
-        // let publisher = ConsolePublisher::new();
+        .take(60);
-        let gen = Generator::new(gpt_model);
-        let console_selector = ConsoleSelector::new();
-        let mut bot = IzzilisBot::new(gen, publisher, console_selector);
-        bot.generate_samples();
 
-        let cfg_interval = cfg.interval_seconds();
+        while let Some(Ok(sample)) = gpt_model.next().await {
-        loop {
+            println!("{}", sample);
-            let wait_seconds = rand::thread_rng().gen_range(cfg_interval.min()..cfg_interval.max());
-            let wait_time = Duration::from_secs(wait_seconds);
-            let now = Local::now();
-            println!("[{}] Next post is in [{}] seconds", now, wait_seconds);
-            Delay::new(wait_time).await;
-            match bot.publish() {
-                Err(err) => println!("Got error from publish: [{}]; continuing", err),
-                Ok(()) => println!("publish() call successful"),
-            }
         }
+
+        return Ok(());
+        // let publisher = FediversePublisher::new(cfg.fediverse_base_url())?;
+        // // let publisher = ConsolePublisher::new();
+        // let console_selector = ConsoleSelector::new();
+        // let mut bot = IzzilisBot::new(gen, publisher, console_selector);
+        // bot.generate_samples();
+
+        // let cfg_interval = cfg.interval_seconds();
+        // loop {
+        //     let wait_seconds = rand::thread_rng().gen_range(cfg_interval.min()..cfg_interval.max());
+        //     let wait_time = Duration::from_secs(wait_seconds);
+        //     let now = Local::now();
+        //     println!("[{}] Next post is in [{}] seconds", now, wait_seconds);
+        //     Delay::new(wait_time).await;
+        //     match bot.publish() {
+        //         Err(err) => println!("Got error from publish: [{}]; continuing", err),
+        //         Ok(()) => println!("publish() call successful"),
+        //     }
+        // }
     })
 }

src/model.rs

@@ -1,7 +1,11 @@
-use std::{io, process::Command};
+use async_std::{io, process::Command};
+use futures::{future::BoxFuture, stream::BoxStream, Future, Stream, StreamExt, TryStreamExt};
 
 pub trait SampleModel {
-    fn get_sample(&self) -> Result<String, io::Error>;
+    type Error;
+    type Sample: Future<Output = Result<String, Self::Error>>;
+
+    fn get_sample(&self) -> Self::Sample;
 }
 
 pub struct GPTSampleModel {
@@ -11,13 +15,15 @@ pub struct GPTSampleModel {
 }
 
 impl SampleModel for GPTSampleModel {
-    fn get_sample(&self) -> Result<String, io::Error> {
+    type Error = io::Error;
-        let cmd_output = Command::new(&self.python_command)
+    type Sample = BoxFuture<'static, Result<String, Self::Error>>;
+
+    fn get_sample(&self) -> Self::Sample {
+        let cmd = Command::new(&self.python_command)
             .current_dir(&self.command_working_path)
             .args(&self.command_args)
-            .output()?;
+            .output();
-
+        Box::pin(async { Ok(String::from_utf8_lossy(&cmd.await?.stdout).to_string()) })
-        Ok(String::from_utf8_lossy(&cmd_output.stdout).to_string())
     }
 }
 
@@ -34,3 +40,40 @@ impl GPTSampleModel {
         }
     }
 }
+
+pub trait SampleModelExt: SampleModel {
+    type Stream: Stream<Item = Result<String, Self::Error>>;
+
+    fn into_stream(self) -> Self::Stream;
+}
+
+const SAMPLE_SPLIT_WORD: &str = "<|endoftext|>";
+const SAMPLE_SAMPLE_LINE: &str =
+    "======================================== SAMPLE 1 ========================================";
+
+impl<T: SampleModel + Send + Sync + 'static> SampleModelExt for T
+where
+    Self::Sample: Send,
+{
+    type Stream = BoxStream<'static, Result<String, Self::Error>>;
+
+    fn into_stream(self) -> Self::Stream {
+        Box::pin(
+            futures::stream::try_unfold(self, |this| async {
+                Ok(Some((this.get_sample().await?, this)))
+            })
+            .map_ok(|samples| {
+                futures::stream::iter(
+                    samples
+                        .replace(SAMPLE_SAMPLE_LINE, "")
+                        .split(SAMPLE_SPLIT_WORD)
+                        .map(|elem| elem.to_owned())
+                        .collect::<Vec<String>>()
+                        .into_iter()
+                        .map(|elem| Ok(elem.trim().to_owned())),
+                )
+            })
+            .try_flatten(),
+        )
+    }
+}