Auto-update: Sun Jun 30 00:13:43 PDT 2024
This commit is contained in:
parent
e4d6ec6412
commit
dadd5421ea
3 changed files with 109 additions and 78 deletions
|
@ -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}")
|
||||
|
||||
|
|
28
sijapi/routers/signal.py
Normal file
28
sijapi/routers/signal.py
Normal file
|
@ -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)
|
|
@ -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}")
|
||||
|
|
Loading…
Reference in a new issue