diff --git a/sijapi/routers/note.py b/sijapi/routers/note.py index 1498f22..c600286 100644 --- a/sijapi/routers/note.py +++ b/sijapi/routers/note.py @@ -243,18 +243,95 @@ async def update_frontmatter(date_time: dt_datetime, key: str, value: str): return {"message": "Frontmatter updated successfully."} @note.post("/note/banner") -async def banner_endpoint(dt: str, location: str = None, mood: str = None, other_context: str = None): +async def banner_endpoint(dt: str, location: str = None, forecast: str = None, mood: str = None, other_context: str = None): ''' Endpoint (POST) that generates a new banner image for the Obsidian daily note for a specified date, taking into account optional additional information, then updates the frontmatter if necessary. ''' L.DEBUG(f"banner_endpoint requested with date: {dt} ({type(dt)})") date_time = await loc.dt(dt) L.DEBUG(f"date_time after localization: {date_time} ({type(date_time)})") + context = await generate_context(dt, location, forecast, mood, other_context) jpg_path = await generate_banner(date_time, location, mood=mood, other_context=other_context) return jpg_path +async def generate_banner(dt, location: Location = None, forecast: str = None, mood: str = None, other_context: str = None): + # L.DEBUG(f"Location: {location}, forecast: {forecast}, mood: {mood}, other_context: {other_context}") + date_time = await loc.dt(dt) + L.DEBUG(f"generate_banner called with date_time: {date_time}") + destination_path, local_path = assemble_journal_path(date_time, filename="Banner", extension=".jpg", no_timestamp = True) + L.DEBUG(f"destination path generated: {destination_path}") + if not location or not isinstance(location, Location): + locations = await loc.fetch_locations(date_time) + if locations: + location = locations[0] + if not forecast: + forecast = await update_dn_weather(date_time, False, location.latitude, location.longitude) + + prompt = await generate_context(date_time, location, forecast, mood, other_context) + L.DEBUG(f"Prompt: {prompt}") + final_path = await sd.workflow(prompt, scene=OBSIDIAN_BANNER_SCENE, destination_path=destination_path) + if not str(local_path) in str(final_path): + L.INFO(f"Apparent mismatch between local path, {local_path}, and final_path, {final_path}") + jpg_embed = f"\"![[{local_path}]]\"" + await update_frontmatter(date_time, "banner", jpg_embed) + return local_path +async def generate_context(date_time, location: Location, forecast: str, mood: str, other_context: str): + display_name = "Location: " + if location and isinstance(location, Location): + lat, lon = location.latitude, location.longitude + override_location = GEO.find_override_location(lat, lon) + display_name += f"{override_location}, " if override_location else "" + if location.display_name: + display_name += f"{location.display_name}" + + else: + display_name += f"{location.road}, " if location.road else "" + display_name += f"the {location.neighbourhood} neighbourhood of " if location.neighbourhood else "" + display_name += f"the {location.suburb} suburb of " if location.suburb else "" + display_name += f"the {location.quarter} quarter, " if location.quarter else "" + display_name += f"{location.city}, " if location.city else "" + display_name += f"{location.state} " if location.state else "" + display_name += f"{location.country} " if location.country else "" + + if display_name == "Location: ": + geocoded_location = await GEO.code((lat, lon)) + if geocoded_location.display_name or geocoded_location.city or geocoded_location.country: + return await generate_context(date_time, geocoded_location, forecast, mood, other_context) + else: + L.WARN(f"Failed to get a useable location for purposes of generating a banner, but we'll generate one anyway.") + elif location and isinstance(location, str): + display_name = f"Location: {location}\n" + else: + display_name = "" + + if not forecast: + forecast = "The weather forecast is: " + await update_dn_weather(date_time) + + sentiment = await sentiment_analysis(date_time) + mood = sentiment if not mood else mood + mood = f"Mood: {mood}" if mood else "" + if mood and sentiment: mood = f"Mood: {mood}, {sentiment}" + elif mood and not sentiment: mood = f"Mood: {mood}" + elif sentiment and not mood: mood = f"Mood: {sentiment}" + else: mood = "" + + events = await cal.get_events(date_time, date_time) + formatted_events = [] + for event in events: + event_str = event.get('name') + if event.get('location'): + event_str += f" at {event.get('location')}" + formatted_events.append(event_str) + + additional_info = ', '.join(formatted_events) if formatted_events else '' + + other_context = f"{other_context}, {additional_info}" if other_context else additional_info + other_context = f"Additional information: {other_context}" if other_context else "" + + prompt = "Generate an aesthetically appealing banner image for a daily note that helps to visualize the following scene information: " + prompt += "\n".join([display_name, forecast, mood, other_context]) @@ -279,80 +356,6 @@ async def sentiment_analysis(date_time: dt_datetime): else: return "" -async def generate_banner(dt, location: Location = None, forecast: str = None, mood: str = None, other_context: str = None): - # L.DEBUG(f"Location: {location}, forecast: {forecast}, mood: {mood}, other_context: {other_context}") - date_time = await loc.dt(dt) - L.DEBUG(f"generate_banner called with date_time: {date_time}") - destination_path, local_path = assemble_journal_path(date_time, filename="Banner", extension=".jpg", no_timestamp = True) - L.DEBUG(f"destination path generated: {destination_path}") - - if not location: - locations = await loc.fetch_locations(date_time) - if locations: - location = locations[0] - - display_name = "Location: " - if location: - lat, lon = location.latitude, location.longitude - override_location = GEO.find_override_location(lat, lon) - display_name += f"{override_location}, " if override_location else "" - if location.display_name: - display_name += f"{location.display_name}" - - else: - display_name += f"{location.road}, " if location.road else "" - display_name += f"the {location.neighbourhood} neighbourhood of " if location.neighbourhood else "" - display_name += f"the {location.suburb} suburb of " if location.suburb else "" - display_name += f"the {location.quarter} quarter, " if location.quarter else "" - display_name += f"{location.city}, " if location.city else "" - display_name += f"{location.state} " if location.state else "" - display_name += f"{location.country} " if location.country else "" - - if display_name == "Location: ": - geocoded_location = await GEO.code((lat, lon)) - if geocoded_location.display_name or geocoded_location.city or geocoded_location.country: - return await generate_banner(dt, geocoded_location, forecast, mood, other_context) - else: - L.WARN(f"Failed to get a useable location for purposes of generating a banner, but we'll generate one anyway.") - - if not forecast: - forecast = "The weather forecast is: " + await update_dn_weather(date_time) - - sentiment = await sentiment_analysis(date_time) - mood = sentiment if not mood else mood - mood = f"Mood: {mood}" if mood else "" - - if mood and sentiment: mood = f"Mood: {mood}, {sentiment}" - elif mood and not sentiment: mood = f"Mood: {mood}" - elif sentiment and not mood: mood = f"Mood: {sentiment}" - else: mood = "" - - events = await cal.get_events(date_time, date_time) - formatted_events = [] - for event in events: - event_str = event.get('name') - if event.get('location'): - event_str += f" at {event.get('location')}" - formatted_events.append(event_str) - - additional_info = ', '.join(formatted_events) if formatted_events else '' - - other_context = f"{other_context}, {additional_info}" if other_context else additional_info - other_context = f"Additional information: {other_context}" if other_context else "" - - prompt = "Generate an aesthetically appealing banner image for a daily note that helps to visualize the following scene information: " - prompt += "\n".join([display_name, forecast, mood, other_context]) - L.DEBUG(f"Prompt: {prompt}") - # sd.workflow(prompt: str, scene: str = None, size: str = None, style: str = "photorealistic", earlyurl: bool = False, destination_path: str = None): - final_path = await sd.workflow(prompt, scene=OBSIDIAN_BANNER_SCENE, size="1080x512", style="romantic", destination_path=destination_path) - if not str(local_path) in str(final_path): - L.INFO(f"Apparent mismatch between local path, {local_path}, and final_path, {final_path}") - - jpg_embed = f"\"![[{local_path}]]\"" - await update_frontmatter(date_time, "banner", jpg_embed) - - return local_path - @note.get("/note/weather", response_class=JSONResponse) async def note_weather_get( @@ -430,7 +433,7 @@ async def update_dn_weather(date_time: dt_datetime, force_refresh: bool = False, DailyWeather = day.get('DailyWeather') HourlyWeather = day.get('HourlyWeather') if DailyWeather: - L.DEBUG(f"Day: {DailyWeather}") + # L.DEBUG(f"Day: {DailyWeather}") icon = DailyWeather.get('icon') L.DEBUG(f"Icon: {icon}") diff --git a/sijapi/routers/signal.py b/sijapi/routers/signal.py new file mode 100644 index 0000000..516e053 --- /dev/null +++ b/sijapi/routers/signal.py @@ -0,0 +1,28 @@ +""" +Signal Bot example, repeats received messages. +""" +import os + +from semaphore import Bot, ChatContext + + +async def echo(ctx: ChatContext) -> None: + if not ctx.message.empty(): + await ctx.message.typing_started() + await ctx.message.reply(ctx.message.get_body()) + await ctx.message.typing_stopped() + + +async def main() -> None: + """Start the bot.""" + # Connect the bot to number. + async with Bot(os.environ["SIGNAL_PHONE_NUMBER"]) as bot: + bot.register_handler("", echo) + + # Run the bot until you press Ctrl-C. + await bot.start() + + +if __name__ == '__main__': + import anyio + anyio.run(main) \ No newline at end of file diff --git a/sijapi/routers/weather.py b/sijapi/routers/weather.py index 324d294..c8093fd 100644 --- a/sijapi/routers/weather.py +++ b/sijapi/routers/weather.py @@ -287,12 +287,12 @@ async def get_weather_from_db(date_time: dt_datetime, latitude: float, longitude # hour_data['datetime'] = await loc.dt(hour_data.get('datetime'), tz) hourly_weather_data.append(hour_data) - L.DEBUG(f"Hourly weather data after tz corrections: {hourly_weather_data}") + # L.DEBUG(f"Hourly weather data after tz corrections: {hourly_weather_data}") day = { 'DailyWeather': daily_weather_data, 'HourlyWeather': hourly_weather_data, } - L.DEBUG(f"day: {day}") + # L.DEBUG(f"day: {day}") return day except Exception as e: L.ERR(f"Unexpected error occurred: {e}")